Dubbo源碼學習總結系列二 dubbo-rpc遠程調用模塊

        dubbo本質是一個RPC框架,咱們首先討論這個骨幹中的骨幹,dubbo-rpc模塊。html

        主要討論一下幾部份內容:java

        1、此模塊在dubbo總體框架中的做用;web

        2、此模塊須要完成的需求功能點及接口定義;spring

        3、DubboProtocol實現細節;api

        4、其餘協議實現;數組

 

        1、此模塊在dubbo總體框架中的做用緩存

        Protocol 是框架中的核心層,也就是隻要有 Protocol + Invoker + Exporter 就能夠完成非透明的 RPC 調用,而後在 Invoker 的主過程上 Filter 攔截點。它抽象了動態代理,只包含一對一的調用,不關心集羣的管理。安全

 

        2、此模塊須要完成的需求功能點及接口定義服務器

        遠程調用層:封裝 RPC 調用,以 InvocationResult 爲中心,擴展接口爲 Protocol,InvokerExporter。        併發

        遠程調用模塊實現Protocol,Invoker, Exporter等上層協議接口定義,實現DubboProtocol協議的上層實現,以及DubboCodec類(dubbo編碼)實現;封裝了Hession協議、RMI協議、Http協議、WebService協議、Rest協議、Thrift等協議的實現;抽象了動態代理,只包含一對一的調用,不關心集羣的管理。

        核心接口定義有:

        協議接口:Protocol

 1 package com.alibaba.dubbo.rpc;
 2 
 3 import com.alibaba.dubbo.common.URL;
 4 import com.alibaba.dubbo.common.extension.Adaptive;
 5 import com.alibaba.dubbo.common.extension.SPI;
 6 
 7 /**
 8  * Protocol. (API/SPI, Singleton, ThreadSafe)
 9  */
10 @SPI("dubbo")
11 public interface Protocol {
12 
13     /**
14      * Get default port when user doesn't config the port.
15      *
16      * @return default port
17      */
18     int getDefaultPort();
19 
20     /**
21      * Export service for remote invocation: <br>
22      * 1. Protocol should record request source address after receive a request:
23      * RpcContext.getContext().setRemoteAddress();<br>
24      * 2. export() must be idempotent, that is, there's no difference between invoking once and invoking twice when
25      * export the same URL<br>
26      * 3. Invoker instance is passed in by the framework, protocol needs not to care <br>
27      *
28      * @param <T>     Service type
29      * @param invoker Service invoker
30      * @return exporter reference for exported service, useful for unexport the service later
31      * @throws RpcException thrown when error occurs during export the service, for example: port is occupied
32      */
33     @Adaptive
34     <T> Exporter<T> export(Invoker<T> invoker) throws RpcException;
35 
36     /**
37      * Refer a remote service: <br>
38      * 1. When user calls `invoke()` method of `Invoker` object which's returned from `refer()` call, the protocol
39      * needs to correspondingly execute `invoke()` method of `Invoker` object <br>
40      * 2. It's protocol's responsibility to implement `Invoker` which's returned from `refer()`. Generally speaking,
41      * protocol sends remote request in the `Invoker` implementation. <br>
42      * 3. When there's check=false set in URL, the implementation must not throw exception but try to recover when
43      * connection fails.
44      *
45      * @param <T>  Service type
46      * @param type Service class
47      * @param url  URL address for the remote service
48      * @return invoker service's local proxy
49      * @throws RpcException when there's any error while connecting to the service provider
50      */
51     @Adaptive
52     <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException;
53 
54     /**
55      * Destroy protocol: <br>
56      * 1. Cancel all services this protocol exports and refers <br>
57      * 2. Release all occupied resources, for example: connection, port, etc. <br>
58      * 3. Protocol can continue to export and refer new service even after it's destroyed.
59      */
60     void destroy();
61 
62 }

        調用者接口Invoker:

 1 package com.alibaba.dubbo.rpc;
 2 
 3 import com.alibaba.dubbo.common.Node;
 4 
 5 /**
 6  * Invoker. (API/SPI, Prototype, ThreadSafe)
 7  *
 8  * @see com.alibaba.dubbo.rpc.Protocol#refer(Class, com.alibaba.dubbo.common.URL)
 9  * @see com.alibaba.dubbo.rpc.InvokerListener
10  * @see com.alibaba.dubbo.rpc.protocol.AbstractInvoker
11  */
12 public interface Invoker<T> extends Node {
13 
14     /**
15      * get service interface.
16      *
17      * @return service interface.
18      */
19     Class<T> getInterface();
20 
21     /**
22      * invoke.
23      *
24      * @param invocation
25      * @return result
26      * @throws RpcException
27      */
28     Result invoke(Invocation invocation) throws RpcException;
29 
30 }

        發佈者接口Exporter:

package com.alibaba.dubbo.rpc;

/**
 * Exporter. (API/SPI, Prototype, ThreadSafe)
 *
 * @see com.alibaba.dubbo.rpc.Protocol#export(Invoker)
 * @see com.alibaba.dubbo.rpc.ExporterListener
 * @see com.alibaba.dubbo.rpc.protocol.AbstractExporter
 */
public interface Exporter<T> {

    /**
     * get invoker.
     *
     * @return invoker
     */
    Invoker<T> getInvoker();

    /**
     * unexport.
     * <p>
     * <code>
     * getInvoker().destroy();
     * </code>
     */
    void unexport();

}

        請求調用上下文接口Invocation:

package com.alibaba.dubbo.rpc;

import java.util.Map;

/**
 * Invocation. (API, Prototype, NonThreadSafe)
 *
 * @serial Don't change the class name and package name.
 * @see com.alibaba.dubbo.rpc.Invoker#invoke(Invocation)
 * @see com.alibaba.dubbo.rpc.RpcInvocation
 */
public interface Invocation {

    /**
     * get method name.
     *
     * @return method name.
     * @serial
     */
    String getMethodName();

    /**
     * get parameter types.
     *
     * @return parameter types.
     * @serial
     */
    Class<?>[] getParameterTypes();

    /**
     * get arguments.
     *
     * @return arguments.
     * @serial
     */
    Object[] getArguments();

    /**
     * get attachments.
     *
     * @return attachments.
     * @serial
     */
    Map<String, String> getAttachments();

    /**
     * get attachment by key.
     *
     * @return attachment value.
     * @serial
     */
    String getAttachment(String key);

    /**
     * get attachment by key with default value.
     *
     * @return attachment value.
     * @serial
     */
    String getAttachment(String key, String defaultValue);

