Dubbo中消费者初始化的过程解析

Posted 张伯雨

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Dubbo中消费者初始化的过程解析相关的知识，希望对你有一定的参考价值。

首先还是Spring碰到dubbo的标签之后，会使用parseCustomElement解析dubbo标签，使用的解析器是dubbo的DubboBeanDefinitionParser，解析完成之后返回BeanDefinition给Spring管理。

服务消费者端对应的是ReferenceBean，实现了ApplicationContextAware接口，Spring会在Bean的实例化那一步回调setApplicationContext方法。也实现了InitializingBean接口，接着会回调afterPropertySet方法。还实现了FactoryBean接口，实现FactoryBean可以在后期获取bean的时候做一些操作，dubbo在这个时候做初始化。另外ReferenceBean还实现了DisposableBean，会在bean销毁的时候调用destory方法。

消费者的初始化是在ReferenceBean的init方法中执行，分为两种情况：

  • reference标签中没有配置init属性，此时是延迟初始化的，也就是只有等到bean引用被注入到其他Bean中，或者调用getBean获取这个Bean的时候，才会初始化。比如在这里的例子里reference没有配置init属性，只有等到HelloService helloService = (HelloService) applicationContext.getBean("helloService");这句getBean的时候，才会开始调用init方法进行初始化。
  • 另外一种情况是立即初始化，即是如果reference标签中init属性配置为true，会立即进行初始化（也就是上面说到的实现了FactoryBean接口）。

初始化开始

这里以没有配置init的reference为例，只要不注入bean或者不调用getBean获取bean的时候，就不会被初始化。HelloService helloService = (HelloService) applicationContext.getBean("helloService");

另外在ReferenceBean这个类在Spring中初始化的时候，有几个静态变量会被初始化：

1
2
3
4
5
private static final Protocol refprotocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();

private static final Cluster cluster = ExtensionLoader.getExtensionLoader(Cluster.class).getAdaptiveExtension();

private static final ProxyFactory proxyFactory = ExtensionLoader.getExtensionLoader(ProxyFactory.class).getAdaptiveExtension();

这几个变量的初始化是根据dubbo的SPI扩展机制动态生成的代码：

refprotocol：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class Protocol$Adpative implements com.alibaba.dubbo.rpc.Protocol {
  public com.alibaba.dubbo.rpc.Invoker refer(java.lang.Class arg0, com.alibaba.dubbo.common.URL arg1) throws java.lang.Class {
    if (arg1 == null) throw new IllegalArgumentException("url == null");

    com.alibaba.dubbo.common.URL url = arg1;
    String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );

    if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
    
    com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
    
    return extension.refer(arg0, arg1);
  }
  
  public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.Invoker {
    if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
    
    if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
    
    String extName = ( url.getProtocol() == null ? "dubbo" : url.getProtocol() );
    
    if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url(" + url.toString() + ") use keys([protocol])");
    
    com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);
    
    return extension.export(arg0);
  }
  
  public void destroy() {
  	throw new UnsupportedOperationException("method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
  }
  
  public int getDefaultPort() {
  	throw new UnsupportedOperationException("method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");
  }
}

cluster：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class Cluster$Adpative implements com.alibaba.dubbo.rpc.cluster.Cluster {

  public com.alibaba.dubbo.rpc.Invoker join(com.alibaba.dubbo.rpc.cluster.Directory arg0) throws com.alibaba.dubbo.rpc.cluster.Directory {
    if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument == null");
    
    if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.cluster.Directory argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
    
    String extName = url.getParameter("cluster", "failover");
    if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.cluster.Cluster) name from url(" + url.toString() + ") use keys([cluster])");
    
    com.alibaba.dubbo.rpc.cluster.Cluster extension = (com.alibaba.dubbo.rpc.cluster.Cluster)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.cluster.Cluster.class).getExtension(extName);
    
    return extension.join(arg0);
  }
}

proxyFactory：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class ProxyFactory$Adpative implements com.alibaba.dubbo.rpc.ProxyFactory {

  public java.lang.Object getProxy(com.alibaba.dubbo.rpc.Invoker arg0) throws com.alibaba.dubbo.rpc.Invoker {
    if (arg0 == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");
    
    if (arg0.getUrl() == null) throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");com.alibaba.dubbo.common.URL url = arg0.getUrl();
    
    String extName = url.getParameter("proxy", "javassist");
    if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
    
    com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
    
    return extension.getProxy(arg0);
  }
  
  public com.alibaba.dubbo.rpc.Invoker getInvoker(java.lang.Object arg0, java.lang.Class arg1, com.alibaba.dubbo.common.URL arg2) throws java.lang.Object {
    if (arg2 == null) throw new IllegalArgumentException("url == null");
    
    com.alibaba.dubbo.common.URL url = arg2;
    String extName = url.getParameter("proxy", "javassist");
    
    if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.ProxyFactory) name from url(" + url.toString() + ") use keys([proxy])");
    
    com.alibaba.dubbo.rpc.ProxyFactory extension = (com.alibaba.dubbo.rpc.ProxyFactory)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.rpc.ProxyFactory.class).getExtension(extName);
    
    return extension.getInvoker(arg0, arg1, arg2);
  }
}

初始化入口

初始化的入口在ReferenceConfig的get()方法：

1
2
3
4
5
6
7
8
9
public synchronized T get() {
  if (destroyed){
  	throw new IllegalStateException("Already destroyed!");
  }
  if (ref == null) {
  	init();
  }
  return ref;
}

init()方法会先检查初始化所有的配置信息，然后调用ref = createProxy(map);创建代理，消费者最终得到的是服务的代理。初始化主要做的事情就是引用对应的远程服务，大概的步骤：

  • 监听注册中心
  • 连接服务提供者端进行服务引用
  • 创建服务代理并返回

文档上关于Zookeeper作为注册中心时，服务消费者启动时要做的事情有：

订阅/dubbo/com.foo.BarService/providers目录下的提供者URL地址。
并向/dubbo/com.foo.BarService/consumers目录下写入自己的URL地址。

