Dubbo的Consumer启动过程（通过URL生成Invoker）

朴灿烈づ我的快乐病毒、 2022-05-23 07:21 205阅读 0赞

在上一篇博客中，我们详解讲述了URL的生成过程，从Zookeeper订阅到Providers的url字符串，解析成URL对象，然后按照匹配规则选择合适的URL。

这篇博客主要讲解根据Provider的URL信息，连接Provider，创建Channel，生成Invoker对象，以为做RPC远程调用的客户端。

/** * AbstractRegistry. (SPI, Prototype, ThreadSafe) * * Registry 抽象类，实现了如下方法： * * 1、通用的注册、订阅、查询、通知等方法 * 2、读取和持久化注册数据到文件，以 properties 格式存储 */
    public abstract class AbstractRegistry implements Registry { 
    
         /** * Turn urls into invokers, and if url has been refer, will not re-reference. * 将每个provider映射为相应的Invoker, Invoker内包含consumer和provider的ExchangeClient，也就是连接和信息转换对象 * * @param urls * @return invokers */
        private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
            ......
            if (enabled) {
                 // 分发式的Invoker, 便于负载均衡 , protocol.refer(serviceType, url) refer Provider,生成Invoker
                 // protocol >> Protocol$Adaptive , 通过ExtesionLoader获取到的是 Wrapper包装,比如 ProtocolFilterWrapper
                 invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);
             }
        }
    
    }

接着上段代码的：**invoker = new InvokerDelegate(protocol.refer(serviceType, url), url, providerUrl);**接续讲。

public class DubboProtocol extends AbstractProtocol { 
    
        @Override
        public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
            // 初始化序列化优化器
            optimizeSerialization(url);
            // 获得远程通信客户端数组
            // 创建 DubboInvoker 对象
            // create rpc invoker.
            DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
            // 添加到 `invokers`
            invokers.add(invoker);
            return invoker;
        }
    
        /** * 获得连接服务提供者的远程通信客户端数组 * * @param url 服务提供者 URL * @return 远程通信客户端 */
        private ExchangeClient[] getClients(URL url) {
            // 是否共享连接
            // whether to share connection
            boolean service_share_connect = false;
            int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
            // if not configured, connection is shared, otherwise, one connection for one service
            if (connections == 0) { // 未配置时，默认共享, 也就是一个Client和Server之间共用一个ExchangeClient
                service_share_connect = true;
                connections = 1;
            }
    
            // 创建连接服务提供者的 ExchangeClient 对象数组
            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;
        }
    
        /** * Get shared connection * * 获得 ExchangeClient 对象。若集合中已经存在，则直接使用，无需创建。否则，创建 ExchangeClient 对象。 * 也就是说一个Consumer连接一个Provider会共用一个ExchangeClient, */
        private ExchangeClient getSharedClient(URL url) {
            // 从集合中，查找 ReferenceCountExchangeClient 对象, key 是 ip：port
            String key = url.getAddress();
            ReferenceCountExchangeClient client = referenceClientMap.get(key);
            if (client != null) {
                // 若未关闭，增加指向该 Client 的数量，并返回它
                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;
            }
        }
    
    
    }

**对于一个Consumer和Provider之间，默认情况下只创建一个Channel，也就是说创建一个通道，他们之间的数据传输共用同一个Channel，这样能够有效减少服务器间的长连接数量，提高性能。**

public class DubboProtocol extends AbstractProtocol { 
    
        /** * Create new connection * * 创建 ExchangeClient 对象，"连接"服务器 * ####################### //TODO 这一步最关键,连接URL指定的Server,创建Channell 封装成Client ####################### * * @param url provider 的 url * @return */
        private ExchangeClient initClient(URL url) {
            // 校验 Client 的 Dubbo SPI 拓展是否存在
            // client type setting. 传输类型,默认值 "netty"
            String transType = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));
            // BIO is not allowed since it has severe performance issue.
            if (transType != null && transType.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(transType)) {
                throw new RpcException("Unsupported client type: " + transType + "," +
                    " supported client type is " + StringUtils.join(ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
            }
    
            // 设置编解码器为 Dubbo ，即 DubboCountCodec
            url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
    
            // 默认开启 heartbeat, 心跳间隔默认为 60*1000 ,也就是一分钟
            // enable heartbeat by default
            url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
    
            // 连接服务器，创建客户端
            ExchangeClient client;
            try {
                // 懒连接，创建 LazyConnectExchangeClient 对象
                // connection should be lazy
                if (url.getParameter(Constants.LAZY_CONNECT_KEY, false)) {
                    client = new LazyConnectExchangeClient(url, requestHandler);
                    // 直接连接，创建 HeaderExchangeClient 对象
                } 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;
        }
    
    }
    
    
    public class Exchangers { 
    
        /** * @param url * @param handler {@link DubboProtocol#requestHandler} * @return {@link HeaderExchangeClient}, 也就是生成Consumer客户端 Client * @throws RemotingException */
        public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
            if (url == null) {
                throw new IllegalArgumentException("url == null");
            }
            if (handler == null) {
                throw new IllegalArgumentException("handler == null");
            }
            url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
            /** * {@link HeaderExchanger#connect(URL, ExchangeHandler)} */
            return getExchanger(url).connect(url, handler);
        }
    
    }

