Dubbo源码分析（六）服务引用的具体流程

 

前言

在前面的章节中，我们已经完成Dubbo服务暴露的流程分析。今天我们一起来看Dubbo怎么引用这些服务的。

关于服务引用，Dubbo有两种方式。一种是基于注册中心进行服务引用，一种是服务直连进行引用。服务直连主要用于测试联调阶段，生产环境不推荐使用。它的配置也比较简单，在消费者端指定服务url即可。

<dubbo:reference id="infoUserService"
    interface="com.viewscenes.netsupervisor.service.InfoUserService"
    url="dubbo://192.168.139.129:20880"/>

本文将重点分析通过注册中心方式，服务引用的过程。开始之前，我们再回顾一下消费者端项目整体的XML文件配置。

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
       xmlns="http://www.springframework.org/schema/beans"
       xsi:schemaLocation="http://www.springframework.org/schema/beans 
       http://www.springframework.org/schema/beans/spring-beans-4.3.xsd
       http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd">
    <!-- 消费者方应用信息，用于计算依赖关系 -->
    <dubbo:application name="dubbo_consumer"/>
    <!-- 用于配置连接注册中心相关信息 -->
    <dubbo:registry protocol="zookeeper" address="192.168.139.131:2181" client="zkclient" />
    <!-- 引用配置   用于创建一个远程接口服务代理 -->
    <dubbo:reference id="infoUserService"
        interface="com.viewscenes.netsupervisor.service.InfoUserService"/>
</beans>

一、引用的是什么？

在上述配置文件中，infoUserService只是Spring中的一个Bean。在代码中，我们通过getBean()来获取它，然后调用它的方法。

我们在RPC基本原理以及如何用Netty来实现RPC这篇文章中已经分析过，这里的bean其实是一个FactoryBean，通过它可以返回一个接口的代理对象，完成调用逻辑的处理。

在代码中，我们通过这样来获取这个Bean。

InfoUserService infoUserService = (InfoUserService) context.getBean("infoUserService");

这里的infoUserService就是一个代理对象，比如像proxy@2903这种。当然了，它必然会包含一个InvocationHandler，如下所示：

 

infoUserService代理对象

服务引用就是其实通过动态代理给接口创建代理对象并返回。当我们调用接口方法时，则调用到InvocationHandler相关方法，处理相关请求。

二、从哪里开始？

在上面的配置文件中，dubbo:reference对应的是ReferenceBean处理类。那么Spring在实例化这个Bean的时候，就调用到里面方法。我们先看看它的类结构

package com.alibaba.dubbo.config.spring;

public class ReferenceBean<T> extends ReferenceConfig<T> implements 
        FactoryBean, ApplicationContextAware, InitializingBean, DisposableBean {

    private transient ApplicationContext applicationContext;

    public ReferenceBean() {
        super();
    }
    public ReferenceBean(Reference reference) {
        super(reference);
    }
    //设置applicationContext
    public void setApplicationContext(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }
    //返回代理对象
    public Object getObject() throws Exception {
        return get();
    }
    //返回代理对象接口类型
    public Class<?> getObjectType() {
        return getInterfaceClass();
    }
    //是否为单例
    @Parameter(excluded = true)
    public boolean isSingleton() {
        return true;
    }
    //Bean初始化方法
    public void afterPropertiesSet() throws Exception {
        if(isInit()){
            getObject();
        }
    }
}

首先，映入眼帘的是FactoryBean接口。有了它，则证明当前bean是一个工厂Bean，所以我们重点关注它的getObject方法，当我们的服务被注入到其他类中时，就会调用到此方法。

其次，是InitializingBean接口。它是Bean的初始化方法，当Bean完成实例化之后被调用。在上述代码中，当init()条件判断成立后调用getObject。我们可以在配置文件中以这种方式来激活它：init="true"

以上就是服务引用的两个不同时机。再专业点来讲，一个是懒汉式，一个是饿汉式。Dubbo默认是懒汉式引用，需要时才会调用。

上述代码中，getObject方法会调用到父类的init方法。这个方法内容比较多，前面的部分是各种配置检查、赋值，然后就是创建代理对象返回。

private void init() {
    //避免重复初始化
    if (initialized) {
        return;
    }   
    //省略相关配置检查、赋值等代码... 
    
    //创建代理
    ref = createProxy(map);
}