createProxy方法

  • 引用远程服务
  • 创建代理

init()中createProxy方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private T createProxy(Map<String, String> map) {
	//先判断是否是本地服务引用injvm
    //判断是否是点对点直连
    //判断是否是通过注册中心连接
    //然后是服务的引用
    //这里url为
    //registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?
    //application=dubbo-consumer&dubbo=2.5.3&pid=12272&
    //refer=application%3Ddubbo-consumer%26dubbo%3D2.5.3%26
    //interface%3Ddubbo.common.hello.service.HelloService%26
    //methods%3DsayHello%26pid%3D12272%26side%3D
    //consumer%26timeout%3D100000%26timestamp%3D1489318676447&
    //registry=zookeeper&timestamp=1489318676641
    //引用远程服务由Protocol的实现来处理
    refprotocol.refer(interfaceClass, url);
    //最后返回服务代理
     return (T) proxyFactory.getProxy(invoker);
}

这里refprotocol是上面生成的代码，会根据协议不同选择不同的Protocol协议。

引用远程服务

对于服务引用refprotocol.refer(interfaceClass, url)会首先进入ProtocolListenerWrapper的refer方法，然后在进入ProtocolFilterWrapper的refer方法，然后再进入RegistryProtocol的refer方法，这里的url协议是registry，所以上面两个Wrapper中不做处理，直接进入了RegistryProtocol，看下RegistryProtocol中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
	//这里获得的url是
    //zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?
    //application=dubbo-consumer&dubbo=2.5.3&pid=12272&
    //refer=application%3Ddubbo-consumer%26dubbo%3D2.5.3%26
    //interface%3Ddubbo.common.hello.service.HelloService%26
    //methods%3DsayHello%26pid%3D12272%26side%3D
    //consumer%26timeout%3D100000%26
    //timestamp%3D1489318676447&timestamp=1489318676641
    url = url.setProtocol(url.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_REGISTRY)).removeParameter(Constants.REGISTRY_KEY);
    //根据url获取Registry对象
    //先连接注册中心，把消费者注册到注册中心
    Registry registry = registryFactory.getRegistry(url);
    //判断引用是否是注册中心RegistryService，如果是直接返回刚得到的注册中心服务
    if (RegistryService.class.equals(type)) {
        return proxyFactory.getInvoker((T) registry, type, url);
    }
	//以下是普通服务，需要进入注册中心和集群下面的逻辑
    // group="a,b" or group="*"
    //获取ref的各种属性
    Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(Constants.REFER_KEY));
    //获取分组属性
    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 );
        }
    }
    //选择配置的集群策略（cluster="failback"）或者默认策略
    return doRefer(cluster, registry, type, url);
}
获取注册中心

连接注册中心Registry registry = registryFactory.getRegistry(url);首先会到AbstractRegistryFactory的getRegistry方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
public Registry getRegistry(URL url) {
	//这里url是
    //zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?
    //application=dubbo-consumer&dubbo=2.5.3&
    //interface=com.alibaba.dubbo.registry.RegistryService&
    //pid=12272&timestamp=1489318676641
    url = url.setPath(RegistryService.class.getName())
            .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName())
            .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY);
    //这里key是
    //zookeeper://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService
    String key = url.toServiceString();
    // 锁定注册中心获取过程，保证注册中心单一实例
    LOCK.lock();
    try {
        Registry registry = REGISTRIES.get(key);
        if (registry != null) {
            return registry;
        }
        //这里用的是ZookeeperRegistryFactory
        //返回的Registry中封装了已经连接到Zookeeper的zkClient实例
        registry = createRegistry(url);
        if (registry == null) {
            throw new IllegalStateException("Can not create registry " + url);
        }
        //放到缓存中
        REGISTRIES.put(key, registry);
        return registry;
    } finally {
        // 释放锁
        LOCK.unlock();
    }
}

ZookeeperRegistryFactory的createRegistry方法：

1
2
3
4
5
public Registry createRegistry(URL url) {
	//直接返回一个新的ZookeeperRegistry实例
    //这里的zookeeperTransporter代码在下面，动态生成的适配类
    return new ZookeeperRegistry(url, zookeeperTransporter);
}

zookeeperTransporter代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
package com.alibaba.dubbo.remoting.zookeeper;
import com.alibaba.dubbo.common.extension.ExtensionLoader;
public class ZookeeperTransporter$Adpative implements com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter {
    public com.alibaba.dubbo.remoting.zookeeper.ZookeeperClient connect(com.alibaba.dubbo.common.URL arg0) {
        if (arg0 == null) throw new IllegalArgumentException("url == null");
        
        com.alibaba.dubbo.common.URL url = arg0;
        String extName = url.getParameter("client", url.getParameter("transporter", "zkclient"));
        
        if(extName == null) throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter) name from url(" + url.toString() + ") use keys([client, transporter])");
        
        com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter extension = (com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter)ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.remoting.zookeeper.ZookeeperTransporter.class).getExtension(extName);
        
        return extension.connect(arg0);
    }
}

上面代码中可以看到，如果我们没有指定Zookeeper的client属性，默认使用zkClient，所以上面的zookeeperTransporter是ZkclientZookeeperTransporter。

