Dubbo 泛化引用和泛化实现

开篇

 在Dubbo官方文档中关于泛化调用和泛化实现的说明，这里针对文档的案例做一些简单的说明和解释。

例子

// 引用远程服务 
// 该实例很重量，里面封装了所有与注册中心及服务提供方连接，请缓存
ReferenceConfig<GenericService> reference = new ReferenceConfig<GenericService>();
reference.setApplication(new ApplicationConfig("dubbo-demo-api-consumer"));
reference.setRegistry(new RegistryConfig("zookeeper://127.0.0.1:2181"));

// 弱类型接口名
reference.setInterface("org.apache.dubbo.demo.DemoService");
// 声明为泛化接口 
reference.setGeneric("true");

// 用org.apache.dubbo.rpc.service.GenericService可以替代所有接口引用  
GenericService genericService = reference.get();
// 基本类型以及Date,List,Map等不需要转换，直接调用 
Object result = genericService.$invoke("sayHello",
        new String[] {"java.lang.String"}, new Object[] {"12345678"});

System.out.println(result.toString());

• 例子在注释中已经注明了泛化引用的一般步骤，这里不再赘述。
• 核心的本质在于reference.get()的流程获取了consumer的对象，等同于xml通过reference初始化consumer的bean对象。
• 本质上解析生成consumer的bean对象的过程就是泛化过程中reference.get()的过程。
• 泛化调用唯一需要注意的是对参数的一些限制，基本上参数不需要转换，Class对象等参数需要把保存Class对象的完整类名。

序列化说明

public class Person {

    public Person() {
    }

    public Person(String name, Integer age) {
        this.name = name;
        this.age = age;
    }

    private String name;
    private Integer age;

    public String getName() {
        return name;
    }

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

    public Integer getAge() {
        return age;
    }

    public void setAge(Integer age) {
        this.age = age;
    }
}



代码例子-------------

public class Demo {
    public static void main(String[] args) {
        Person person = new Person();
        person.setName("person");
        person.setAge(123);

        System.out.println("序列化后对象：");
        System.out.println(JSON.toJSONString(PojoUtils.generalize(person)));

        System.out.println("反序列化后对象：");
        System.out.println(JSON.toJSONString(PojoUtils.realize(PojoUtils.generalize(person), Person.class)));
    }
}

运行结果-----------------

序列化后对象：
{"name":"person","class":"org.apache.dubbo.demo.consumer.Person","age":123}

反序列化后对象：
{"age":123,"name":"person"}

• 一般的泛化调用都是普通的reference引用，在泛化调用过程中通过PojoUtils.generalize/realize进行序列化和反序列化。
• 在序列化过程中Person对象通过PojoUtils.generalize()序列化的结果当中增加了class字段，然后对序列化的结果进行反序列才能正常反序列化。
• 在dubbo官网有下列说明，指明在泛化调用参数为POJO对象的时候通过使用Map需要带上class类型。

泛化调用序列化补充

源码分析

• Dubbo泛化调用实际是在filter过滤链上执行的序列化和反序列化操作。
• genericimpl对应consumer侧的泛化调用操作。
• generic对应的为provider侧的返回调用操作。

com.alibaba.dubbo.rpc.Filter文件内容

generic=com.alibaba.dubbo.rpc.filter.GenericFilter
genericimpl=com.alibaba.dubbo.rpc.filter.GenericImplFilter

泛化引用

@Activate(group = CommonConstants.CONSUMER, value = GENERIC_KEY, order = 20000)
public class GenericImplFilter extends ListenableFilter {

    private static final Class<?>[] GENERIC_PARAMETER_TYPES = 
                new Class<?>[]{String.class, String[].class, Object[].class};

