Dubbo学习笔记11：Dubbo服务消费方启动流程源码分析

同理我们看下服务消费端启动流程时序图：

在《Dubbo整体架构分析》一文中，我们提到服务消费方需要使用ReferenceConfig API来消费服务，具体是调用代码（1）get()方法来生成远程调用代理类。get()方法最终会调用createProxy方法来具体创建代理类，其中createProxy结合时序图的核心代码如下：

@SuppressWarnings({"unchecked" , "rawtypes" , "deprecation"})
private T createProxy(Map<String , String> map){
    ...
    if(isJvmRefer){
        ...
    }else{
        ...
        // (1) 当只配置一个服务中心的时候
        if(urls.size() == 1){
            invoker = refprotocol.refer(interfaceClass , urls.get(0));
        }else{
            // (2) 多个服务中心的时候
            ...
        }
    }
    ...
    // (3) 创建代理服务
    return (T)proxyFactory.getProxy(invoker);
}

其中refprotocol的定义如下：

private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

可知refprotocol是Protocol扩展接口的适配器类，这里调用refprotocol.refer(interfaceClass , urls.get(0));实际调用的是Protocol$Adaptive的refer方法。从Protocol$Adaptive的refer方法内部，我们可以发现当前协议类型为registry，也就是这里需要调用RegistryProtocol的refer方法，但是RegistryProtocol被QosProtocolWrapper / ProtocolFilterWrapper / ProtocolListenerWrapper三个wrapper类增强了。所以这里经过一层层调用后，最后调用到了RegistryProtocol的refer方法，其内部主要是调用了doRefer方法，doRefer代码如下：

private <T> Invoker<T> doRefer(Cluster cluster , Registry registry , Class<T> type , URL url){
    ...
    // (3)
    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY , Constants.PROVIDERS_CATEGORY + "," + Constants.CONFIGURATORS_CATEGORY + "," + Constants.ROUTERS_CATEGORY));
    // (4)
    return cluster.join(directory);
}

如上代码（3）的作用是向服务注册中心订阅服务提供者的服务，代码（4）则是调用扩展接口Cluster的适配器类的join方法，根据参数选择配置的集群容错策略。这里我们先讲讲代码（3）的逻辑，看看如何把服务消费方远程服务转换到Invoker，这里结合Zookeeper作为服务治理中心来讲解，首先看看时序图：

如上时序图步骤（2）（3）（4），从Zookeeper获取服务提供者的地址列表，等Zookeeper返回地址列表后会调用RegistryDirectory的notify方法，代码（6）（7）（8）根据获取的最新的服务提供者url地址转换为具体的invoker列表，也就是每个提供者的url会被转换为一个Invoker对象，具体转换在toInvokers方法中进行：

private Map<String , Invoker<T>> toInvokers(List<URL> urls){
    Map<String , Invoker<T>> newUrlInvokerMap = new HashMap<String , Invoker<T>>();
    ...
    for(URL providerUrl : urls){
        ...
        Map<String , Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;    //
        Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
        if(invoker == null){
            try{
                ...
                if(enabled){
                    // 这里具体调用dubbo协议转换服务为invoker
                    invoker = new InvokerDelegete<T>(protocol.refer(serviceType , url) , url , providerUrl);
                }
            }catch(Throwable t){
                
            }
            if(invoker != null){
                newUrlInvokerMap.put(key , invoker);
            }
        }else{
            newUrlInvokerMap.put(key , invoker);
        }
    }
    return newUrlInvokerMap;
}

如上代码，具体转换服务到invoker对象是通过调用protocol.refer(serviceType , url)来完成的，这里的protocol对象也是Protocol扩展接口的适配器对象，所以调用protocol.refer实际是调用Protocol$Adaptive的refer方法。url中协议默认为dubbo，所以适配器里调用的应该是DubboProtocol的refer方法。

前面章节也讲过，Dubbo默认提供了一系列Wrapper类对扩展实现类进行了功能增强，当然这里也不例外，Dubbo使用了ProtocolListenerWrapper / ProtocolFilterWrapper等类对DubboProtocol进行了功能增强。所以这里经过一次次调用后才调用到DubboProtocol的refer方法，DubboProtocol的refer代码内容如下：

public <T> Invoker<T> refer(Class<T> serviceType , URL url) throws RpcException{
    // create rpc invoker.
    DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType , url ,getClients(url) , invokers);
    invokers.add(invoker);
    return invoker;
}

如上代码，首先getClients方法从（12）到（18）创建服务消费端的NettyClient对象用来连接服务提供者。另外可知refer方法内部返回了一个DubboInvoker，这个就是原生的invoker对象，服务方远程服务转换就是为了这个invoker。代码（12）则对这个invoker进行装饰，使用一系列filter形成了责任链，invoker被放到责任链的末尾，下面看看buildInvokerChain如何形成责任链，代码如下：