继续看new ZookeeperRegistry(url, zookeeperTransporter);：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) {
	//这里会先经过AbstractRegistry的处理，然后经过FailbackRegistry的处理（解释在下面）
    super(url);
    if (url.isAnyHost()) {
        throw new IllegalStateException("registry address == null");
    }
    //服务分组，默认dubbo
    String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT);
    if (! group.startsWith(Constants.PATH_SEPARATOR)) {
        group = Constants.PATH_SEPARATOR + group;
    }
    //注册中心的节点
    this.root = group;
    //ZkclientZookeeperTransporter的connect方法
    //直接返回一个ZkclientZookeeperClient实例
    //具体的步骤是，new一个ZkClient实例，然后订阅了一个状态变化的监听器
    zkClient = zookeeperTransporter.connect(url);
    //添加一个状态改变的监听器
    zkClient.addStateListener(new StateListener() {
        public void stateChanged(int state) {
            if (state == RECONNECTED) {
                try {
                    recover();
                } catch (Exception e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    });
}

AbstractRegistry的处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
public AbstractRegistry(URL url) {
	//设置registryUrl
    setUrl(url);
    // 启动文件保存定时器
    syncSaveFile = url.getParameter(Constants.REGISTRY_FILESAVE_SYNC_KEY, false);
    //会先去用户主目录下的.dubbo目录下加载缓存注册中心的缓存文件比如：dubbo-registry-127.0.0.1.cache
    String filename = url.getParameter(Constants.FILE_KEY, System.getProperty("user.home") + "/.dubbo/dubbo-registry-" + url.getHost() + ".cache");
    File file = null;
    if (ConfigUtils.isNotEmpty(filename)) {
        file = new File(filename);
        if(! file.exists() && file.getParentFile() != null && ! file.getParentFile().exists()){
            if(! file.getParentFile().mkdirs()){
                throw new IllegalArgumentException("Invalid registry store file " + file + ", cause: Failed to create directory " + file.getParentFile() + "!");
            }
        }
    }
    this.file = file;
    //缓存文件存在的话就把文件读进内存中
    loadProperties();
    //先获取backup url
    //然后通知订阅
    notify(url.getBackupUrls());
}

notify方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
protected void notify(List<URL> urls) {
    if(urls == null || urls.isEmpty()) return;
	//getSubscribed()方法获取订阅者列表
    for (Map.Entry<URL, Set<NotifyListener>> entry : getSubscribed().entrySet()) {
        URL url = entry.getKey();

        if(! UrlUtils.isMatch(url, urls.get(0))) {
            continue;
        }

        Set<NotifyListener> listeners = entry.getValue();
        if (listeners != null) {
            for (NotifyListener listener : listeners) {
                try {
                	//通知每个监听器
                    notify(url, listener, filterEmpty(url, urls));
                } catch (Throwable t) { }
            }
        }
    }
}

notify(url, listener, filterEmpty(url, urls));代码：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    Map<String, List<URL>> result = new HashMap<String, List<URL>>();
    for (URL u : urls) {
        if (UrlUtils.isMatch(url, u)) {
        	//分类
            String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            List<URL> categoryList = result.get(category);
            if (categoryList == null) {
                categoryList = new ArrayList<URL>();
                result.put(category, categoryList);
            }
            categoryList.add(u);
        }
    }
    if (result.size() == 0) {
        return;
    }
    Map<String, List<URL>> categoryNotified = notified.get(url);
    if (categoryNotified == null) {
        notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
        categoryNotified = notified.get(url);
    }
    for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
        String category = entry.getKey();
        List<URL> categoryList = entry.getValue();
        categoryNotified.put(category, categoryList);
        saveProperties(url);
        //通知
        listener.notify(categoryList);
    }
}

AbstractRegistry构造完，接着是FailbackRegistry的处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public FailbackRegistry(URL url) {
    super(url);
    int retryPeriod = url.getParameter(Constants.REGISTRY_RETRY_PERIOD_KEY, Constants.DEFAULT_REGISTRY_RETRY_PERIOD);
    //启动失败重试定时器
    this.retryFuture = retryExecutor.scheduleWithFixedDelay(new Runnable() {
        public void run() {
            // 检测并连接注册中心
            try {
            	//重试方法由每个具体子类实现
                //获取到注册失败的，然后尝试注册
                retry();
            } catch (Throwable t) { // 防御性容错 }
        }
    }, retryPeriod, retryPeriod, TimeUnit.MILLISECONDS);
}

这里会启动一个新的定时线程，主要是有连接失败的话，会进行重试连接retry();，启动完之后返回ZookeeperRegistry中继续处理。接下来的处理在代码注释中，不再详细写，看下一步服务的引用。

引用远程服务

继续看ref方法中最后一步，服务的引用，返回的是一个Invoker，return doRefer(cluster, registry, type, url)；

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
	//初始化Directory
    //组装Directory，可以看成一个消费端的List，可以随着注册中心的消息推送而动态的变化服务的Invoker
    //封装了所有服务真正引用逻辑，覆盖配置，路由规则等逻辑
    //初始化时只需要向注册中心发起订阅请求，其他逻辑均是异步处理，包括服务的引用等
    //缓存接口所有的提供者端Invoker以及注册中心接口相关的配置等
    RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
    directory.setRegistry(registry);
    directory.setProtocol(protocol);
    //此处的subscribeUrl为
    //consumer://192.168.1.100/dubbo.common.hello.service.HelloService?
    //application=dubbo-consumer&dubbo=2.5.3&
    //interface=dubbo.common.hello.service.HelloService&
    //methods=sayHello&pid=16409&
    //side=consumer&timeout=100000&timestamp=1489322133987
    URL subscribeUrl = new URL(Constants.CONSUMER_PROTOCOL, NetUtils.getLocalHost(), 0, type.getName(), directory.getUrl().getParameters());
    if (! Constants.ANY_VALUE.equals(url.getServiceInterface())
            && url.getParameter(Constants.REGISTER_KEY, true)) {
            //到注册中心注册服务
            //此处regist是上面一步获得的registry，即是ZookeeperRegistry，包含zkClient的实例
            //会先经过AbstractRegistry的处理，然后经过FailbackRegistry的处理（解析在下面）
        registry.register(subscribeUrl.addParameters(Constants.CATEGORY_KEY, Constants.CONSUMERS_CATEGORY,
                Constants.CHECK_KEY, String.valueOf(false)));
    }
    //订阅服务
    //有服务提供的时候，注册中心会推送服务消息给消费者，消费者再进行服务的引用。
    directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
            Constants.PROVIDERS_CATEGORY 
            + "," + Constants.CONFIGURATORS_CATEGORY 
            + "," + Constants.ROUTERS_CATEGORY));
    //服务的引用与变更全部由Directory异步完成
    //集群策略会将Directory伪装成一个Invoker返回
    //合并所有相同的invoker
    return cluster.join(directory);
}

