调用特性（二）之泛化引用

本文基于 Dubbo 2.6.1 版本，望知悉。

1. 概述

本文分享泛化引用。我们来看下 《用户指南 —— 泛化引用》 的定义：

泛化接口调用方式主要用于客户端没有 API 接口及模型类元的情况，参数及返回值中的所有 POJO 均用 Map 表示，通常用于框架集成，比如：实现一个通用的服务测试框架，可通过 GenericService 调用所有服务实现。

请注意，消费消费者没有 API 接口 及 模型类元。那就是说，Dubbo 在泛化引用中，需要做两件事情：

• 没有 API 接口 com.alibaba.dubbo.rpc.service.GenericService，所以提供一个泛化服务接口，目前是，代码如下：

public interface GenericService {

    /**
     * Generic invocation
     *
     * 泛化调用
     *
     * @param method         Method name, e.g. findPerson. If there are overridden methods, parameter info is
     *                       required, e.g. findPerson(java.lang.String)
     *                       方法名
     * @param parameterTypes Parameter types
     *                       参数类型数组
     * @param args           Arguments
     *                       参数数组
     * @return invocation return value 调用结果
     * @throws Throwable potential exception thrown from the invocation
     */
    Object $invoke(String method, String[] parameterTypes, Object[] args) throws GenericException;

}

• 一个泛化引用，只对应一个服务实现。
• 通过 $invoke(method, parameterTypes, args) 方法，可以实现服务的泛化调用。

• 具体的使用方式，我们在 「2. 示例」 中看。

• 没有 模型类元转换处理，所以方法参数和方法返回若是 POJO ( 例如 User 和 Order 等 ) ，需要：
  • 服务消费者，将 POJO 转成 Map ，然后再调用服务提供者。
  • 服务提供者，接收到 Map ，转换成 POJO ，再调用 Service 方法。若返回值有 POJO ，则转换成 Map 再返回。
  • 此处的 Map 只是举例子，实际在下文中，我们会看到还有两种转换方式。

整体流程如下：

2. 示例

服务提供者

在 dubbo-generic-reference-demo-provider ，我们提供了例子。普通的服务提供者，不需要做任何处理，胖友自己查看。

服务消费者

在 dubbo-generic-reference-demo-consumer ，我们提供了例子。我们挑重点的地方说。

① 在 Spring 配置申明 generic=”true”：

<dubbo:reference id="demoService"  interface="com.alibaba.dubbo.demo.DemoService" generic="true" />

• interface 配置项，泛化引用的服务接口。通过该配置，可以从注册中心，获取到所有该服务的提供方的地址。
• generic 配置项，默认为 false，不使用配置项。目前有三种配置项的值，开启泛化引用的功能：
  • generic=true，使用 com.alibaba.dubbo.common.utils.PojoUtils，实现 POJO <=> Map 的互转。
  • generic=nativejava，使用 com.alibaba.dubbo.common.serialize.support.nativejava.NativeJavaSerialization，实现 POJO <=> byte[] 的互转。
  • generic=bean，使用 com.alibaba.dubbo.common.beanutil.JavaBeanSerializeUtil，实现 POJO <=> JavaBeanDescriptor 的互转。
  • 总的来说，三种方式的差异，在于使用互转( 序列化和反序列化 )的方式不同。未来如果我们有需要，完成实现 generic=json，使用 FastJSON 来序列化和反序列化。

② 在 Java 代码获取 barService 并开始泛化调用：

ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext(new String[]{"META-INF/spring/dubbo-demo-consumer.xml"});

GenericService genericService = (GenericService) context.getBean("demoService");
Object result = genericService.$invoke("say01", new String[]{"java.lang.String"}, new Object[]{"123"});
System.out.println("result: " + result);

那么问题就来了，为什么可以使用 GenericService 转换？答案在 ReferenceConfig#init() 方法中，代码如下：

// ReferenceConfig.java
if (ProtocolUtils.isGeneric(getGeneric())) {
    interfaceClass = GenericService.class;
}

// ProtocolUtils.java
public static boolean isGeneric(String generic) {
    return generic != null
            && !"".equals(generic)
            && (Constants.GENERIC_SERIALIZATION_DEFAULT.equalsIgnoreCase(generic)  /* Normal generalization cal */
            || Constants.GENERIC_SERIALIZATION_NATIVE_JAVA.equalsIgnoreCase(generic) /* Streaming generalization call supporting jdk serialization */
            || Constants.GENERIC_SERIALIZATION_BEAN.equalsIgnoreCase(generic));
}