创建ExchangeClient的过程太过繁杂，我仅仅列出一些关键的步骤，加上代码的注释和说明：

public class NettyClient extends AbstractClient { 
    
        /** * 实例化ChannelHandler,此类间接实现了ChannelHandler的所有方法,用于信息的处理 * * @param url * @param handler * @throws RemotingException */
        public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
            super(url, wrapChannelHandler(url, handler));
        }
    
        protected void doOpen() {
            // 设置日志工厂
            NettyHelper.setNettyLoggerFactory();
    
            // 实例化 ServerBootstrap
            bootstrap = new ClientBootstrap(channelFactory);
            // 设置可选项
            // config
            // @see org.jboss.netty.channel.socket.SocketChannelConfig
            bootstrap.setOption("keepAlive", true);
            bootstrap.setOption("tcpNoDelay", true);
            bootstrap.setOption("connectTimeoutMillis", getTimeout());
    
            /**创建 NettyHandler 对象,其作为Netty的一个ChannelHandler，参与Channel内数据的处理 * NettyHandler extends {@link org.jboss.netty.channel.SimpleChannelHandler} , 间接实现了 {@link ChannelUpstreamHandler},{@link ChannelDownstreamHandler} * ##################### 其将Netty的Channel里获取到的数据导入Dubbo体系中,至关重要 ##################### */
            final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
    
            // 设置责任链路
            bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
                @Override
                public ChannelPipeline getPipeline() {
                    // 创建 NettyCodecAdapter 对象
                    NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
                    ChannelPipeline pipeline = Channels.pipeline();
                    pipeline.addLast("decoder", adapter.getDecoder());  // 解码 ChannelHandler
                    pipeline.addLast("encoder", adapter.getEncoder());  // 编码 ChannelHandler
                    // ##################### 处理器 , 将NettyHandler(url,this)作为Netty的一个ChannelHandler, 用于处理数据的交互 #####################
                    pipeline.addLast("handler", nettyHandler);
                    return pipeline;
                }
            });
        }
    
    
    }

实例化ExchangeClient时会得到一个NettyClient 对象，同时在开启Channel时，生成一个**NettyHandler** 对象：

final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
    
    @Sharable
    public class NettyHandler extends SimpleChannelHandler { 
    
        /** * URL */
        private final URL url;
    
        /** * 属性类型为：NettyClient, 当前对象的很多行为转发到 {@link NettyClient} * NettyClient将部分行为转发给其持有的{@link NettyClient#getChannelHandler()} */
        private final ChannelHandler handler;
    
        @Override
        public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
            // 创建 NettyChannel 对象
            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
            try {
                // 添加到 `channels` 中
                if (channel != null) {
                    channels.put(NetUtils.toAddressString((InetSocketAddress)ctx.getChannel().getRemoteAddress()), channel);
                }
                // 提交给 `NettyClient(superClass AbstractPeer)` 处理器。
                handler.connected(channel);
            } finally {
                // 移除 NettyChannel 对象，若已断开
                NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
            }
        }
    
        @Override
        public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
            try {
                channels.remove(NetUtils.toAddressString((InetSocketAddress)ctx.getChannel().getRemoteAddress()));
                handler.disconnected(channel);
            } finally {
                NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
            }
        }
    
        @Override
        public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
            try {
                handler.received(channel, e.getMessage());
            } finally {
                NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
            }
        }
    
        @Override
        public void writeRequested(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
            super.writeRequested(ctx, e);
            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
            try {
                handler.sent(channel, e.getMessage());
            } finally {
                NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
            NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler);
            try {
                handler.caught(channel, e.getCause());
            } finally {
                NettyChannel.removeChannelIfDisconnected(ctx.getChannel());
            }
        }
    
    }

**NettyHandler 实现了SimpleChannelHandler 接口，这个接口是Netty框架的一个通道处理器，实现其特定的方法，然后将此ChannelHandler嵌入Netty的ChannelPipeline 中，就能够在Netty发生ChannelEvent（连接，断开，写入，读取等）时，执行相应的逻辑。**

**NettyHandler将Netty的时间处理转交给了其内部持有的ChannelHandler 对象，这个对象其内部最终会将处理逻辑转交给DubboProtocol的匿名属性requestHandler ：**

public class DubboProtocol extends AbstractProtocol { 
    