注册中心接收到消费者发送的订阅请求后，会根据提供者注册服务的列表，推送服务消息给消费者。消费者端接收到注册中心发来的提供者列表后，进行服务的引用。触发Directory监听器的可以是订阅请求，覆盖策略消息，路由策略消息。

AbstractRegistry的register方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
public void register(URL url) {
	//此时url是
    //consumer://192.168.1.100/dubbo.common.hello.service.HelloService?
    //application=dubbo-consumer&
    //category=consumers&check=false&dubbo=2.5.3&
    //interface=dubbo.common.hello.service.HelloService&methods=sayHello
    //&pid=16409&side=consumer&timeout=100000&timestamp=1489322133987
    if (url == null) {
        throw new IllegalArgumentException("register url == null");
    }
    if (logger.isInfoEnabled()){
        logger.info("Register: " + url);
    }
    registered.add(url);
}

上面只是把url添加到registered这个set中。

接着看FailbackRegistry的register方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
public void register(URL url) {
    super.register(url);
    failedRegistered.remove(url);
    failedUnregistered.remove(url);
    try {
        // 向服务器端发送注册请求
        //这里调用的是ZookeeperRegistry中的doRegister方法
        doRegister(url);
    } catch (Exception e) {
        Throwable t = e;

        // 如果开启了启动时检测，则直接抛出异常
        boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                && url.getParameter(Constants.CHECK_KEY, true)
                && ! Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
        boolean skipFailback = t instanceof SkipFailbackWrapperException;
        if (check || skipFailback) {
            if(skipFailback) {
                t = t.getCause();
            }
            throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
        } else {
            logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
        }

        // 将失败的注册请求记录到失败列表，定时重试
        failedRegistered.add(url);
    }
}

接着看下doRegister(url);方法，向服务器端发送注册请求，在ZookeeperRegistry中：

1
2
3
4
5
6
7
8
protected void doRegister(URL url) {
    try {
    	//直接调用create，在AbstractZookeeperClient类中
        zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
    } catch (Throwable e) {
        throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

zkClient.create()方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
//path为
///dubbo/dubbo.common.hello.service.HelloService/consumers/
//consumer%3A%2F%2F192.168.1.100%2F
//dubbo.common.hello.service.HelloService%3Fapplication%3D
//dubbo-consumer%26category%3Dconsumers%26check%3Dfalse%26
//dubbo%3D2.5.3%26interface%3D
//dubbo.common.hello.service.HelloService%26
//methods%3DsayHello%26pid%3D28819%26
//side%3Dconsumer%26timeout%3D100000%26timestamp%3D1489332839677
public void create(String path, boolean ephemeral) {
    int i = path.lastIndexOf(‘/‘);
    if (i > 0) {
        create(path.substring(0, i), false);
    }
    //循环完得到的path为/dubbo
    //dynamic=false 表示该数据为持久数据，当注册方退出时，数据依然保存在注册中心
    if (ephemeral) {
    	//创建临时的节点
        createEphemeral(path);
    } else {
    	//创建持久的节点，/dubbo/dubbo.common.hello.service.HelloService/consumers/
        //consumer%3A%2F%2F192.168.110.197%2F
        //dubbo.common.hello.service.HelloService%3Fapplication%3Ddubbo-consumer%26
        //category%3Dconsumers%26check%3Dfalse%26
        //dubbo%3D2.5.3%26interface%3D
        //dubbo.common.hello.service.HelloService%26
        //methods%3DsayHello%26pid%3D6370%26side%3D
        //consumer%26timeout%3D100000%26timestamp%3D1489367959659
        createPersistent(path);
    }
}

经过上面create之后，Zookeeper中就存在了消费者需要订阅的服务的节点：

1
2
3
4
5
6
7
8
9
10
11
12
13
/dubbo
	/dubbo.common.hello.service.HelloService
    	/consumers
        	/http://0.0.0.0:4550/?path=dubbo%2F
            dubbo.common.hello.service.HelloService%2F
            consumers%2Fconsumer%253A%252F%252F192.168.110.197%252F
            dubbo.common.hello.service.HelloService%253F
            application%253Ddubbo-consumer%2526category%253D
            consumers%2526check%253Dfalse%2526
            dubbo%253D2.5.3%2526interface%253D
            dubbo.common.hello.service.HelloService%2526
            methods%253DsayHello%2526pid%253D22392%2526side%253D
            consumer%2526timeout%253D100000%2526timestamp%253D1490063394184

消费者自己注册到注册中心之后，接着是订阅服务提供者，directory.subscribe()：

1
2
3
4
5
6
public void subscribe(URL url) {
	//设置消费者url
    setConsumerUrl(url);
    //这里的registry是ZookeeperRegistry
    registry.subscribe(url, this);
}

看下registry.subscribe(url, this);，这里registry是ZookeeperRegistry，会先经过AbstractRegistry的处理，然后是FailbackRegistry的处理。

在AbstractRegistry中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
//此时url为consumer://192.168.1.100/dubbo.common.hello.service.HelloService?application=dubbo-consumer&
//category=providers,configurators,routers&dubbo=2.5.3&interface=dubbo.common.hello.service.HelloService&methods=
//sayHello&pid=28819&side=consumer&timeout=100000&timestamp=1489332839677
public void subscribe(URL url, NotifyListener listener) {
	//先根据url获取已注册的监听器
    Set<NotifyListener> listeners = subscribed.get(url);
    //没有监听器，就创建，并添加进去
    if (listeners == null) {
        subscribed.putIfAbsent(url, new ConcurrentHashSet<NotifyListener>());
        listeners = subscribed.get(url);
    }
    //有监听器，直接把当前RegistryDirectory添加进去
    listeners.add(listener);
}

然后是FailbackRegistry中：

1
2
3
4
5
6
7
8
public void subscribe(URL url, NotifyListener listener) {
    super.subscribe(url, listener);
    removeFailedSubscribed(url, listener);
    try {
        // 向服务器端发送订阅请求
        doSubscribe(url, listener);
    } catch (Exception e) {...}
}

继续看doSubscribe(url, listener);向服务端发送订阅请求，在ZookeeperRegistry中：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
protected void doSubscribe(final URL url, final NotifyListener listener) {
    try {
        if (Constants.ANY_VALUE.equals(url.getServiceInterface())) {... } else {
            List<URL> urls = new ArrayList<URL>();
            for (String path : toCategoriesPath(url)) {
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
                if (listeners == null) {
                    zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
                    listeners = zkListeners.get(url);
                }
                //将zkClient的事件IZkChildListener转换到registry事件NotifyListener
                ChildListener zkListener = listeners.get(listener);
                if (zkListener == null) {
                    listeners.putIfAbsent(listener, new ChildListener() {
                        public void childChanged(String parentPath, List<String> currentChilds) {
                            ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds));
                        }
                    });
                    zkListener = listeners.get(listener);
                }
                //创建三个节点
                // /dubbo/dubbo.common.hello.service.HelloService/providers/
                // /dubbo/dubbo.common.hello.service.HelloService/configurators/
                // /dubbo/dubbo.common.hello.service.HelloService/routers/
                //上面三个路径会被消费者端监听，当提供者，配置，路由发生变化之后，
                //注册中心会通知消费者刷新本地缓存。
                zkClient.create(path, false);
                List<String> children = zkClient.addChildListener(path, zkListener);
                if (children != null) {
                    urls.addAll(toUrlsWithEmpty(url, path, children));
                }
            }
            notify(url, listener, urls);
        }
    } catch (Throwable e) {
        throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    }
}