2.1 有关泛化类型的进一步解释

本小节为引用 《Dubbo 用户指南 —— 泛化引用》 。

假设存在 POJO 如：

package com.xxx;

public class PersonImpl implements Person {
    private String name;
    private String password;

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    public String getPassword() {
        return password;
    }

    public void setPassword(String password) {
        this.password = password;
    }
}

则 POJO 数据：

Person person = new PersonImpl();

person.setName("xxx");

person.setPassword("yyy");

【服务消费者】可用下面 Map 表示：

Map<String, Object> map = new HashMap<String, Object>();

// 注意：如果参数类型是接口，或者List等丢失泛型，可通过class属性指定类型。
map.put("class", "com.xxx.PersonImpl");

map.put("name", "xxx");
map.put("password", "yyy");

• Map 中的 class 属性，在 PojoUtils 中，会根据该属性，将 Map 转换成 POJO 对象。

3. 服务消费者 GenericImplFilter

com.alibaba.dubbo.rpc.filter.GenericImplFilter ，实现 Filter 接口，服务消费者的泛化调用过滤器。代码如下：

@Activate(group = Constants.CONSUMER, value = Constants.GENERIC_KEY, order = 20000)
public class GenericImplFilter implements Filter {

    // ... 省略无关属性

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        // 获得 `generic` 配置项
        String generic = invoker.getUrl().getParameter(Constants.GENERIC_KEY);

        // 省略代码...泛化实现的调用

        // 泛化引用的调用
        if (invocation.getMethodName().equals(Constants.$INVOKE) // 方法名为 `$invoke`
                && invocation.getArguments() != null
                && invocation.getArguments().length == 3
                && ProtocolUtils.isGeneric(generic)) {
            Object[] args = (Object[]) invocation.getArguments()[2];
            // `nativejava` ，校验方法参数都为 byte[]
            if (ProtocolUtils.isJavaGenericSerialization(generic)) {
                for (Object arg : args) {
                    if (!(byte[].class == arg.getClass())) {
                        error(byte[].class.getName(), arg.getClass().getName());
                    }
                }
            // `bean` ，校验方法参数为 JavaBeanDescriptor
            } else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
                for (Object arg : args) {
                    if (!(arg instanceof JavaBeanDescriptor)) {
                        error(JavaBeanDescriptor.class.getName(), arg.getClass().getName());
                    }
                }
            }

            // 通过隐式参数，传递 `generic` 配置项
            ((RpcInvocation) invocation).setAttachment(Constants.GENERIC_KEY, generic);
        }
        return invoker.invoke(invocation);
    }

    // 省略 `#error(...)` 方法

}

• 使用 Dubbo SPI Adaptive 机制，自动加载服务消费者，仅限，并且有 generic 配置项。
  • 此处笔者就产生了一个疑问，从注册中心，获得到的服务 URL ，并没有设置 generic 的配置项，那岂不是在 ProtocolFilterWrapper 中，使用 Dubbo SPI Adaptive 加载不到 GenericImplFilter 这个过滤器？
  • 于是，笔者就开始慢慢调试，直到发现 覆盖 RegistryDirectory#mergeUrl(providerUrl) 方法，它会将服务消费者的配置( URL )项，到服务提供者的 URL 。
  • 又因为泛化引用时，我们会在服务消费者的配置 generic = true，那么服务提供者 URL 自然就有了该配置项，所以便有了 GenericImplFilter 过滤器。
• 第 9 行：获得 generic 配置项。
• 第 11 行：省略用于调用泛化实现服务的代码，对应文档为 《Dubbo 用户指南 —— 泛化实现》，下一篇文章，我们详细分享。
• 第 13 至 37 行：泛化引用的调用，通过方法( 包括方法名、参数 ) + generic 配置项，进行判断。
  • 第 19 至 33 行：根据不同的 序列化 generic 配置项，校验方法参数是否已经正确。若不合法，调用 #error(expected, actual) 方法，抛出 RpcException 异常。代码如下：