    public GenericImplFilter() {
        super.listener = new GenericImplListener();
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {

        String generic = invoker.getUrl().getParameter(GENERIC_KEY);
        //服务端是泛化暴露，客户端不是使用泛化调用场景
        if (ProtocolUtils.isGeneric(generic)
                && (!$INVOKE.equals(invocation.getMethodName()) && !$INVOKE_ASYNC.equals(invocation.getMethodName()))
                && invocation instanceof RpcInvocation) {
            RpcInvocation invocation2 = new RpcInvocation(invocation);
            String methodName = invocation2.getMethodName();
            Class<?>[] parameterTypes = invocation2.getParameterTypes();
            Object[] arguments = invocation2.getArguments();

            String[] types = new String[parameterTypes.length];
            for (int i = 0; i < parameterTypes.length; i++) {
                types[i] = ReflectUtils.getName(parameterTypes[i]);
            }

            Object[] args;
            // 客户端(非泛化)到服务端(泛化)根据不同协议进行序列化
            if (ProtocolUtils.isBeanGenericSerialization(generic)) {
                args = new Object[arguments.length];
                for (int i = 0; i < arguments.length; i++) {
                    args[i] = JavaBeanSerializeUtil.serialize(arguments[i], JavaBeanAccessor.METHOD);
                }
            } else {
                args = PojoUtils.generalize(arguments);
            }

            if (RpcUtils.isReturnTypeFuture(invocation)) {
                invocation2.setMethodName($INVOKE_ASYNC);
            } else {
                invocation2.setMethodName($INVOKE);
            }
            invocation2.setParameterTypes(GENERIC_PARAMETER_TYPES);
            invocation2.setArguments(new Object[]{methodName, types, args});
            // 客户端调用转换为服务端的泛化调用
            return invoker.invoke(invocation2);

         // 服务端非泛化暴露，消费使用泛化调用  
        } else if ((invocation.getMethodName().equals($INVOKE) || invocation.getMethodName().equals($INVOKE_ASYNC))
                && invocation.getArguments() != null
                && invocation.getArguments().length == 3
                && ProtocolUtils.isGeneric(generic)) {

            Object[] args = (Object[]) invocation.getArguments()[2];
            // 校验不同的序列化格式是否正确
            if (ProtocolUtils.isJavaGenericSerialization(generic)) {

                for (Object arg : args) {
                    if (!(byte[].class == arg.getClass())) {
                        error(generic, byte[].class.getName(), arg.getClass().getName());
                    }
                }
            } else if (ProtocolUtils.isBeanGenericSerialization(generic)) {
                for (Object arg : args) {
                    if (!(arg instanceof JavaBeanDescriptor)) {
                        error(generic, JavaBeanDescriptor.class.getName(), arg.getClass().getName());
                    }
                }
            }

            invocation.setAttachment(
                    GENERIC_KEY, invoker.getUrl().getParameter(GENERIC_KEY));
        }

        return invoker.invoke(invocation);
    }
}

Dubbo泛化引用过程 - consumer

泛化实现

@Activate(group = CommonConstants.PROVIDER, order = -20000)
public class GenericFilter extends ListenableFilter {