服务订阅完成之后，接着就是notify(url, listener, urls);：

会先经过FailbackRegistry将失败的通知请求记录到失败列表，定时重试。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    try {
        doNotify(url, listener, urls);
    } catch (Exception t) {
        // 将失败的通知请求记录到失败列表，定时重试
        Map<NotifyListener, List<URL>> listeners = failedNotified.get(url);
        if (listeners == null) {
            failedNotified.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, List<URL>>());
            listeners = failedNotified.get(url);
        }
        listeners.put(listener, urls);
        logger.error("Failed to notify for subscribe " + url + ", waiting for retry, cause: " + t.getMessage(), t);
    }
}

doNotify(url, listener, urls);：

1
2
3
4
protected void doNotify(URL url, NotifyListener listener, List<URL> urls) {
	//父类实现
    super.notify(url, listener, urls);
}

AbstractRegistry中的doNotify实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
protected void notify(URL url, NotifyListener listener, List<URL> urls) {
    Map<String, List<URL>> result = new HashMap<String, List<URL>>();
    for (URL u : urls) {
        if (UrlUtils.isMatch(url, u)) {
        	//不同类型的数据分开通知，providers，consumers，routers，overrides
            //允许只通知其中一种类型，但该类型的数据必须是全量的，不是增量的。
            String category = u.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
            List<URL> categoryList = result.get(category);
            if (categoryList == null) {
                categoryList = new ArrayList<URL>();
                result.put(category, categoryList);
            }
            categoryList.add(u);
        }
    }
    if (result.size() == 0) {
        return;
    }
    Map<String, List<URL>> categoryNotified = notified.get(url);
    if (categoryNotified == null) {
        notified.putIfAbsent(url, new ConcurrentHashMap<String, List<URL>>());
        categoryNotified = notified.get(url);
    }
    //对这里得到的providers，configurators，routers分别进行通知
    for (Map.Entry<String, List<URL>> entry : result.entrySet()) {
        String category = entry.getKey();
        List<URL> categoryList = entry.getValue();
        categoryNotified.put(category, categoryList);
        saveProperties(url);
        //这里的listener是RegistryDirectory
        listener.notify(categoryList);
    }
}

到RegistryDirectory中查看notify方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
public synchronized void notify(List<URL> urls) {
    List<URL> invokerUrls = new ArrayList<URL>();
    List<URL> routerUrls = new ArrayList<URL>();
    List<URL> configuratorUrls = new ArrayList<URL>();
    for (URL url : urls) {
        String protocol = url.getProtocol();
        String category = url.getParameter(Constants.CATEGORY_KEY, Constants.DEFAULT_CATEGORY);
        if (Constants.ROUTERS_CATEGORY.equals(category) 
                || Constants.ROUTE_PROTOCOL.equals(protocol)) {
            routerUrls.add(url);
        } else if (Constants.CONFIGURATORS_CATEGORY.equals(category) 
                || Constants.OVERRIDE_PROTOCOL.equals(protocol)) {
            configuratorUrls.add(url);
        } else if (Constants.PROVIDERS_CATEGORY.equals(category)) {
            invokerUrls.add(url);
        } else {
            logger.warn("Unsupported category " + category + " in notified url: " + url + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost());
        }
    }
    // configurators 更新缓存的服务提供方配置
    if (configuratorUrls != null && configuratorUrls.size() >0 ){
        this.configurators = toConfigurators(configuratorUrls);
    }
    // routers//更新缓存的路由规则配置
    if (routerUrls != null && routerUrls.size() >0 ){
        List<Router> routers = toRouters(routerUrls);
        if(routers != null){ // null - do nothing
            setRouters(routers);
        }
    }
    List<Configurator> localConfigurators = this.configurators; // local reference
    // 合并override参数
    this.overrideDirectoryUrl = directoryUrl;
    if (localConfigurators != null && localConfigurators.size() > 0) {
        for (Configurator configurator : localConfigurators) {
            this.overrideDirectoryUrl = configurator.configure(overrideDirectoryUrl);
        }
    }
    // providers
    //重建invoker实例
    refreshInvoker(invokerUrls);
}