上述代码中，重点是createProxy方法。它负责创建Invoker实例和创建代理对象。

也许我们还记得，在服务暴露的时候，会分为本地暴露和远程暴露。在这里，服务引用也是这样。Dubbo首先判断服务引用是本地引用还是远程引用，默认是远程引用。然后判断是否为直连服务，根据协议调用refer方法创建invoker对象 。最后创建服务代理对象并返回。

private T createProxy(Map<String, String> map) {
    URL tmpUrl = new URL("temp", "localhost", 0, map);
    final boolean isJvmRefer;
    //获取配置判断是否为本地引用
    if (isInjvm() == null) {
        if (url != null && url.length() > 0) {
            isJvmRefer = false;
        } else if (InjvmProtocol.getInjvmProtocol().isInjvmRefer(tmpUrl)) {
            // by default, reference local service if there is
            isJvmRefer = true;
        } else {
            isJvmRefer = false;
        }
    } else {
        isJvmRefer = isInjvm().booleanValue();
    }
    //本地引用
    if (isJvmRefer) {
        URL url = new URL(Constants.LOCAL_PROTOCOL, NetUtils.LOCALHOST, 0, interfaceClass.getName()).addParameters(map);
        invoker = refprotocol.refer(interfaceClass, url);
        if (logger.isInfoEnabled()) {
            logger.info("Using injvm service " + interfaceClass.getName());
        }
    //远程引用
    } else {
        //url不为空则代表是服务直连
        if (url != null && url.length() > 0) {
            //切割多个url
            String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
            if (us != null && us.length > 0) {
                for (String u : us) {
                    URL url = URL.valueOf(u);
                    if (url.getPath() == null || url.getPath().length() == 0) {
                        url = url.setPath(interfaceName);
                    }
                    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                        urls.add(url.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                    } else {
                        urls.add(ClusterUtils.mergeUrl(url, map));
                    }
                }
            }
        } else {
            //加载注册中心的url
            List<URL> us = loadRegistries(false);
            if (us != null && !us.isEmpty()) {
                for (URL u : us) {
                    URL monitorUrl = loadMonitor(u);
                    if (monitorUrl != null) {
                        map.put(Constants.MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                    }
                    urls.add(u.addParameterAndEncoded(Constants.REFER_KEY, StringUtils.toQueryString(map)));
                }
            }
        }
        //单个注册中心
        if (urls.size() == 1) {
            invoker = refprotocol.refer(interfaceClass, urls.get(0));
        //多个注册中心
        } else {
            List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
            URL registryURL = null;
            for (URL url : urls) {
                invokers.add(refprotocol.refer(interfaceClass, url));
                if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
                    registryURL = url; 
                }
            }
            if (registryURL != null) {
                // use AvailableCluster only when register's cluster is available
                URL u = registryURL.addParameter(Constants.CLUSTER_KEY, AvailableCluster.NAME);
                invoker = cluster.join(new StaticDirectory(u, invokers));
            } else { 
                invoker = cluster.join(new StaticDirectory(invokers));
            }
        }
    }
    //创建代理类
    return (T) proxyFactory.getProxy(invoker);
}

前面的整个流程我们可以分为两大部分：创建Invoker和创建代理。

三、创建Invoker

1、连接注册中心

我们以本文开头的XML配置文件为例，它是基于单个注册中心的远程服务引用。所以，我们重点分析invoker = refprotocol.refer(interfaceClass, urls.get(0));

我们需要清楚的认识到，urls.get(0)获取到的是注册中心的地址。内容如下：registry://192.168.139.131:2181/org.apache.dubbo.registry.RegistryService?...

所以，我们先来看RegistryProtocol.refer方法。它主要是获取注册中心，并执行服务引用。

public class RegistryProtocol implements Protocol {
    
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        //转换协议
        url = url.setProtocol(url.getParameter("registry", "dubbo")).removeParameter("registry");
        //通过zookeeper连接到注册中心
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }
        // group="a,b" or group="*"
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded("refer"));
        String group = qs.get(Constants.GROUP_KEY);
        if (group != null && group.length() > 0) {
            if ((Constants.COMMA_SPLIT_PATTERN.split(group)).length > 1
                    || "*".equals(group)) {
                return doRefer(getMergeableCluster(), registry, type, url);
            }
        }
        return doRefer(cluster, registry, type, url);
    }
}