private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker , String key , String group){
    Invoker<T> last = invoker;
    // 获取所有激活的filter，然后使用链表方式形成责任链
    List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl() , key , group);
    if(filters.size() > 0){
        for(int i=filters.size()-1 ; i>=0 ; i--){
            final Filter filter = filters.get(i);
            final Invoker<T> next = last;
            last = new Invoker<T>(){
                public Class<T> getInterface(){
                    return invoker.getInterface();
                }
                public URL getUrl(){
                    return invoker.getUrl();
                }
                public boolean isAvailable(){
                    return invoker.isAvailable();
                }
                public Result invoke(Invocation invocation) throws RpcException{
                    return filter.invoke(next , invocation);    
                }
                public void destroy(){
                    invoker.destroy();
                }
                @Override
                public String toString(){
                    return invoker.toString();
                }
            };
        }
    }
    return last;
}

其中扩展接口Filter对应的实现类，如下所示：

其中MonitorFilter和监控中心进行交互，FutureFilter用来实现异步调用，GenericFilter用来实现泛化调用，ActiveLimitFilter用来控制消费端最大并发调用量，ExecuteLimitFilter用来控制服务提供方最大并发处理量等，当然你可以写自己的filter。由于是责任链，所以ProtocolFilterWrapper的refer返回的是责任链头部的filter到ProtocolListenerWrapper，而ProtocolListenerWrapper的refer方法内容如下：

public <T> Invoker<T> refer(Class<T> type , URL url) throws RpcException{
    ...
    return new ListenerInvokerWrapper<T>(protocol.refer(type,url) , Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(InvokerListener.class).getActivateExtension(url , Constants.INVOKER_LISTENER_KEY)));
}

ProtocolListenerWrapper的refer返回的是ListenerInvokerWrapper对象，所以代码（9）返回的也是url对应的ListenerInvokerWrapper对象，然后再回到时序图代码（8）的toInvokers方法，可知toInvokers返回的是使用InvokerDelegete对ListenerInvokerWrapper包裹的对象。到这里RegistryDirectory里就维护了所有服务者的invoker列表，消费端发送信息时就是根据集群容错和负载均衡算法从invoker列表里选择一个进行调用，当服务提供者挂了的时候，Zookeeper会通知更新这invoker列表。

到这里我们讲完了本节第一个时序图中的步骤（3），下面我们接着讲解步骤（4） 。

默认下步骤（4）会调用FailoverCluster的join方法，FailoverCluster的join方法代码如下：

public class FailoverCluster implements Cluster{
    private static final String NAME = "failover";
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException{
        return new FailoverClusterInvoker<T>(directory);
    }
}

这里把directory对象包裹到了FailoverClusterInvoker，这里需要注意下directory就是上面讲解的RegistryDirectory，其内部维护了所有服务提供者的invoker列表，而FailoverCluster就是集群容错策略。

其实Dubbo对cluster扩展接口实现类使用wrapper类MockClusterWrapper进行增强，这个从下图可以得到证明：

实际上的调用时序图如下图所示：

该时序图中（3）返回了FailbackClusterInvoker对象到（2），下面看看MockClusterWrapper的代码：

public class MockClusterWrapper implements Cluster{
    private Cluster cluster;
    public MockClusterWrapper(Cluster cluster){
        this.cluster = cluster;
    }
    public <T> Invoker<T> join(Directory<T> directory) throws RpcException{
        return new MockClusterInvoker<T>(directory , this.cluster.join(directory));
    }
}

可知MockClusterWrapper类把FailoverClusterInvoker包装成了MockClusterInvoker实例，所以整个调用链最终调用返回的是MockClusterInvoker对象。也就是本文第一个时序图步骤（4）返回的是MockClusterWrapper。然后执行代码（13）获取MockClusterInvoker的代理实现invoker到客户端接口的转换，这里默认调用的是JavassistProxyFactory的getProxy方法，代码如下：

public <T> T getProxy(Invoker<T> invoker , Class<?>[] interfaces){
    return (T)Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
}

其中InvokerInvocationHandler为具体拦截器。至此我们按照逆序的方式把服务消费端启动流程讲解完毕，下面在顺过来讲解一次远程调用过程。

Dubbo服务消费端一次调用流程原理分析

同理先上时序图：

由于服务消费端通过ReferenceConfig的get()方法返回的是一个代理类，并且方法拦截器为InvokerInvocationHandler，所以当调用了服务的方法后会被InvokerInvocationHandler拦截，执行如上时序图流程。

如上流程，首先步骤（1）（2）（3）调用了集群容错策略FailoverClusterInvoker，其内部首先根据设置的负载均衡策略选择一个invoker作为FailoverClusterInvoker具体的远程调用者，如果调用发生异常，则根据FailoverClusterInvoker的策略重新选择一个invoker进行调用。

FailoverClusterInvoker内每次调用具体invoker的invoke方法都会走到步骤（8）（9），然后步骤（10）（11）（12）是filter内创建的责任链，最后调用了原生的DubboInvoker，具体使用nettyclient与服务提供者进行交互。