refreshInvoker(invokerUrls);：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
/**
 * 根据invokerURL列表转换为invoker列表。转换规则如下：
 * 1.如果url已经被转换为invoker，则不在重新引用，直接从缓存中获取，注意如果url中任何一个参数变更也会重新引用
 * 2.如果传入的invoker列表不为空，则表示最新的invoker列表
 * 3.如果传入的invokerUrl列表是空，则表示只是下发的override规则或route规则，需要重新交叉对比，决定是否需要重新引用。
 * @param invokerUrls 传入的参数不能为null
 */
private void refreshInvoker(List<URL> invokerUrls){
    if (invokerUrls != null && invokerUrls.size() == 1 && invokerUrls.get(0) != null
            && Constants.EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
        this.forbidden = true; // 禁止访问
        this.methodInvokerMap = null; // 置空列表
        destroyAllInvokers(); // 关闭所有Invoker
    } else {
        this.forbidden = false; // 允许访问
        Map<String, Invoker<T>> oldUrlInvokerMap = this.urlInvokerMap; // local reference
        if (invokerUrls.size() == 0 && this.cachedInvokerUrls != null){
            invokerUrls.addAll(this.cachedInvokerUrls);
        } else {
            this.cachedInvokerUrls = new HashSet<URL>();
            this.cachedInvokerUrls.addAll(invokerUrls);//缓存invokerUrls列表，便于交叉对比
        }
        if (invokerUrls.size() ==0 ){
            return;
        }
        //会重新走一遍服务的引用过程
        //给每个提供者创建一个Invoker
        Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;// 将URL列表转成Invoker列表
        Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap); // 换方法名映射Invoker列表
        // state change
        //如果计算错误，则不进行处理.
        if (newUrlInvokerMap == null || newUrlInvokerMap.size() == 0 ){
            logger.error(new IllegalStateException("urls to invokers error .invokerUrls.size :"+invokerUrls.size() + ", invoker.size :0. urls :"+invokerUrls.toString()));
            return ;
        }
        //服务提供者Invoker保存在这个map中
        this.methodInvokerMap = multiGroup ? toMergeMethodInvokerMap(newMethodInvokerMap) : newMethodInvokerMap;
        this.urlInvokerMap = newUrlInvokerMap;
        try{
            destroyUnusedInvokers(oldUrlInvokerMap,newUrlInvokerMap); // 关闭未使用的Invoker
        }catch (Exception e) {
            logger.warn("destroyUnusedInvokers error. ", e);
        }
    }
}

toInvokers(invokerUrls) 方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
private Map<String, Invoker<T>> toInvokers(List<URL> urls) {
    Map<String, Invoker<T>> newUrlInvokerMap = new HashMap<String, Invoker<T>>();
    if(urls == null || urls.size() == 0){
        return newUrlInvokerMap;
    }
    Set<String> keys = new HashSet<String>();
    String queryProtocols = this.queryMap.get(Constants.PROTOCOL_KEY);
    for (URL providerUrl : urls) {
    	//此时url是dubbo://192.168.110.197:20880/dubbo.common.hello.service.HelloService?anyhost=true&
        //application=dubbo-provider&application.version=1.0&dubbo=2.5.3&environment=product&
        //interface=dubbo.common.hello.service.HelloService&methods=sayHello&organization=china&
        //owner=cheng.xi&pid=5631&side=provider&timestamp=1489367571986
        //从注册中心获取到的携带提供者信息的url
        //如果reference端配置了protocol，则只选择匹配的protocol
        if (queryProtocols != null && queryProtocols.length() >0) {
            boolean accept = false;
            String[] acceptProtocols = queryProtocols.split(",");
            for (String acceptProtocol : acceptProtocols) {
                if (providerUrl.getProtocol().equals(acceptProtocol)) {
                    accept = true;
                    break;
                }
            }
            if (!accept) {
                continue;
            }
        }
        if (Constants.EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
            continue;
        }
        if (! ExtensionLoader.getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {
            logger.error(new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() + " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() + " to consumer " + NetUtils.getLocalHost() 
                    + ", supported protocol: "+ExtensionLoader.getExtensionLoader(Protocol.class).getSupportedExtensions()));
            continue;
        }
        URL url = mergeUrl(providerUrl);

        String key = url.toFullString(); // URL参数是排序的
        if (keys.contains(key)) { // 重复URL
            continue;
        }
        keys.add(key);
        // 缓存key为没有合并消费端参数的URL，不管消费端如何合并参数，如果服务端URL发生变化，则重新refer
        Map<String, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap; // local reference
        Invoker<T> invoker = localUrlInvokerMap == null ? null : localUrlInvokerMap.get(key);
        if (invoker == null) { // 缓存中没有，重新refer
            try {
                boolean enabled = true;
                if (url.hasParameter(Constants.DISABLED_KEY)) {
                    enabled = ! url.getParameter(Constants.DISABLED_KEY, false);
                } else {
                    enabled = url.getParameter(Constants.ENABLED_KEY, true);
                }
                if (enabled) {
                	//根据扩展点加载机制，这里使用的protocol是DubboProtocol
                    invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);
                }
            } catch (Throwable t) {
                logger.error("Failed to refer invoker for interface:"+serviceType+",url:("+url+")" + t.getMessage(), t);
            }
            if (invoker != null) { // 将新的引用放入缓存
                newUrlInvokerMap.put(key, invoker);
            }
        }else {
            newUrlInvokerMap.put(key, invoker);
        }
    }
    keys.clear();
    return newUrlInvokerMap;
}

创建invoker invoker = new InvokerDelegete<T>(protocol.refer(serviceType, url), url, providerUrl);：

  • 先使用DubboProtocol的refer方法，这一步会依次调用ProtocolFIlterListenerWrapper，ProtocolFilterWrapper，DubboProtocol中的refer方法。经过两个Wrapper中，会添加对应的InvokerListener并构建Invoker Filter链，在DubboProtocol中会创建一个DubboInvoker对象，该Invoker对象持有服务Class，providerUrl，负责和服务提供端通信的ExchangeClient。
  • 接着使用得到的Invoker创建一个InvokerDelegete