我们接着看doRefer(cluster, registry, type, url);我们可以认为它主要做了3件事：

• 向注册中心注册服务消费者(本身)
• 向注册中心订阅服务提供者
• 通过Cluster创建Invoker对象

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
    // 创建 RegistryDirectory 对象，并设置注册中心
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
    //创建订阅 URL
    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, parameters.
        remove("register.ip"), 0, type.getName(), parameters);
    if (!Constants.ANY_VALUE.equals(url.getServiceInterface())
            && url.getParameter(Constants.REGISTER_KEY, true)) {
        //向注册中心注册服务消费者
        registry.register(subscribeUrl.addParameters("category", "consumers",
                "check", String.valueOf(false)));
    }
    // 向注册中心订阅服务提供者
    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY,
            Constants.PROVIDERS_CATEGORY
                    + "," + Constants.CONFIGURATORS_CATEGORY
                    + "," + Constants.ROUTERS_CATEGORY));

    //创建Invoker对象
    Invoker invoker = cluster.join(directory);
    ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
    return invoker;
}

如上代码， 它的重点在于directory.subscribe ，它通过订阅注册中心的服务提供者，调用相关协议的引用方法。比如DubboProtocol.refer，下面我们来看它是怎么调用的。

2、订阅服务提供者

我们通过directory.subscribe这句代码， 会定位到ZookeeperRegistry.doSubscribe方法。

public class ZookeeperRegistry extends FailbackRegistry {
    
    protected void doSubscribe(final URL url, final NotifyListener listener) {
        
        List<URL> urls = new ArrayList<URL>();
        //获取注册中心除了消费者信息之外的所有节点
        //比如configurators, routers, providers
        String[] paths = toCategoriesPath(url);
        for (String path :paths) {
            //获取节点监听器管理容器
            ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
            if (listeners == null) {
                zkListeners.putIfAbsent(url, 
                    new ConcurrentHashMap<NotifyListener, ChildListener>());
                listeners = zkListeners.get(url);
            }
            //获取节点监听器
            ChildListener zkListener = listeners.get(listener);
            //如果为空，则创建一个监听器
            if (zkListener == null) {
                listeners.putIfAbsent(listener, new ChildListener() {
                    @Override
                    public void childChanged(String parentPath, List<String> currentChilds) {
                        ZookeeperRegistry.this.notify(url, listener, 
                                toUrlsWithEmpty(url, parentPath, currentChilds));
                    }
                });
                zkListener = listeners.get(listener);
            }
            //创建节点
            zkClient.create(path, false);
            //订阅节点path 数据变化，通知到zkListener
            //即通知到ZookeeperRegistry.this.notify
            List<String> children = zkClient.addChildListener(path, zkListener);
            if (children != null) {
                urls.addAll(toUrlsWithEmpty(url, path, children));
            }
        }
        //通知
        notify(url, listener, urls);  
    }
}

如上代码，重点就是通过注册中心获取到服务提供者的url，调用notify方法。同时还订阅了服务提供者的节点数据变化，如果有变化也调用notify方法。
而在notify方法中，最终会根据url调用服务引用方法refer，再封装成InvokerDelegate对象。
invoker = new InvokerDelegate<T>(protocol.refer(serviceType, url), url, providerUrl);

这样不仅完成了本次的服务引用，而且当服务提供者上下线的时候，消费者也可以动态感知。

3、Listener和Filter

以DubboProtocol为例，它是Wrapper包装类的对象，关于这一点我们一定要牢记。

回忆服务暴露的过程，我们可以监听服务暴露的过程，在构建调用链的时候可以增加Filter。在这里，服务引用也是一样的。

• 服务引用监听

在InvokerListener 接口中，有两个方法。分别是服务引用后调用和服务销毁时调用。

@SPI
public interface InvokerListener {  
    //服务引用
    void referred(Invoker<?> invoker) throws RpcException;
    //服务销毁
    void destroyed(Invoker<?> invoker);
}

我们先自定义监听器，来实现接口的方法。

public class MyInvokerListener1 implements InvokerListener{

