service reference

在编写好服务之后,dubbo会将服务export出去,这个时候就可以编写consumer来调用这个服务了。dubbo作为一个rpc框架,使用者使用远程服务和使用本地服务是类似的,不用关心远程服务在哪里,怎么引用的,因为dubbo包含了自动发现和引用服务的功能。

dubbo引用服务主要工作:

  • 创建proxy和Invoker(DubboInvoker里面会启动NettyClient)
  • 将consumer注册到注册中心
  • 订阅configurators、providers、routers

通过Java代码引用

ReferenceConfig<TestDubboService> reference = new ReferenceConfig();
reference.setApplication(new ApplicationConfig(appName));
reference.setRegistry(new RegistryConfig(dubboRegistry));
reference.setInterface(TestDubboService.class);
reference.setTimeout(timeout);
TestDubboService service = (TestDubboService)reference.get();

使用Java代码配置很明显,直接使用ReferenceConfig.get获取一个proxy

通过spring配置引用

<dubbo:reference interface="com.test.service.TestDubboService" id="testDubboService" />

spring解析该dubbo自定义标签的时候(请提前学习spring如何解析自定义标签),会初始化ReferenceBean,该bean是一个factoryBean并且继承自ReferenceConfig,在getBean方法中调用了ReferenceConfig.get,接下来的方式就和上面“使用Java代码引用”一致了。所以dubbo引用服务的工作就主要在于如何创建proxy。

ReferenceConfig初始化

ReferenceConfig的主要作用是配置并引用远程服务,创建远程服务的本地代理。ReferenceBean继承自ReferenceConfig,ReferenceConfig是一个FactoryBean ,实现了getObject方法,在spring容器初始化完成的时候会初始化配置为非lazyInit的bean,也就会调用ReferenceBean.getObject方法,里面会调用ReferenceConfig.get方法,从而触发ServiceConfig的初始化方法ServiceConfig.init。

inti方法的主要逻辑是:

  1. 判断是否已经初始化,get方法是同步方法,所以只需直接判断标志位initialized即可
  2. 判断配置的interface是否正确
  3. 判断-D参数配置或者配置文件是否配置是直连提供者
  4. 配置application、module、registries、monitor等
  5. 检查stub和mock配置(类似provider端的检查)
  6. 搜集需要配置到URL中的参数,先将参数收集到map中,URL参数在refer的过程中极其重要,dubbo中的所有配置几乎都是靠URL传递,从URL中获取或者设置到URL中
  7. 创建远程服务的本地代理,createProxy

如何创建proxy

由于是远程服务,consumer需要有一个代理来处理consumer发起的远程过程调用。dubbo通过远程调用的可执行体Invoker的代理来实现。接下来主要就是先创建Invoker,然后创建Invoker的proxy。

创建Invoker调用堆栈如下

createProxy的主要功能:

  1. 判断ReferenceConfig中是否配置了url,如果配置了url,则不从registry中获取,直接使用配置的url。
  2. 没有配置url,从registry配置拼装url
  3. 根据上面配置好的URL来refer对应的服务,创建远程服务的可执行体Invoker
  4. 将所有的invokers聚合成一个可执行实体MockClusterInvoker
  5. 给MockClusterInvoker创建一个代理类,这个代理类就是我们在consumer端使用的远程服务代理,该代理实现了对应的service接口,对应的InvocationHandler就是作为代理类构造方法入参的MockClusterInvoker,在后面一节分析中会说明consumer怎么发起调用