private void error(String expected, String actual) throws RpcException {
    throw new RpcException(
            new StringBuilder(32)
                    .append("Generic serialization [")
                    .append(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
                    .append("] only support message type ")
                    .append(expected)
                    .append(" and your message type is ")
                    .append(actual).toString());
}

• 第 36 行：调用 RpcInvocation#setAttachment(key, value) 通过隐式参数，传递 generic 配置项。

• 第 38 行：调用 Invoker#invoke(invocation) 方法，继续过滤链的调用，最终 RPC 调用。

4. 服务提供者 GenericFilter

com.alibaba.dubbo.rpc.filter.GenericFilter ，实现 Filter 接口，服务消费者的泛化调用过滤器。代码如下：

@Activate(group = Constants.PROVIDER, order = -20000)
public class GenericFilter implements Filter {

    @Override
    public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
        // 泛化引用的调用
        if (inv.getMethodName().equals(Constants.$INVOKE)
                && inv.getArguments() != null
                && inv.getArguments().length == 3
                && !ProtocolUtils.isGeneric(invoker.getUrl().getParameter(Constants.GENERIC_KEY))) { // 非泛化实现的调用
            String name = ((String) inv.getArguments()[0]).trim();
            String[] types = (String[]) inv.getArguments()[1];
            Object[] args = (Object[]) inv.getArguments()[2];
            try {
                // 获得对应的方法 Method 对象
                Method method = ReflectUtils.findMethodByMethodSignature(invoker.getInterface(), name, types);
                // 获得方法参数类型和方法参数数组
                Class<?>[] params = method.getParameterTypes();
                if (args == null) {
                    args = new Object[params.length];
                }
                // 获得 `generic` 配置项
                String generic = inv.getAttachment(Constants.GENERIC_KEY);
                // 【第一步】`true` ，反序列化参数，仅有 Map => POJO
                if (StringUtils.isEmpty(generic) || ProtocolUtils.isDefaultGenericSerialization(generic)) {
                    args = PojoUtils.realize(args, params, method.getGenericParameterTypes());
                // 【第一步】`nativejava` ，反序列化参数，byte[] => 方法参数
                } else if (ProtocolUtils.isJavaGenericSerialization(generic)) {
                    for (int i = 0; i < args.length; i++) {
                        if (byte[].class == args[i].getClass()) {
                            try {
                                UnsafeByteArrayInputStream is = new UnsafeByteArrayInputStream((byte[]) args[i]);
                                args[i] = ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
                                        .deserialize(null, is).readObject();
                            } catch (Exception e) {
                                throw new RpcException("Deserialize argument [" + (i + 1) + "] failed.", e);
                            }
                        } else {
                            throw new RpcException(
                                    new StringBuilder(32).append("Generic serialization [")
                                            .append(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
                                            .append("] only support message type ")
                                            .append(byte[].class)
                                            .append(" and your message type is ")
                                            .append(args[i].getClass()).toString());
                        }
                    }
                // 【第一步】`bean` ，反序列化参数，JavaBeanDescriptor => 方法参数
                } else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
                    for (int i = 0; i < args.length; i++) {
                        if (args[i] instanceof JavaBeanDescriptor) {
                            args[i] = JavaBeanSerializeUtil.deserialize((JavaBeanDescriptor) args[i]);
                        } else {
                            throw new RpcException(
                                    new StringBuilder(32)
                                            .append("Generic serialization [")
                                            .append(Constants.GENERIC_SERIALIZATION_BEAN)
                                            .append("] only support message type ")
                                            .append(JavaBeanDescriptor.class.getName())
                                            .append(" and your message type is ")
                                            .append(args[i].getClass().getName()).toString());
                        }
                    }
                }
                // 【第二步】方法调用
                Result result = invoker.invoke(new RpcInvocation(method, args, inv.getAttachments()));
                // 【第三步】若是异常结果，并且非 GenericException 异常，则使用 GenericException 包装
                if (result.hasException()
                        && !(result.getException() instanceof GenericException)) {
                    return new RpcResult(new GenericException(result.getException()));
                }
                // 【第三步】`nativejava` ，序列化结果，结果 => byte[]
                if (ProtocolUtils.isJavaGenericSerialization(generic)) {
                    try {
                        UnsafeByteArrayOutputStream os = new UnsafeByteArrayOutputStream(512);
                        ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(Constants.GENERIC_SERIALIZATION_NATIVE_JAVA)
                                .serialize(null, os).writeObject(result.getValue());
                        return new RpcResult(os.toByteArray());
                    } catch (IOException e) {
                        throw new RpcException("Serialize result failed.", e);
                    }
                // 【第三步】`bean` ，序列化结果，结果 => JavaBeanDescriptor
                } else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
                    return new RpcResult(JavaBeanSerializeUtil.serialize(result.getValue(), JavaBeanAccessor.METHOD));
                // 【第三步】`true` ，序列化结果，仅有 POJO => Map
                } else {
                    return new RpcResult(PojoUtils.generalize(result.getValue()));
                }
            } catch (NoSuchMethodException e) {
                throw new RpcException(e.getMessage(), e);
            } catch (ClassNotFoundException e) {
                throw new RpcException(e.getMessage(), e);
            }
        }
        // 普通调用
        return invoker.invoke(inv);
    }

}