    /**
     * get the invoker in current context.
     *
     * @return invoker.
     * @transient
     */
    Invoker<?> getInvoker();

}

       

        dubbo-rpc-api模塊定義了基本接口,給出了默認實現類:

        Protocol接口的默認實現爲AbstractProtocol,AbstractProxyProtocol類

        (1)AbstractProtocol實現了destory()方法,主要邏輯爲:循環銷燬invoker、exporter。

        (2)AbstractProxyProtocol給出了export()和refer()的默認實現,分別返回AbstractExport和AbstractInvoker實例。

 

        3、DubboProtocol實現細節(dubbo-rpc-default);

        Dubbo 缺省協議採用單一長鏈接和 NIO 異步通信,適合於小數據量大併發的服務調用,以及服務消費者機器數遠大於服務提供者機器數的狀況。

        反之,Dubbo 缺省協議不適合傳送大數據量的服務,好比傳文件,傳視頻等,除非請求量很低。

        特性

        缺省協議,使用基於 mina 1.1.7 和 hessian 3.2.1 的 tbremoting 交互。

  • 鏈接個數:單鏈接
  • 鏈接方式:長鏈接
  • 傳輸協議:TCP
  • 傳輸方式:NIO 異步傳輸(默認netty)
  • 序列化:Hessian2 二進制序列化
  • 適用範圍:傳入傳出參數數據包較小(建議小於100K),消費者比提供者個數多,單一消費者沒法壓滿提供者,儘可能不要用 dubbo 協議傳輸大文件或超大字符串。
  • 適用場景:常規遠程服務方法調用

        Dubbo協議詳細說明參加dubbo用戶手冊

 

        核心類:DubboProtocol

        一、實現了export()方法,此方法實現了服務提供者接口的對外發布,返回Exporter對象,用於獲得invoker並調用,以及銷燬exporter對象。

        實現邏輯爲:

        (1)判斷是否須要建立stub支持,須要則在stubServiceMethodsMap緩存中存儲dubbo.stub.event.methods方法名;

    Stub本地存根:遠程服務後,客戶端一般只剩下接口,而實現全在服務器端,但提供方有些時候想在客戶端也執行部分邏輯,好比:作 ThreadLocal 緩存,提早驗證參數,調用失敗後僞造容錯數據等等,此時就須要在 API 中帶上 Stub,客戶端生成 Proxy 實例,會把 Proxy 經過構造函數傳給 Stub 1,而後把 Stub 暴露給用戶,Stub 能夠決定要不要去調 Proxy。

 

        (2)建立或更新(reset)遠程服務,並放到serverMap緩存中;建立遠程服務邏輯爲:給傳入的URL設置默認屬性,包括:當服務通道關閉時發送只讀事件(channel.readonly.sent),開啓心跳檢測(維護長鏈接),設置默認編碼方式爲dubbo,建立綁定url和請求處理類requestHandler的ExchangeServer並返回;

  1 public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
  2         URL url = invoker.getUrl();
  3 
  4         // export service.
  5         String key = serviceKey(url);
  6         DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
  7         exporterMap.put(key, exporter);
  8 
  9         //發佈一個本地存根服務,用於分發事件(猜想,不肯定)
 10         Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT);
 11         Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);
 12         if (isStubSupportEvent && !isCallbackservice) {
 13             String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
 14             if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
 15                 if (logger.isWarnEnabled()) {
 16                     logger.warn(new IllegalStateException("consumer [" + url.getParameter(Constants.INTERFACE_KEY) +
 17                             "], has set stubproxy support event ,but no stub methods founded."));
 18                 }
 19             } else {
 20                 stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
 21             }
 22         }
 23 
 24         openServer(url);
 25         optimizeSerialization(url);
 26         return exporter;
 27     }
 28 
 29     private void openServer(URL url) {
 30         // find server.
 31         String key = url.getAddress();
 32         //client can export a service which's only for server to invoke
 33         boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
 34         if (isServer) {
 35             ExchangeServer server = serverMap.get(key);
 36             if (server == null) {
 37                 serverMap.put(key, createServer(url));
 38             } else {
 39                 // server supports reset, use together with override
 40                 server.reset(url);
 41             }
 42         }
 43     }
 44 
 45     private ExchangeServer createServer(URL url) {
 46         // 當服務通道關閉時,發送只讀事件,默認爲True
 47         url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
 48         // 默認開啓心跳檢測
 49         url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
 50         //約定傳輸服務爲netty實現
 51         String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
 52 
 53         if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
 54             throw new RpcException("Unsupported server type: " + str + ", url: " + url);
 55         //默認使用dubbo序列化
 56         url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
 57         ExchangeServer server;
 58         try {
 59             //建立信息交換接口實現類,默認爲HeaderExchanger類實例,
 60             server = Exchangers.bind(url, requestHandler);
 61         } catch (RemotingException e) {
 62             throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
 63         }
 64         str = url.getParameter(Constants.CLIENT_KEY);
 65         if (str != null && str.length() > 0) {
 66             Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
 67             if (!supportedTypes.contains(str)) {
 68                 throw new RpcException("Unsupported client type: " + str);
 69             }
 70         }
 71         return server;
 72     }
 73 
 74     private void optimizeSerialization(URL url) throws RpcException {
 75         //獲得序列化優化參數,一般爲kryo或fst,將序列化實現設置爲制定類實例
 76         String className = url.getParameter(Constants.OPTIMIZER_KEY, "");
 77         if (StringUtils.isEmpty(className) || optimizers.contains(className)) {
 78             return;
 79         }
 80         
 81         logger.info("Optimizing the serialization process for Kryo, FST, etc...");
 82         
 83         try {
 84             Class clazz = Thread.currentThread().getContextClassLoader().loadClass(className);
 85             if (!SerializationOptimizer.class.isAssignableFrom(clazz)) {
 86                 throw new RpcException("The serialization optimizer " + className + " isn't an instance of " + SerializationOptimizer.class.getName());
 87             }
 88             
 89             SerializationOptimizer optimizer = (SerializationOptimizer) clazz.newInstance();
 90             
 91             if (optimizer.getSerializableClasses() == null) {
 92                 return;
 93             }
 94             
 95             for (Class c : optimizer.getSerializableClasses()) {
 96                 SerializableClassRegistry.registerClass(c);
 97             }
 98             
 99             optimizers.add(className);
100         } catch (ClassNotFoundException e) {
101             throw new RpcException("Cannot find the serialization optimizer class: " + className, e);
102         } catch (InstantiationException e) {
103             throw new RpcException("Cannot instantiate the serialization optimizer class: " + className, e);
104         } catch (IllegalAccessException e) {
105             throw new RpcException("Cannot instantiate the serialization optimizer class: " + className, e);
106         }
107     }

         (3)加載指定的序列化類

        調用optimizeSerialization(URL url)方法,將指定的序列化類加載到緩存,一般配置kryo, fst序列化,不配置,默認用hessian2序列化。

 

        二、實現refer()方法,此方法實現了服務消費者指向一個遠程地址的代理接口(invoker,多是一個假裝的集羣或本地調用),返回Invoker對象,用於服務調用。

        實現邏輯爲:

        (1)加載指定的序列化類

         調用optimizeSerialization(URL url)方法,將指定的序列化類加載到緩存,一般配置kryo, fst序列化,不配置,默認用hessian2序列化。

        (2)建立DubboInvoker對象,加入invokers內存緩存,返回dubboInvoker對象,結束。