下面主要说说Invoker的创建过程:

  1. 和provider一样,会调用ProtocolFilterWrapper#refer和ProtocolListenerWrapper#refer,分别构造filter链和调用对应的listener.referred
  2. RegistryProtocol#refer会判断是否配置了group
  3. 根据上面的调用堆栈会调用ZookeeperRegistry#doSubscribe,该方法中会订阅providers、configurators等,并通过notify方法来调用到AbstractRegistry#notify,里面会针对每一个category调用对应listener.notify,consumer端的listener是RegistryDirectory,所以这里会调用RegistryDirectory#notify
  4. RegistryDirectory#notify,这个方法也是registry对应节点变化后监听的listener,会对每一种监听的节点类型做处理,这里先只看provider的处理,调用refreshInvoker方法
  5. refreshInvoker方法就是将配置好的url转换为Invoker,如果转化后的invoker至少有一个,并且少于原来的invoker(缓存的invoker),则会把废弃的invoker销毁掉(destroy)

这里具体说明一下consumer订阅,ZookeeperRegistry#doSubscribe中会将url中配置的category取出来拼接成registry的目录节点形式,然后订阅这些节点

// RegistryProtocol类
private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
  // ... 省略中间代码
  // 这里调用的是RegistryDirectory.subscribe方法
  // 在这里将consumer端想要订阅的category添加到url,包括providers,configurators,routers
  directory.subscribe(subscribeUrl.addParameter(Constants.CATEGORY_KEY, 
                                                Constants.PROVIDERS_CATEGORY 
                                                + "," + Constants.CONFIGURATORS_CATEGORY 
                                                + "," + Constants.ROUTERS_CATEGORY));
  return cluster.join(directory);
}

// ZookeeperRegistry在doSubscribe调用自己的下面这个方法,将URL中的category转化为registry中的目录对应的url
private String[] toCategoriesPath(URL url) {
  String[] categroies;
  if (Constants.ANY_VALUE.equals(url.getParameter(Constants.CATEGORY_KEY))) {
    // 如果配置的category是*,则取所有的category:providers,consumer,routers,configurators
    categroies = new String[] {Constants.PROVIDERS_CATEGORY, Constants.CONSUMERS_CATEGORY, 
                               Constants.ROUTERS_CATEGORY, Constants.CONFIGURATORS_CATEGORY};
  } else {
    categroies = url.getParameter(Constants.CATEGORY_KEY, new String[] {Constants.DEFAULT_CATEGORY});
  }
  String[] paths = new String[categroies.length];
  for (int i = 0; i < categroies.length; i ++) {
    // 将category拼接成registry中的目录形式,类似:/dubbo/com.test.service.TestDubboService/providers
    paths[i] = toServicePath(url) + Constants.PATH_SEPARATOR + categroies[i];
  }
  return paths;
}


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) {
          // 如果之前该路径没有添加过listener,则创建一个map来放置listener
          zkListeners.putIfAbsent(url, new ConcurrentHashMap<NotifyListener, ChildListener>());
          listeners = zkListeners.get(url);
        }
        ChildListener zkListener = listeners.get(listener);
        if (zkListener == null) {
          // 如果没有添加过对于子节点的listener,则创建
          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);
        }
        zkClient.create(path, false);
        // 添加listener到该目录及其子节点
        List<String> children = zkClient.addChildListener(path, zkListener);
        if (children != null) {
          urls.addAll(toUrlsWithEmpty(url, path, children));
        }
      }
      // 这个方法本身会导致监听的目录及其子节点变化,直接调用notify
      notify(url, listener, urls);
    }
  } catch (Throwable e) {
    throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  }
}

上面是consumer的订阅部分的源码,在consumer订阅的时候会调用FailbackRegistry#notify,接下来就是将url转换为Invoker,接下来的调用链路可以参考上面方法调用堆栈的图,转化的主要代码为:

com.alibaba.dubbo.registry.integration.RegistryDirectory#refreshInvoker
com.alibaba.dubbo.registry.integration.RegistryDirectory#toInvokers

这两个方法中的源代码注释较为详细了就不再赘述了。

在toInvokers方法中会调用DubboProtocol#refer,在该方法中启动NettyClient。

总结

provider提供服务后,consumer端就可以找到并引用该服务,接下来就可以像使用本地服务一样使用该服务了,发起远程该过程调用。