    public void referred(Invoker<?> invoker) throws RpcException {      
        System.out.println("服务引用111.."+invoker.getUrl());
    }
    public void destroyed(Invoker<?> invoker) {
        System.out.println("服务销毁111.."+invoker.getUrl());
    }
}

然后创建扩展点配置文件，文件名称为：
com.alibaba.dubbo.rpc.InvokerListener
内容为：
listener1=com.viewscenes.netsupervisor.listener.MyInvokerListener1

最后，修改XML配置文件
<dubbo:consumer listener="listener1"/>

• Filter过滤器

首先自定义过滤器

public class MyFilter implements Filter{

    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        System.out.println("调用之前-----------");
        Result result = invoker.invoke(invocation);
        System.out.println("调用之后-----------");
        return result;
    }
}

然后创建扩展点配置文件，文件名称为：
com.alibaba.dubbo.rpc.Filter
内容为：
myfilter1=com.viewscenes.netsupervisor.filter.MyFilter

最后，修改XML配置文件，说明自定义的过滤器在默认过滤器之前。
filter="myfilter1,default""/>

4、DubboProtocol

前面啰啰嗦嗦一大堆，终于来到正主。这里主要是获取通信客户端，拿到ExchangeClient，然后封装为DubboInvoker对象并返回。

public class DubboProtocol extends AbstractProtocol {
    
    public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
        optimizeSerialization(url);
        //获取通信客户端
        ExchangeClient[] client = getClients(url);
        //创建DubboInvoker对象
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, client, invokers);
        invokers.add(invoker);
        return invoker;
    }
}

这里主要分为两个步骤：

• 创建通信客户端

在getClients方法里面，主要判断是否为共享连接的客户端。

private ExchangeClient[] getClients(URL url) {
    // 是否共享连接
    boolean service_share_connect = false;
    // 获取连接数，默认为0，表示未配置
    int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
    // 如果未配置 connections，则共享连接
    if (connections == 0) {
        service_share_connect = true;
        connections = 1;
    }
    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;
}

如果未配置，Dubbo默认创建共享连接客户端。我们继续看getSharedClient方法。

private ExchangeClient getSharedClient(URL url) {
    //服务地址
    String key = url.getAddress();
    //查询缓存是否已经创建了客户端
    ReferenceCountExchangeClient client = referenceClientMap.get(key);
    if (client != null) {
        if (!client.isClosed()) {
            client.incrementAndGetCount();
            return client;
        } else {
            referenceClientMap.remove(key);
        }
    }
    synchronized (key.intern()) {
        //初始化客户端
        ExchangeClient exchangeClient = initClient(url);
        //封装成ReferenceCountExchangeClient对象，用于引用计数
        client = new ReferenceCountExchangeClient(exchangeClient, ghostClientMap);
        //放入缓存并返回
        referenceClientMap.put(key, client);
        ghostClientMap.remove(key);
        return client;
    }
}

如上代码，先从缓存中获取客户端，未命中则调用initClient方法初始化客户端。并将它封装成ReferenceCountExchangeClient 对象返回。在initClient方法中，设置客户端类型、编解码和心跳信息，并调用connect创建客户端。