    public GenericFilter() {
        super.listener = new GenericListener();
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation inv) throws RpcException {
        // 如果方法名为$invoker，并且只有3个参数，
        // 并且服务端暴露的invoker不是GenericService的相关类
        // 则认为本次服务调用时客户端泛化引用服务端，客户端的泛化调用，
        // 需要将请求参数反序列化为该接口真实的pojo对象。
        if ((inv.getMethodName().equals($INVOKE) || inv.getMethodName().equals($INVOKE_ASYNC))
                && inv.getArguments() != null
                && inv.getArguments().length == 3
                && !GenericService.class.isAssignableFrom(invoker.getInterface())) {

            String name = ((String) inv.getArguments()[0]).trim();
            String[] types = (String[]) inv.getArguments()[1];
            Object[] args = (Object[]) inv.getArguments()[2];
            try {
                // 根据接口名（API类）、方法名、方法参数类型列表，根据反射机制获取对应的方法。
                Method method = ReflectUtils.findMethodByMethodSignature(invoker.getInterface(), name, types);
                Class<?>[] params = method.getParameterTypes();
                if (args == null) {
                    args = new Object[params.length];
                }
                String generic = inv.getAttachment(GENERIC_KEY);

                if (StringUtils.isBlank(generic)) {
                    generic = RpcContext.getContext().getAttachment(GENERIC_KEY);
                }
                
              // 处理普通的泛化引用调用，即处理<dubbo:referecnce generic=“true” …/>,
              // 只需要将参数列表Object[]反序列化为pojo即可，
              // 具体的反序列化为PojoUtils#realize,
              // 其实现原理如下：在JAVA的世界中，pojo通常用map来表示，
              // 也就是一个Map可以用来表示一个对象的值，那从一个Map如果序列化一个对象呢？
              // 其关键的要素是要在Map中保留该对象的类路径名，
              // 也就是通过class来标识该Map需要反序列化的pojo类型。

                if (StringUtils.isEmpty(generic)
                        || ProtocolUtils.isDefaultGenericSerialization(generic)
                        || ProtocolUtils.isGenericReturnRawResult(generic)) {

                    args = PojoUtils.realize(args, params, method.getGenericParameterTypes());

                // 处理< dubbo:reference generic=“nativejava” /> 启用泛化引用，
                // 并使用nativejava序列化参数，在服务端这边通过nativejava反序列化参数成pojo对象。
                } 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(GENERIC_SERIALIZATION_NATIVE_JAVA)
                                        .deserialize(null, is).readObject();
                            } catch (Exception e) {
                                throw new RpcException("Deserialize argument [" + (i + 1) + "] failed.", e);
                            }
                        } else {
                            throw new RpcException(
                                    "Generic serialization [" +
                                            GENERIC_SERIALIZATION_NATIVE_JAVA +
                                            "] only support message type " +
                                            byte[].class +
                                            " and your message type is " +
                                            args[i].getClass());
                        }
                    }

                // 处理< dubbo:reference generic=“bean” /> 启用泛化引用，
                // 并使用javabean序列化参数，在服务端这边通过javabean反序列化参数成pojo对象。
                } 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(
                                    "Generic serialization [" +
                                            GENERIC_SERIALIZATION_BEAN +
                                            "] only support message type " +
                                            JavaBeanDescriptor.class.getName() +
                                            " and your message type is " +
                                            args[i].getClass().getName());
                        }
                    }

                } else if (ProtocolUtils.isProtobufGenericSerialization(generic)) {
                    // as proto3 only accept one protobuf parameter
                    if (args.length == 1 && args[0] instanceof String) {
                        try (UnsafeByteArrayInputStream is =
                                     new UnsafeByteArrayInputStream(((String) args[0]).getBytes())) {
                            args[0] = ExtensionLoader.getExtensionLoader(Serialization.class)
                                    .getExtension("" + GENERIC_SERIALIZATION_PROTOBUF)
                                    .deserialize(null, is).readObject(method.getParameterTypes()[0]);
                        } catch (Exception e) {
                            throw new RpcException("Deserialize argument failed.", e);
                        }
                    } else {
                        throw new RpcException(
                                "Generic serialization [" +
                                        GENERIC_SERIALIZATION_PROTOBUF +
                                        "] only support one" + String.class.getName() +
                                        " argument and your message size is " +
                                        args.length + " and type is" +
                                        args[0].getClass().getName());
                    }
                }
                // 序列化API方法中声明的类型，构建new RpcInvocation(method, args, inv.getAttachments())调用环境，继续调用后续过滤器。
                return invoker.invoke(new RpcInvocation(method, args, inv.getAttachments()));
            } catch (NoSuchMethodException e) {
                throw new RpcException(e.getMessage(), e);
            } catch (ClassNotFoundException e) {
                throw new RpcException(e.getMessage(), e);
            }
        }
        return invoker.invoke(inv);
    }
}

Dubbo泛化引用过程 - provider

参考

源码分析Dubbo 泛化调用与泛化实现原理
dubbo泛化调用原理