1     public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
2         optimizeSerialization(url);
3         // create rpc invoker.
4         DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
5         invokers.add(invoker);
6         return invoker;
7     }

 

        (3)下面咱們來分析建立DubboInvoker對象的過程。

        建立時調用以下語句:new DubboInvoker<T>(serviceType, url, getClients(url), invokers)。原始構造方法定義以下:

1     public DubboInvoker(Class<T> serviceType, URL url, ExchangeClient[] clients, Set<Invoker<?>> invokers) {
2         super(serviceType, url, new String[]{Constants.INTERFACE_KEY, Constants.GROUP_KEY, Constants.TOKEN_KEY, Constants.TIMEOUT_KEY});
3         this.clients = clients;
4         // get version.
5         this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
6         this.invokers = invokers;
7     }

 

        傳入四個參數:

        serviceType:要調用的服務接口;

        url:根據消費端聲明肯定的調用url;

        clients:用於底層遠程通信信息交換的客戶端數組;

        invokers:DubboProtocol建立的全部invoker集合,此時共享給DubboInvoker對象;

        clients數組由DubboProtocol.refer()中使用getClients(url)獲得,咱們看看實現邏輯:

 1     private ExchangeClient[] getClients(URL url) {
 2         // 此變量用於判斷是不是共享鏈接
 3         boolean service_share_connect = false;
//獲得url中的connections數量,默認爲0,此設置的意義爲限制客戶端鏈接數量,具體含義見下面標註
4 int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0); 5 // if not configured, connection is shared, otherwise, one connection for one service 6 if (connections == 0) { 7 service_share_connect = true; 8 connections = 1; 9 } 10 11 ExchangeClient[] clients = new ExchangeClient[connections]; 12 for (int i = 0; i < clients.length; i++) { 13 if (service_share_connect) { 14 clients[i] = getSharedClient(url); 15 } else { 16 clients[i] = initClient(url); 17 } 18 } 19 return clients; 20 }

 

鏈接控制,此服務若是調用很是頻繁,能夠啓用鏈接池,指定最大長鏈接數。
客戶端鏈接控制:限制客戶端服務使用鏈接不能超過 10 個
<dubbo:reference interface="com.foo.BarService" connections="10" /><dubbo:service interface="com.foo.BarService" connections="10" />