private ExchangeClient initClient(URL url) {

    //客户端类型 默认为netty
    String str = url.getParameter("client", url.getParameter("server","netty"));
    //编解码
    url = url.addParameter("codes", "dubbo");
    //心跳
    url = url.addParameterIfAbsent("heartbeat","60000");

    ExchangeClient client;
    try {
        if (url.getParameter("lazy", 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;
}

如上代码，connect会调用到HeaderExchangeClient.connect方法。

public class HeaderExchanger implements Exchanger {

    public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
        HeaderExchangeHandler headerExchangeHandler = new HeaderExchangeHandler(handler);
        DecodeHandler decodeHandler = new DecodeHandler(headerExchangeHandler);
        Client client = Transporters.connect(url, decodeHandler);
        return new HeaderExchangeClient(client,true);
    }
}

我们重点来看Transporters.connect，它会调用到NettyClient的构造方法。最终调用doOpen和doConnect方法，连接通信服务器。

public class NettyClient extends AbstractClient {

    protected void doOpen() throws Throwable {
        NettyHelper.setNettyLoggerFactory();
        bootstrap = new ClientBootstrap(channelFactory);
        bootstrap.setOption("keepAlive", true);
        bootstrap.setOption("tcpNoDelay", true);
        bootstrap.setOption("connectTimeoutMillis", getTimeout());
        
        //处理器
        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
            @Override
            public ChannelPipeline getPipeline() {
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), 
                        getUrl(), NettyClient.this);
                ChannelPipeline pipeline = Channels.pipeline();
                pipeline.addLast("decoder", adapter.getDecoder());
                pipeline.addLast("encoder", adapter.getEncoder());
                pipeline.addLast("handler", nettyHandler);
                return pipeline;
            }
        });
    }
    
    
    protected void doConnect() throws Throwable {
        long start = System.currentTimeMillis();
        //获取服务器地址，并连接
        ChannelFuture future = bootstrap.connect(getConnectAddress());
        try {
            boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
            //连接成功后获取Channel
            if (ret && future.isSuccess()) {
                Channel newChannel = future.getChannel();
                newChannel.setInterestOps(Channel.OP_READ_WRITE);
                try {
                    // Close old channel
                    Channel oldChannel = NettyClient.this.channel; // copy reference
                    if (oldChannel != null) {
                        try {                  
                            oldChannel.close();
                        } finally {
                            NettyChannel.removeChannelIfDisconnected(oldChannel);
                        }
                    }
                } finally {
                    if (NettyClient.this.isClosed()) {
                        try {                           
                            newChannel.close();
                        } finally {
                            NettyClient.this.channel = null;
                            NettyChannel.removeChannelIfDisconnected(newChannel);
                        }
                    } else {
                        //设置当前Channel
                        NettyClient.this.channel = newChannel;
                    }
                }
            }
        }
    }
}

以上大部分都是Netty的代码，连接服务器成功后获取到Channel保存。 我们重点看NettyHandler，它是事件处理器。这里的messageReceived负责接收处理返回的消息。接收到消息后，通过线程池拿到线程去执行后续的处理。不同的是，这里的线程池类型是CachedThreadPool ，即缓存型线程池。

这里的流程处理完毕，我们再把目光回到DubboInvoker，它通过一个构造方法返回Invoker对象。

public DubboInvoker(Class<T> serviceType, URL url, 
            ExchangeClient[] clients, Set<Invoker<?>> invokers) {
    this.clients = clients;
    this.version = url.getParameter(Constants.VERSION_KEY, "0.0.0");
    this.invokers = invokers;
}

至此，Invoker对象已经创建完毕。我们回到开头的createProxy方法。最后一步是为这个Invoker对象创建代理。

五、创建代理

在Dubbo，默认使用Javassist来创建代理。

public class JavassistProxyFactory extends AbstractProxyFactory {

    public <T> T getProxy(Invoker<T> invoker, Class<?>[] interfaces) {
        //调用处理程序
        InvocationHandler invocationHandler = new InvokerInvocationHandler(invoker)
        //创建代理
        return (T) Proxy.getProxy(interfaces).newInstance(invocationHandler);
    }
}

创建的过程很简单，我们主要来看InvocationHandler 。既然是代理对象，我们每次调用接口方法的时候，会调用到这个对象的invoke方法。

public class InvokerInvocationHandler implements InvocationHandler {

    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String methodName = method.getName();
        Class<?>[] parameterTypes = method.getParameterTypes();
        if (method.getDeclaringClass() == Object.class) {
            return method.invoke(invoker, args);
        }
        if ("toString".equals(methodName) && parameterTypes.length == 0) {
            return invoker.toString();
        }
        if ("hashCode".equals(methodName) && parameterTypes.length == 0) {
            return invoker.hashCode();
        }
        if ("equals".equals(methodName) && parameterTypes.length == 1) {
            return invoker.equals(args[0]);
        }
        //把请求的方法名和参数等封装成RpcInvocation对象
        //调用invoke
        return invoker.invoke(new RpcInvocation(method, args)).recreate();
    }
}

至此，服务引用的大体流程我们已经分析完了。我们再来总结一下它的过程：

• 通过FactoryBean触发服务引用方法
• 连接注册中心，获取服务提供者列表，创建invoker
• 订阅服务提供者节点，数据变更时获得通知
• 连接服务器，获得客户端
• 创建代理