Dubbo服务引用：Invoker对象创建与远程调用详解 1. AC/DC降压转换器设计基础与核心概念作为一名从业十余年的电源工程师我见过太多新手在AC/DC降压转换器设计上栽跟头。降压转换器看似简单实则暗藏玄机。让我们先明确一个基本概念AC/DC降压转换器的本质是将交流输入电压转换为更低的直流输出电压。这种转换过程涉及整流、滤波、稳压等多个环节每个环节都需要精心设计。降压转换器# 1. 概述本文分享服务引用的最后一部分创建 Invoker 对象。在 《精尽 Dubbo 源码分析 —— 服务引用一本地引用Injvm》 一文中我们看到#createInvokerForRemote()方法中在引用服务时会创建 Invoker 对象。代码如下/** * 创建远程引用 Invoker 对象 * * param map 集合参数 * param url 远程URL * param type 服务接口类型 * return Invoker 对象 * throws RpcException 当发生异常 */ private T InvokerT createInvokerForRemote(MapString, String map, URL url, Class? type) throws RpcException { // 如果为 injvm 协议直接返回空说明已经在 #createInvokerForLocal(map, url, type) 方法中生成 if (url.getProtocol().equals(Constants.REGISTRY_PROTOCOL)) { return null; } // 获得直连的地址 // 直连提供者地址 // 直连提供者是指在引用服务时不在通过注册中心而是直接向服务提供者发起调用 String extName url.getParameter(Constants.EXTENSION_KEY, Constants.DEFAULT_REMOTING_CLIENT); if (!Constants.DEFAULT_REMOTING_CLIENT.equals(extName) !url.getParameter(Constants.CLIENT_KEY, Constants.DEFAULT_REMOTING_CLIENT).equals(extName)) { url url.addParameter(Constants.CLIENT_KEY, extName); } // 交换协议由 registry 转换成具体协议 url url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME); // 创建 Invoker 对象 ExchangeClient[] clients getClients(url); // 创建 Dubbo Invoker 对象 DubboInvokerT invoker new DubboInvokerT(type, url, clients, invokers); invokers.add(invoker); return invoker; }第 10 至 12 行如果为registry协议直接返回空说明在Protocol$Adaptive#refer(ClassT type, URL url)方法中已经生成 Invoker 对象。例如RegistryProtocol 的#refer(...)方法。第 14 至 19 行获得直连的地址。即不通过注册中心直接连接服务提供者。第 21 行交换协议由registry转换成具体协议。例如dubbo。第 23 行调用#getClients(url)方法创建 ExchangeClient 对象数组。第 25 至 27 行创建 DubboInvoker 对象。第 28 行添加到invokers集合。2. ExchangeClient在 Dubbo 中Client指的是向服务器建立连接发起请求并接收响应的客户端。如下图所示在dubbo-remoting-api项目中com.alibaba.dubbo.remoting.Client接口代码如下public interface Client extends Endpoint, Channel, Resetable { /** * reconnect. * * 重连 */ void reconnect() throws RemotingException; }一个 Client 也是一个 Endpoint 和 Channel 。在dubbo-rpc-default项目中com.alibaba.dubbo.rpc.protocol.dubbo.Client接口代码如下public interface Client extends com.alibaba.dubbo.remoting.Client { /** * get channel. * * 获得通道 * * return channel */ Channel getChannel(); }继承com.alibaba.dubbo.remoting.Client接口增加了#getChannel()方法。在dubbo-rpc-default项目中com.alibaba.dubbo.rpc.protocol.dubbo.ReferenceCountExchangeClient类实现 Client 接口支持计数功能的 Client 包装实现类。在dubbo-rpc-default项目中com.alibaba.dubbo.rpc.protocol.dubbo.ExchangeClient接口代码如下public interface ExchangeClient extends Client, ExchangeChannel { }继承 Client 和 ExchangeChannel 接口。2.1 getClients#getClients(url)方法获得连接服务提供者的 ExchangeClient 对象数组。代码如下/** * 每个 DubboInvoker 的共享连接数 */ private static final int DEFAULT_CONNECTIONS 1; /** * 连接服务提供者 * * param url 服务提供者 URL * return Client 数组 */ private ExchangeClient[] getClients(URL url) { // 是否共享连接 boolean service_share_connect false; // 获得连接数 int connections url.getParameter(Constants.CONNECTIONS_KEY, 0); // 如果未配置 connections 则共享连接 if (connections 0) { service_share_connect true; connections DEFAULT_CONNECTIONS; } // 创建连接数组 ExchangeClient[] clients new ExchangeClient[connections]; for (int i 0; i clients.length; i) { // 共享连接 if (service_share_connect) { clients[i] getSharedClient(url); // 不共享连接 } else { clients[i] initClient(url); } } return clients; }connections属性每个 DubboInvoker 的共享连接数即每个 DubboInvoker 持有connections个 ExchangeClient 。默认值1。service_share_connect属性是否共享连接。有两种情况情况一当connections 0时共享连接。其中connections参数配置在url.parameters.connect键值对。情况二当connections ! 0时不共享连接。第 16 至 22 行创建 ExchangeClient 对象数组。第 18 行共享连接调用#getSharedClient(url)方法获得 ExchangeClient 对象。第 21 行不共享连接调用#initClient(url)方法直接创建 ExchangeClient 对象。2.1.1 getSharedClient#getSharedClient(url)方法获得共享的 ExchangeClient 对象。代码如下/** * 通信客户端集合 * * key: 服务地址。例如dubbo://127.0.0.1:20813/com.alibaba.dubbo.demo.DemoService?applicationdemo-consumercallbacks1000cellinvokemodesharingchannel.readonly.senttrueclientnetty4codecdubboconnections1heartbeat60000iddubboConsumerinterfacecom.alibaba.dubbo.demo.DemoServicemethodssayHello,callbackParam,save,update,say03,delete,say04,demo,say01,bye,say02,savespayload1000pid63400qos.port33333register.ip30.5.121.131remote.timestamp1527056490164sayHello.asynctruesayHello.timeout1000servernetty4sideconsumertimeout1000timestamp1527056489048 */ private final MapString, ReferenceCountExchangeClient referenceClientMap new ConcurrentHashMapString, ReferenceCountExchangeClient(); // host:port,ExchangerClient /** * TODO 8030 这个变量目前没用到。 */ private final ConcurrentMapString, LazyConnectExchangeClient ghostClientMap new ConcurrentHashMapString, LazyConnectExchangeClient(); /** * 获得共享连接 * * param url 服务提供者 URL * return 共享连接 */ private ExchangeClient getSharedClient(URL url) { // 从集合中查找 ReferenceCountExchangeClient 对象 String key url.toFullString(); ReferenceCountExchangeClient client referenceClientMap.get(key); if (client ! null) { // 若未关闭增加引用计数返回它 if (!client.isClosed()) { client.incrementAndGetCount(); return client; // 若已关闭移除 } else { referenceClientMap.remove(key); } } // 同步创建 ExchangeClient 对象 synchronized (key.intern()) { // 创建 ExchangeClient 对象 ExchangeClient exchangeClient initClient(url); // 将 ExchangeClient 对象包装创建 ReferenceCountExchangeClient 对象 client new ReferenceCountExchangeClient(exchangeClient, ghostClientMap); // 添加到集合 referenceClientMap.put(key, client); // 添加到 ghostClientMap ghostClientMap.remove(key); } return client; }referenceClientMap属性通信客户端集合。其中Key 为服务提供者 URL 的全字符串。ghostClientMap属性幽灵客户端集合。在 ReferenceCountExchangeClient 中会调用#close(...)方法时会调用GhostExchangeClient#close()方法将自己添加到ghostClientMap中。目前这个变量没用到官方 TODO 可能有其它用途。第 18 至 26 行从referenceClientMap集合中查找已存在的 ReferenceCountExchangeClient 对象。若未关闭增加引用计数并返回它。若已关闭移除。第 28 至 36 行同步创建 ExchangeClient 对象。第 30 行调用#initClient(url)方法创建 ExchangeClient 对象。第 32 行将 ExchangeClient 对象包装创建 ReferenceCountExchangeClient 对象。第 34 行添加到referenceClientMap集合。第 36 行从ghostClientMap中移除。2.1.2 initClient#initClient(url)方法直接创建 ExchangeClient 对象。代码如下/** * 创建新连接. * * param url 服务器地址 * return Client */ private ExchangeClient initClient(URL url) { // 校验 Client 的 Dubbo SPI 拓展是否存在 // client type setting. String str url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT)); // 添加编解码和心跳包参数 url url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME); // 默认开启 heartbeat // enable heartbeat by default url url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT)); // 校验传输层的 Dubbo SPI 拓展是否存在 // BIO is not allowed since it has severe performance issue. if (str ! null str.length() 0 !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str)) { throw new RpcException(Unsupported client type: str , supported client type is StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), )); } // 创建 ExchangeClient 对象 ExchangeClient client; try { // 懒连接创建时不建立连接 // connection should be lazy if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) { client new LazyConnectExchangeClient(url, requestHandler); // 直接连接 } else { client Exchangers.connect(url, requestHandler); } } catch (RemotingException e) { throw new RpcException(Fail to create remoting client for service( url ): e.getMessage(), e); } return client; }第 10 行获得 Client 的 Dubbo SPI 拓展名。若client参数不存在默认使用netty。第 12 行添加编解码参数。第 14 行添加心跳间隔参数。第 17 至 20 行校验 Client 的 Dubbo SPI 拓展是否存在。第 23 至 30 行创建 ExchangeClient 对象。第 25 至 26 行配置懒连接创建 LazyConnectExchangeClient 对象。在真正发生 RPC 调用时才建立连接。第 28 行调用Exchangers#connect(url, handler)方法创建 ExchangeClient 对象。后续文章详细解析。3. DubboInvokercom.alibaba.dubbo.rpc.protocol.dubbo.DubboInvoker实现 AbstractExporter 类Dubbo Invoker 实现类。代码如下/** * 使用的 {link #clients} 的位置 */ private final AtomicPositiveInteger index new AtomicPositiveInteger(); /** * 连接服务提供者的 ExchangeClient 集合 */ private final ExchangeClient[] clients; /** * 使用的 {link #clients} 的位置 */ private final AtomicPositiveInteger index new AtomicPositiveInteger(); public DubboInvoker(ClassT serviceType, URL url, ExchangeClient[] clients) { this(serviceType, url, clients, null); } public DubboInvoker(ClassT serviceType, URL url, ExchangeClient[] clients, SetInvoker? invokers) { super(serviceType, url, new String[]{Constants.INTERFACE_KEY, Constants.GROUP_KEY, Constants.VERSION_KEY, Constants.DEFAULT_KEY}); this.clients clients; // 首先指向第一个 ExchangeClient // get a random available client AtomicPositiveInteger index new AtomicPositiveInteger(); if (clients.length 1) { index.incrementAndGet(); } this.index index; if (invokers null) { invokers new ConcurrentHashSetInvoker?(); } this.invokers invokers; }clients属性连接服务提供者的 ExchangeClient 集合。在#getClients(url)方法中我们已经看到 ExchangeClient 的创建。index属性使用的clients的位置。在#doInvoke(Invocation)方法中我们会看到它的使用。3.1 doInvoke1: Override 2: protected Result doInvoke(final Invocation invocation) throws Throwable { 3: RpcInvocation inv (RpcInvocation) invocation; 4: // 获得方法名 5: final String methodName RpcUtils.getMethodName(invocation); 6: // 设置 path( 服务名 )和 version 到 attachments 中。 7: inv.setAttachment(Constants.PATH_KEY, getUrl().getPath()); 8: inv.setAttachment(Constants.VERSION_KEY, version); 9: 10: // 获得 ExchangeClient 对象 11: ExchangeClient currentClient; 12: if (clients.length 1) { 13: currentClient clients[0]; 14: } else { 15: currentClient clients[index.getAndIncrement() % clients.length]; 16: } 17: // 远程调用 18: try { 19: // 获得是否异步调用 20: boolean isAsync RpcUtils.isAsync(getUrl(), invocation); 21: // 获得是否单向调用 22: boolean isOneway RpcUtils.isOneway(getUrl(), invocation); 23: // 获得超时时间 24: int timeout getUrl().getMethodParameter(methodName, Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT); 25: // 单向调用 26: if (isOneway) { 27: boolean isSent getUrl().getMethodParameter(methodName, Constants.SENT_KEY, false); 28: currentClient.send(inv, isSent); 29: RpcContext.getContext().setFuture(null); 30: return new RpcResult(); 31: // 异步调用 32: } else if (isAsync) { 33: ResponseFuture future currentClient.request(inv, timeout); 34: RpcContext.getContext().setFuture(new FutureAdapterObject(future)); 35: return new RpcResult(); 36: // 同步调用 37: } else { 38: RpcContext.getContext().setFuture(null); 39: return (Result) currentClient.request(inv, timeout).get(); 40: } 41: } catch (TimeoutException e) { 42: throw new RpcException(RpcException.TIMEOUT_EXCEPTION, Invoke remote method timeout. method: invocation.getMethodName() , provider: getUrl() , cause: e.getMessage(), e); 43: } catch (RemotingException e) { 44: throw new RpcException(RpcException.NETWORK_EXCEPTION, Failed to invoke remote method: invocation.getMethodName() , provider: getUrl() , cause: e.getMessage(), e); 45: } 46: }第 4 行调用RpcUtils#getMethodName(invocation)方法获得方法名。第 5 至 8 行设置path( 服务名 )和version到attachments中。第 10 至 16 行获得 ExchangeClient 对象。若有多个 ExchangeClient 对象循环获得实现负载均衡。第 20 行调用RpcUtils#isAsync(url, invocation)方法判断是否异步调用。第 22 行调用RpcUtils#isOneway(url, invocation)方法判断是否异步调用。第 24 行调用URL#getMethodParameter(method, key, defaultValue)方法获得远程调用超时时间单位毫秒。 如下部分和 Dubbo 协议头flag的位操作相关 第 25 至 30 行单向调用。第 27 行调用URL#getMethodParameter(method, key, defaultValue)方法获得是否等待消息发出。sent true等待消息发出消息发送失败将抛出异常。sent false不等待消息发出将消息放入 IO 队列即刻返回。第 28 行调用ExchangeClient#send(invocation, sent)方法发送请求不需要Response 。第 29 行设置RpcContext.future null无需 FutureFilter 。第 30 行创建 RpcResult 对象空返回。第 31 至 36 行异步调用。第 33 行调用ExchangeClient#request(invocation, timeout)方法发送请求需要Response 。第 34 行调用RpcContext#setFuture(future)方法在 FutureFitler 中异步回调。第 35 行创建 RpcResult 对象空返回。第 37 至 40 行同步调用。第 38 行设置RpcContext.future null无需 FutureFilter 。第 39 行调用ExchangeClient#request(invocation, timeout)方法发送请求。调用ResponseFuture#get()方法阻塞等待返回结果。3.2 destroy1: Override 2: public void destroy() { 3: // 忽略若已经销毁 4: if (super.isDestroyed()) { 5: return; 6: // 标记已经销毁 7: } else { 8: // double check to avoid dup close 9: destroyLock.lock(); 10: try { 11: if (super.isDestroyed()) { 12: return; 13: } 14: super.destroy(); 15: // 移除出 invokers 集合 16: if (invokers ! null) { 17: invokers.remove(this); 18: } 19: // 关闭 ExchangeClient 连接 20: for (ExchangeClient client : clients) { 21: try { 22: client.close(); 23: } catch (Throwable t) { 24: logger.warn(t.getMessage(), t); 25: } 26: } 27: } finally { 28: destroyLock.unlock(); 29: } 30: } 31: }第 3 至 14 行通过双重检查保证仅销毁一次。第 15 至 18 行移除出invokers集合。第 19 至 26 行调用ExchangeClient#close()方法关闭 ExchangeClient 连接。666. 彩蛋 小文一篇希望对大家理解有帮助。