服務端鏈接控制:限制服務器端接受的鏈接不能超過 10 個
<dubbo:provider protocol="dubbo" accepts="10" /><dubbo:protocol name="dubbo" accepts="10" />
若是是長鏈接,好比 Dubbo 協議,connections 表示該服務對每一個提供者創建的長鏈接數,若是服務提供方和消費方都設置了,默認用覆蓋策略,reference覆蓋provider。
getSharedClient(url)是獲得共享客戶端的方法,代碼以下:
 1     /**
 2      * Get shared connection
 3      */
 4     private ExchangeClient getSharedClient(URL url) {
 5         String key = url.getAddress();
 6         ReferenceCountExchangeClient client = referenceClientMap.get(key);
 7         if (client != null) {
 8             if (!client.isClosed()) {
 9                 client.incrementAndGetCount();
10                 return client;
11             } else {
12                 referenceClientMap.remove(key);
13             }
14         }
15         synchronized (key.intern()) {
16             ExchangeClient exchangeClient = initClient(url);
17             client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
18             referenceClientMap.put(key, client);
19             ghostClientMap.remove(key);
20             return client;
21         }
22     }
initClient(url)是根據傳入的url初始化鏈接客戶端,代碼以下:
 1 /**
 2      * Create new connection
 3      */
 4     private ExchangeClient initClient(URL url) {
 5 
 6         // client type setting.默認用netty
 7         String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
 8 
 9         String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
10         boolean compatible = (version != null && version.startsWith("1.0."));
11         url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
12         // enable heartbeat by default
13         url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
14 
15         // BIO is not allowed since it has severe performance issue.
16         if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) {
17             throw new RpcException("Unsupported client type: " + str + "," +
18                     " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
19         }
20 
21         ExchangeClient client;
22         try {
23             // connection should be lazy,建立可延遲鏈接的客戶端
24             if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
25                 client = new LazyConnectExchangeClient(url, requestHandler);
26             } else {
27                 client = Exchangers.connect(url, requestHandler);
28             }
29         } catch (RemotingException e) {
30             throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
31         }
32         return client;
33     }
三、 下面咱們分析DubboProtocol對ExchangeHandler接口的實現邏輯,就是創建鏈接用到的requestHandler對象,它是對request處理的接口對象。如下代碼把實現代碼隱藏了。
 1     private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
 2         //核心方法,對request傳來的message(即Invocation)的回覆,即調用相應invoker的invoke(invocation)方法,返回調用結果
 3         public Object reply(ExchangeChannel channel, Object message) throws RemotingException {}