在DubboProtocol中创建DubboInvoker的时候代码如下：

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

查看getClients方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
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 = 1;
    }

    ExchangeClient[] clients = new ExchangeClient[connections];
    for (int i = 0; i < clients.length; i++) {
        if (service_share_connect){
        	//这里没有配置connections，就使用getSharedClient
            //getSharedClient中先去缓存中查找，没有的话就会新建，也是调用initClient方法
            clients[i] = getSharedClient(url);
        } else {
            clients[i] = initClient(url);
        }
    }
    return clients;
}

直接看initClient方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
//创建新连接
private ExchangeClient initClient(URL url) {
        
    // client type setting.
    String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));

    String version = url.getParameter(Constants.DUBBO_VERSION_KEY);
    boolean compatible = (version != null && version.startsWith("1.0."));
    url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() && compatible ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
    //默认开启heartbeat
    url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));

    // BIO存在严重性能问题，暂时不允许使用
    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 client ;
    try {
    	//如果lazy属性没有配置为true（我们没有配置，默认为false）ExchangeClient会马上和服务端建立连接
        //设置连接应该是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;
}

和服务端建立连接，Exchangers.connect(url ,requestHandler);，其实最后使用的是HeaderExchanger，Exchanger目前只有这一个实现：

1
2
3
4
5
6
7
public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
	//先经过HeaderExchangeHandler包装
    //然后是DecodeHandler
    //然后是Transporters.connect
    //返回一个HeaderExchangerClient，这里封装了client，channel，启动心跳的定时器等
    return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
}

Transporters.connect中也是根据SPI扩展获取Transport的具体实现，这里默认使用NettyTransporter.connect()，在NettyTransporter的connect方法中直接返回一个NettyClient(url, listener);，下面看下具体的NettyClient初始化细节，会先初始化AbstractPeer这里只是吧url和handler赋值；然后是AbstractEndpoint初始化：

1
2
3
4
5
6
7
public AbstractEndpoint(URL url, ChannelHandler handler) {
    super(url, handler);
    //获取编解码器，这里是DubboCountCodec
    this.codec = getChannelCodec(url);
    this.timeout = url.getPositiveParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT);
    this.connectTimeout = url.getPositiveParameter(Constants.CONNECT_TIMEOUT_KEY, Constants.DEFAULT_CONNECT_TIMEOUT);
}

接着是AbstractClient的初始化：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
    super(url, handler);
    send_reconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false);
    shutdown_timeout = url.getParameter(Constants.SHUTDOWN_TIMEOUT_KEY, Constants.DEFAULT_SHUTDOWN_TIMEOUT);
    //默认重连间隔2s，1800表示1小时warning一次.
    reconnect_warning_period = url.getParameter("reconnect.waring.period", 1800);

    try {
    	//具体实现在子类中
        doOpen();
    } catch (Throwable t) {。。。 }
    try {
        // 连接
        connect();
    } catch (RemotingException t) {。。。} 
	// TODO暂没理解
    executor = (ExecutorService) ExtensionLoader.getExtensionLoader(DataStore.class)
        .getDefaultExtension().get(Constants.CONSUMER_SIDE, Integer.toString(url.getPort()));
    ExtensionLoader.getExtensionLoader(DataStore.class)
        .getDefaultExtension().remove(Constants.CONSUMER_SIDE, Integer.toString(url.getPort()));
}

看下在NettyClient中doOpen()的实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
protected void doOpen() throws Throwable {
    NettyHelper.setNettyLoggerFactory();
    bootstrap = new ClientBootstrap(channelFactory);
    // config
    // @see org.jboss.netty.channel.socket.SocketChannelConfig
    bootstrap.setOption("keepAlive", true);
    bootstrap.setOption("tcpNoDelay", true);
    bootstrap.setOption("connectTimeoutMillis", getTimeout());
    final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
    bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
        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;
        }
    });
}

这里是Netty3中的客户端连接的一些常规步骤，暂不做具体解析。open之后，就是真正连接服务端的操作了，connect()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
protected void connect() throws RemotingException {
    connectLock.lock();
    try {
        if (isConnected()) {
            return;
        }
        //初始化重连的线程
        initConnectStatusCheckCommand();
        //连接，在子类中实现
        doConnect();
        reconnect_count.set(0);
        reconnect_error_log_flag.set(false);
    } catch (RemotingException e) {。。。} finally {
        connectLock.unlock();
    }
}

NettyClient中的doConnect方法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
protected void doConnect() throws Throwable {
    long start = System.currentTimeMillis();
    //消费者端开始连接，这一步的时候，服务提供者端就接到了连接请求，开始处理了
    ChannelFuture future = bootstrap.connect(getConnectAddress());
    try{
        boolean ret = future.awaitUninterruptibly(getConnectTimeout(), TimeUnit.MILLISECONDS);
        if (ret && future.isSuccess()) {
            Channel newChannel = future.getChannel();
            newChannel.setInterestOps(Channel.OP_READ_WRITE);
            try {
                // 关闭旧的连接
                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 {
                    NettyClient.this.channel = newChannel;
                }
            }
        } else if (future.getCause() != null) { throw。。。  } else {throw 。。。 }
    }finally{
        if (! isConnected()) {
            future.cancel();
        }
    }
}

这里连接的细节都交给了netty。

NettyClient初始化完成之后，返回给Transporters，再返回给HeaderExchanger，HeaderExchanger中将NettyClient包装成HeaderExchangeClient返回给DubboProtocol的initClient方法中，到此在getSharedClient中就获取到了一个ExchangeClient，然后包装一下返回client = new ReferenceCountExchangeClient(exchagneclient, ghostClientMap);。

到这里在DubboProtocol的refer方法中这句DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);创建DubboInvoker就已经解析完成，创建过程中连接了服务端，包含一个ExchangeClient等：

1
2
3
4
5
6
7
8
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);
    //将invoker缓存
    invokers.add(invoker);
    //返回invoker
    return invoker;
}

接着返回ProtocolFilterWrapper的refer方法，在这里会构建invoker链：