        /** * 信息转化处理器, 从Netty的接受和发送消息最终都会调用此处 */
        private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {
    
            @Override
            public Object reply(ExchangeChannel channel, Object message) throws RemotingException {
                if (message instanceof Invocation) {
                    Invocation inv = (Invocation)message;
                    // 获得请求对应的 Invoker 对象
                    Invoker<?> invoker = getInvoker(channel, inv);
                    // 如果是callback 需要处理高版本调用低版本的问题
                    // need to consider backward-compatibility if it's a callback
                    if (Boolean.TRUE.toString().equals(inv.getAttachments().get(IS_CALLBACK_SERVICE_INVOKE))) {
                        String methodsStr = invoker.getUrl().getParameters().get("methods");
                        boolean hasMethod = false;
                        if (methodsStr == null || !methodsStr.contains(",")) {
                            hasMethod = inv.getMethodName().equals(methodsStr);
                        } else {
                            String[] methods = methodsStr.split(",");
                            for (String method : methods) {
                                if (inv.getMethodName().equals(method)) {
                                    hasMethod = true;
                                    break;
                                }
                            }
                        }
                        if (!hasMethod) {
                            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);
                            return null;
                        }
                    }
                    // 设置调用方的地址
                    RpcContext.getContext().setRemoteAddress(channel.getRemoteAddress());
                    // 执行调用
                    return invoker.invoke(inv);
                }
                throw new RemotingException(channel, message.getClass().getName() + ": " + message
                    + ", channel: consumer: " + channel.getRemoteAddress() + " --> provider: " + channel.getLocalAddress());
            }
    
            @Override
            public void received(Channel channel, Object message) throws RemotingException {
                if (message instanceof Invocation) {
                    this.reply((ExchangeChannel)channel, message);
                } else {
                    super.received(channel, message);
                }
            }
    
            @Override
            public void connected(Channel channel) {
                this.invoke(channel, Constants.ON_CONNECT_KEY);
            }
    
            @Override
            public void disconnected(Channel channel) throws RemotingException {
                if (logger.isInfoEnabled()) {
                    logger.info("disconected from " + channel.getRemoteAddress() + ",url:" + channel.getUrl());
                }
                this.invoke(channel, Constants.ON_DISCONNECT_KEY);
            }
    
            /** * 调用方法 * * @param channel 通道 * @param methodKey 方法名 */
            private void invoke(Channel channel, String methodKey) {
                // 创建 Invocation 对象
                Invocation invocation = createInvocation(channel, channel.getUrl(), methodKey);
                // 调用 received 方法，执行对应的方法
                if (invocation != null) {
                    try {
                        this.received(channel, invocation);
                    } catch (Throwable t) {
                        logger.warn("Failed to invoke event method " + invocation.getMethodName() + "(), cause: " + t.getMessage(), t);
                    }
                }
            }
    
            private Invocation createInvocation(Channel channel, URL url, String methodKey) {
                String method = url.getParameter(methodKey);
                if (method == null || method.length() == 0) {
                    return null;
                }
                RpcInvocation invocation = new RpcInvocation(method, new Class<?>[0], new Object[0]);
                invocation.setAttachment(Constants.PATH_KEY, url.getPath());
                invocation.setAttachment(Constants.GROUP_KEY, url.getParameter(Constants.GROUP_KEY));
                invocation.setAttachment(Constants.INTERFACE_KEY, url.getParameter(Constants.INTERFACE_KEY));
                invocation.setAttachment(Constants.VERSION_KEY, url.getParameter(Constants.VERSION_KEY));
                if (url.getParameter(Constants.STUB_EVENT_KEY, false)) {
                    invocation.setAttachment(Constants.STUB_EVENT_KEY, Boolean.TRUE.toString());
                }
                return invocation;
            }
        };
    
    }

**Dubbo对DubboProtocol 里的匿名属性requestHandler 做了多次封装，使用装饰模式在每一个装饰器上都添加了额外的功能，装饰链如下：**

ChannelHandler封装层级： MultiMessageHandler >> HeartbeatHandler >> AllChannelHandler >> DecodeHandler >> HeaderExchangeHandler >> requestHandler

最终生成的Invoker对象就是DubboInvoker，当然如果有多个Invoker，那么外层会封装成一个集群式的Invoker ： FailoverClusterInvoker，内部持有多个Invoker组合成的目录：Directory实现对象，以便做负载均衡和集群容错。集群容错有多个策略可选择，默认是FailoverClusterInvoker。

以上就是Dubbo通过URL生成Invoker的过程，当然中间省去了很多过程，不过关键的几个步骤基本都将了，如果真的对其源码和流程很感兴趣，还是需要亲自翻翻源码的，仅靠一篇博客肯定是不够的，不过这篇博客应该对如何理解Dubbo的初始化过程有一定的帮助。