• 使用 Dubbo SPI Adaptive 机制，自动加载服务提供者，仅限。
• 第 96 行：若是普通调用( 非泛化引用的调用 )，调用 Invoker#invoke(invocation) 方法，继续过滤链的调用，最终调用 Service 服务。
• 第 6 至 95 行：若是泛化引用的调用，通过方法( 包括方法名、参数 )判断。注意【第 10 行】非泛化实现的调用。
• 第 16 行：调用 ReflectUtils#findMethodByMethodSignature(Class<?> clazz, String methodName, String[] parameterTypes) 方法，通过反射，获得对应的方法 Method 对象。具体的代码实现，胖友自己查看哈。
• 第 17 至 21 行：获得方法参数类型和方法参数数组。
• 第 23 行：获得 generic 配置项。
• ========== 【第一步：反序列化参数】 ==========
• 第 24 至 26 行：PojoUtils#realize(Object[] objs, Class<?>[] types, Type[] gtypes) 注意 generic = true，调用方法，反序列化参数。在该方法中，只有带有 class 属性的 Map ，需要反序列化成对应的 POJO 对象。
• 第 27 至 47 行：generic = nativejava，调用 NativeJavaSerialization#deserialize(url, input) 方法，反序列化参数，即 byte[] => 方法参数。
• 第 48 至 64 行：generic = bean，调用 JavaBeanSerializeUtil#deserialize(JavaBeanDescriptor) 方法，反序列化参数，即 JavaBeanDescriptor => 方法参数。
• ========== 【第二步：方法调用】 ==========
• 第 66 行：创建新的 RpcInvocation 对象。这是非常关键具体的一步，$invoke 的泛化调用，被转换成具体的普通调用。
• 第 66 行：调用 Invoker#invoke(invocation) 方法，继续过滤链的调用，最终调用 Service 服务。
• ========== 【第三步：序列化结果】 ==========
• 第 68 至 71 行：若是异常结果，并且非 GenericException 异常，可能这个异常在服务消费端是没有的，因此需要使用 GenericException 包装后返回。GenericException 的代码如下：

public class GenericException extends RuntimeException {

    private String exceptionClass;

    private String exceptionMessage;

    public GenericException() {
    }

    public GenericException(String exceptionClass, String exceptionMessage) {
        super(exceptionMessage);
        this.exceptionClass = exceptionClass;
        this.exceptionMessage = exceptionMessage;
    }

    public GenericException(Throwable cause) {
        super(StringUtils.toString(cause));
        this.exceptionClass = cause.getClass().getName();
        this.exceptionMessage = cause.getMessage();
    }

    // ... 省略 getting/setting 的方法
}

• 第 72 至 81 行：generic = nativejava，调用 NativeJavaSerialization#serialize(url, output) 方法，序列化结果，即 结果 => byte[]。
• 第 82 至 84 行：generic = bean，调用 JavaBeanSerializeUtil#serialize(Object, JavaBeanAccessor) 方法，序列化结果，即 结果 => JavaBeanDescriptor。
• 第 85 至 88 行：generic = true，调用 PojoUtils#generalize(Object) 方法，序列化结果，仅有 POJO => Map。

666. 彩蛋

艿艿在本文，并未解析 POJO 的序列化和反序列化的相关代码，感兴趣的胖友，可以自己研究。