1
2
3
4
5
6
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
        return protocol.refer(type, url);
    }
    return buildInvokerChain(protocol.refer(type, url), Constants.REFERENCE_FILTER_KEY, Constants.CONSUMER);
}

接着再返回到ProtocolListenerWrapper的refer方法，这里会初始化监听器，包装：

1
2
3
4
5
6
7
8
9
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
    if (Constants.REGISTRY_PROTOCOL.equals(url.getProtocol())) {
        return protocol.refer(type, url);
    }
    return new ListenerInvokerWrapper<T>(protocol.refer(type, url), 
            Collections.unmodifiableList(
                    ExtensionLoader.getExtensionLoader(InvokerListener.class)
                    .getActivateExtension(url, Constants.INVOKER_LISTENER_KEY)));
}

接着在返回到toInvokers方法，然后返回refreshInvoker方法的Map<String, Invoker<T>> newUrlInvokerMap = toInvokers(invokerUrls) ;这就获得了Invoker，接着就是方法名映射Invoker列表：Map<String, List<Invoker<T>>> newMethodInvokerMap = toMethodInvokers(newUrlInvokerMap);这里将invokers列表转成与方法的映射关系。到这里refreshInvoker方法就完成了，在往上就返回到AbstractRegistry的notify方法，到这里也完成了。

创建服务代理

到这里有关消费者端注册到注册中心和订阅注册中心就完事儿了，这部分是在RegistryProtocol.doRefer方法中，这个方法最后一句是return cluster.join(directory);，这里由Cluster组件创建一个Invoker并返回，这里的cluster默认是用FailoverCluster，最后返回的是经过MockClusterInvoker包装过的FailoverCluster。继续返回到ReferenceConfig中createProxy方法，这时候我们已经完成了消费者端引用服务的Invoker。然后最后返回的是根据我们得到的invoker创建的服务代理return (T) proxyFactory.getProxy(invoker);。这里proxyFactory是我们在最上面列出的动态生成的代码。

首先经过AbstractProxyFactory的处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public <T> T getProxy(Invoker<T> invoker) throws RpcException {
    Class<?>[] interfaces = null;
    String config = invoker.getUrl().getParameter("interfaces");
    if (config != null && config.length() > 0) {
        String[] types = Constants.COMMA_SPLIT_PATTERN.split(config);
        if (types != null && types.length > 0) {
            interfaces = new Class<?>[types.length + 2];
            interfaces[0] = invoker.getInterface();
            interfaces[1] = EchoService.class;
            for (int i = 0; i < types.length; i ++) {
                interfaces[i + 1] = ReflectUtils.forName(types[i]);
            }
        }
    }
    if (interfaces == null) {
        interfaces = new Class<?>[] {invoker.getInterface(), EchoService.class};
    }
    //这里默认使用的是JavassistProxyFactory的实现
    return getProxy(invoker, interfaces);
}

然后经过StubProxyFactoryWrapper的处理：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
public <T> T getProxy(Invoker<T> invoker) throws RpcException {
    T proxy = proxyFactory.getProxy(invoker);
    if (GenericService.class != invoker.getInterface()) {
        String stub = invoker.getUrl().getParameter(Constants.STUB_KEY, invoker.getUrl().getParameter(Constants.LOCAL_KEY));
        if (ConfigUtils.isNotEmpty(stub)) {
            Class<?> serviceType = invoker.getInterface();
            if (ConfigUtils.isDefault(stub)) {
                if (invoker.getUrl().hasParameter(Constants.STUB_KEY)) {
                    stub = serviceType.getName() + "Stub";
                } else {
                    stub = serviceType.getName() + "Local";
                }
            }
            try {
                Class<?> stubClass = ReflectUtils.forName(stub);
                if (! serviceType.isAssignableFrom(stubClass)) {
                    throw new IllegalStateException("The stub implemention class " + stubClass.getName() + " not implement interface " + serviceType.getName());
                }
                try {
                    Constructor<?> constructor = ReflectUtils.findConstructor(stubClass, serviceType);
                    proxy = (T) constructor.newInstance(new Object[] {proxy});
                    //export stub service
                    URL url = invoker.getUrl();
                    if (url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT)){
                        url = url.addParameter(Constants.STUB_EVENT_METHODS_KEY, StringUtils.join(Wrapper.getWrapper(proxy.getClass()).getDeclaredMethodNames(), ","));
                        url = url.addParameter(Constants.IS_SERVER_KEY, Boolean.FALSE.toString());
                        try{
                            export(proxy, (Class)invoker.getInterface(), url);
                        }catch (Exception e) {
                            LOGGER.error("export a stub service error.", e);
                        }
                    }
                } catch (NoSuchMethodException e) {
                    throw new IllegalStateException("No such constructor \"public " + stubClass.getSimpleName() + "(" + serviceType.getName() + ")\" in stub implemention class " + stubClass.getName(), e);
                }
            } catch (Throwable t) {
                LOGGER.error("Failed to create stub implemention class " + stub + " in consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", cause: " + t.getMessage(), t);
                // ignore
            }
        }
    }
    return proxy;
}

返回代理。到此HelloService helloService = (HelloService) applicationContext.getBean("helloService");就解析完成了，得到了服务的代理，代理会被注册到Spring容器中，可以调用服务方法了。接下来的方法调用过程，是消费者发送请求，提供者处理，然后消费者接受处理结果的请求。

初始化的过程：主要做了注册到注册中心，监听注册中心，连接到服务提供者端，创建代理。这些都是为了下面消费者和提供者之间的通信做准备。






















































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































































以上是关于Dubbo中消费者初始化的过程解析的主要内容，如果未能解决你的问题，请参考以下文章

Dubbo中服务消费者和服务提供者之间的请求和响应过程

Dubbo-服务消费者初始化

Dubbo源码解析：网络通信

Dubbo源码解析：网络通信

[Dubbo3.0.8源码解析系列]-14-Dubbo配置加载全解析

Dubbo-服务提供者初始化