32         //接受請求,若是message是一個Invocation,則調用reply()不返回結果,不然調用父類received(),其實什麼也不幹
34 public void received(Channel channel, Object message) throws RemotingException {} 41 //調用onconnect設置的方法,即當鏈接成功時觸發的方法
43 public void connected(Channel channel) throws RemotingException {} 46 //調用disconnect設置的方法,即當鏈接關閉時觸發的方法
48 public void disconnected(Channel channel) throws RemotingException {} 54 //調用方法的實現,內部根據methodKey調用createInvocation()建立Invocation對象,調用received()方法 55 private void invoke(Channel channel, String methodKey) {} 65 //建立Invocation對象並返回,其中根據URL的屬性和methedKey初始化Invocation對象 66 private Invocation createInvocation(Channel channel, URL url, String methodKey) {} 81 };

 

 1         public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
 2             if (message instanceof Invocation) {
 3                 Invocation inv = (Invocation) message;
//從緩存中獲得invoker對象
4 Invoker<?> invoker = getInvoker(channel, inv); 5 // 若是設置了回調服務方法,比較設置的回調方法在invoker對象中是否聲明瞭該方法 6 if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) { 7 String methodsStr = invoker.getUrl().getParameters().get("methods"); 8 boolean hasMethod = false; 9 if (methodsStr == null || methodsStr.indexOf(",") == -1) { 10 hasMethod = inv.getMethodName().equals(methodsStr); 11 } else { 12 String[] methods = methodsStr.split(","); 13 for (String method : methods) { 14 if (inv.getMethodName().equals(method)) { 15 hasMethod = true; 16 break; 17 } 18 } 19 } 20 if (!hasMethod) { 21 logger.warn(new IllegalStateException("The methodName " + inv.getMethodName() + " not found in callback service interface ,invoke will be ignored. please update the api interface. url is:" + invoker.getUrl()) + " ,invocation is :" + inv); 22 return null; 23 } 24 } 25 RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
//調用invoke並返回結果
26 return invoker.invoke(inv); 27 } 28 throw new RemotingException(channel, "Unsupported request: " + message == null ? null : (message.getClass().getName() + ": " + message) + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress()); 29 }

 

        代碼實現中常常看見ExtensionLoader.getExtensionLoader(Protocol.class).getExtension(DubboProtocol.NAME)相似的語句,這是Dubbo默認的類動態加載實例化的方式SPI,替代了SpringBean管理機制,實現了IOC和AOP的功能。之後的章節會詳細討論。

 

        最後,咱們分析一下實際的方法調用實現,即DubboInvoker.doInvoke()

 1     protected Result doInvoke(final Invocation invocation) throws Throwable {
 2         RpcInvocation inv = (RpcInvocation) invocation;
 3         final String methodName = RpcUtils.getMethodName(invocation);
 4         inv.setAttachment(Constants.PATH_KEY, getUrl().getPath());
 5         inv.setAttachment(Constants.VERSION_KEY, version);
 6 
 7         ExchangeClient currentClient;
 8         if (clients.length == 1) {
 9             currentClient = clients[0];
10         } else {
11             currentClient = clients[index.getAndIncrement() % clients.length];
12         }
13         try {
14             boolean isAsync = RpcUtils.isAsync(getUrl(), invocation);
15             boolean isOneway = RpcUtils.isOneway(getUrl(), invocation);
16             int timeout = getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
17             if (isOneway) { //不須要返回值,返回空的RpcResult對象
18                 boolean isSent = getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false);
19                 currentClient.send(inv, isSent);
20                 RpcContext.getContext().setFuture(null);
21                 return new RpcResult();
22             } else if (isAsync) { //異步調用,調用後能夠用RpcContext.getFuture()異步獲取返回結果
23                 ResponseFuture future = currentClient.request(inv, timeout);
24                 RpcContext.getContext().setFuture(new FutureAdapter<Object>(future));
25                 return new RpcResult();
26             } else {  //默認是同步調用,須要有返回結果,此時要清空Future
27                 RpcContext.getContext().setFuture(null);
28                 return (Result) currentClient.request(inv, timeout).get();
29             }
30         } catch (TimeoutException e) {
31             throw new RpcException(RpcException.TIMEOUT_EXCEPTION, "Invoke remote method timeout. method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
32         } catch (RemotingException e) {
33             throw new RpcException(RpcException.NETWORK_EXCEPTION, "Failed to invoke remote method: " + invocation.getMethodName() + ", provider: " + getUrl() + ", cause: " + e.getMessage(), e);
34         }
35     }

 

 

        4、其餘協議實現

        一、HessianProtocol實現

        以HTTP、HessianSkeleton爲底層實現,實現了doExport()和doRefer()方法。

        Hessian協議用於集成 Hessian 的服務,Hessian 底層採用 Http 通信,採用 Servlet 暴露服務,Dubbo 缺省內嵌 Jetty 做爲服務器實現。

        Dubbo 的 Hessian 協議能夠和原生 Hessian 服務互操做,即:

  • 提供者用 Dubbo 的 Hessian 協議暴露服務,消費者直接用標準 Hessian 接口調用
  • 或者提供方用標準 Hessian 暴露服務,消費方用 Dubbo 的 Hessian 協議調用。

        特性

  • 鏈接個數:多鏈接
  • 鏈接方式:短鏈接
  • 傳輸協議:HTTP
  • 傳輸方式:同步傳輸
  • 序列化:Hessian二進制序列化
  • 適用範圍:傳入傳出參數數據包較大,提供者比消費者個數多,提供者壓力較大,可傳文件。
  • 適用場景:頁面傳輸,文件傳輸,或與原生hessian服務互操做

        約束

  • 參數及返回值需實現 Serializable 接口
  • 參數及返回值不能自定義實現 ListMapNumberDateCalendar 等接口,只能用 JDK 自帶的實現,由於 hessian 會作特殊處理,自定義實現類中的屬性值都會丟失。

        配置

        <dubbo:protocol name="hessian" port="8080" server="jetty" />

         詳見dubbo用戶手冊Hessian協議說明

        二、RmiProtocol實現

        以spring的Rmi框架爲底層實現,實現了doExport()和doRefer()方法。

        RMI 協議採用 JDK 標準的 java.rmi.* 實現,採用阻塞式短鏈接和 JDK 標準序列化方式。

        注意:若是正在使用 RMI 提供服務給外部訪問 1,同時應用裏依賴了老的 common-collections 包 2 的狀況下,存在反序列化安全風險 3。

        特性

        鏈接個數:多鏈接
        鏈接方式:短鏈接
        傳輸協議:TCP
        傳輸方式:同步傳輸
        序列化:Java 標準二進制序列化
        適用範圍:傳入傳出參數數據包大小混合,消費者與提供者個數差很少,可傳文件。
        適用場景:常規遠程服務方法調用,與原生RMI服務互操做

        約束

        參數及返回值需實現 Serializable 接口
        dubbo 配置中的超時時間對 RMI 無效,需使用 java 啓動參數設置:-Dsun.rmi.transport.tcp.responseTimeout=3000,參見下面的 RMI 配置

        詳情見dubbo用戶手冊RMI協議說明

        三、其餘協議實現

        還實現了http協議,webservice協議,rest協議,thrift協議等,協議說明詳見dubbo用戶手冊

相關文章
相關標籤/搜索