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);
    }
}

@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 泛化調用與泛化實作原理 dubbo泛化調用原理

Dubbo 泛化引用和泛化實作

開篇

例子

序列化說明

源碼分析

泛化引用

參考

繼續閱讀

初識dubbo整合spring出現的service注入失敗的問題

手繪 SpringCloud微服務 VS DubboRPC服務

Dubbo的叢集容錯

Dubbo的負載均衡、叢集容錯、服務降級等機制詳解1. Dubbo與RPC的關系2. Dubbo的基本使用3. dubbo的REST協定4. dubbo的控制台5. dubbo的服務路由

springboot與dubbo版本比對問題

Dubbo|基礎知識之解析自定義标簽流程

Apache Dubbo 雲原生可觀測性的探索與實踐

RPC 分布式服務架構-Dubbo之負載均衡一、概述二、負載均衡政策三、配置

分布式與微服務系列（二）、分布式RPC架構Apache Dubbo服務一、 Apache Dubbo概述二、服務注冊中心Zookeeper三、Dubbo快速入門四、Dubbo管理控制台

持續內建篇【3】：SonarQube代碼品質管理平台的安裝及使用一、SonarQube代碼品質管理平台的安裝二、SonarQube代碼品質管理平台的配置與使用

Dubbo學習記錄1

Dubbo負載均衡政策Dubbo負載均衡政策

手把手實作RPC架構--簡易版Dubbo構造（三）用戶端動态代理

阿裡巴巴分布式服務架構 Dubbo 團隊成員梁飛專訪

阿裡巴巴分布式服務架構Dubbo介紹引言SOA化服務架構DubboDubbo的主要特點

Dubbo：來自于阿裡巴巴的分布式服務架構 Dubbo：來自于阿裡巴巴的分布式服務架構