dubbo服务引入流程

发表于 更新于 分类于 ，
本文字数： 13k 阅读时长 ≈ 12 分钟

服务引入流程

服务消费者消费一个服务的详细过程

服务引入时序图

服务引入时序图

服务发现与引入

引用服务

  服务发现与引入的流程与暴露流程相似，只是经过标签初始化后，采用的是ReferenceBean对象，而服务暴露采用的是ServiceBean对象。
  与暴露过程中相同，spring初始化完成bean的组装后会调用InitializingBean的afterPropertiesSet方法。
  这个方法中会调用getObject方法，继续调用get方法，进入ReferenceConfig对象的init方法中。

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
// ReferenceBean
@Override
@SuppressWarnings({"unchecked"})
public void afterPropertiesSet() throws Exception {
     // 省略一堆流程

     if (shouldInit()) {
          getObject();
     }
}

@Override
public Object getObject() {
     return get();
}

public synchronized T get() {
     checkAndUpdateSubConfigs();

     if (destroyed) {
          throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
     }
     if (ref == null) {
          init();
     }
     return ref;
}

  在ReferenceConfig的init方法中对参数进行组装，紧接者就会调用createProxy方法创建代理对象。

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
// ReferenceConfig
private void init() {
     if (initialized) {
          return;
     }
     initialized = true;
     checkStubAndLocal(interfaceClass);
     checkMock(interfaceClass);
     Map<String, String> map = new HashMap<String, String>();

     map.put(Constants.SIDE_KEY, Constants.CONSUMER_SIDE);
     appendRuntimeParameters(map);
     if (!isGeneric()) {
          String revision = Version.getVersion(interfaceClass, version);
          if (revision != null && revision.length() > 0) {
               map.put("revision", revision);
          }

          String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
          if (methods.length == 0) {
               logger.warn("No method found in service interface " + interfaceClass.getName());
               map.put("methods", Constants.ANY_VALUE);
          } else {
               map.put("methods", StringUtils.join(new HashSet<String>(Arrays.asList(methods)), ","));
          }
     }
     map.put(Constants.INTERFACE_KEY, interfaceName);
     appendParameters(map, application);
     appendParameters(map, module);
     appendParameters(map, consumer, Constants.DEFAULT_KEY);
     appendParameters(map, this);
     Map<String, Object> attributes = null;
     if (CollectionUtils.isNotEmpty(methods)) {
          attributes = new HashMap<String, Object>();
          for (MethodConfig methodConfig : methods) {
               appendParameters(map, methodConfig, methodConfig.getName());
               String retryKey = methodConfig.getName() + ".retry";
               if (map.containsKey(retryKey)) {
               String retryValue = map.remove(retryKey);
               if ("false".equals(retryValue)) {
                    map.put(methodConfig.getName() + ".retries", "0");
               }
               }
               attributes.put(methodConfig.getName(), convertMethodConfig2AyncInfo(methodConfig));
          }
     }

     String hostToRegistry = ConfigUtils.getSystemProperty(Constants.DUBBO_IP_TO_REGISTRY);
     if (StringUtils.isEmpty(hostToRegistry)) {
          hostToRegistry = NetUtils.getLocalHost();
     }
     map.put(Constants.REGISTER_IP_KEY, hostToRegistry);

     ref = createProxy(map);

     String serviceKey = URL.buildKey(interfaceName, group, version);
     ApplicationModel.initConsumerModel(serviceKey, buildConsumerModel(serviceKey, attributes));
}

  createProxy方法中，会进行根据SPI机制选择不同的Protocol进行服务的发现与引入，获得invoker对象，然后对获取到的invoker对象创建代理并返回。

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
71
72
73
74
75
76
77
78
79
80
81
82
private T createProxy(Map<String, String> map) {
     if (shouldJvmRefer(map)) {
          URL url = new URL(Constants.LOCAL_PROTOCOL, Constants.LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
          invoker = refprotocol.refer(interfaceClass, url);
          if (logger.isInfoEnabled()) {
               logger.info("Using injvm service " + interfaceClass.getName());
          }
     } else {
          if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
               String[] us = Constants.SEMICOLON_SPLIT_PATTERN.split(url);
               if (us != null && us.length > 0) {
               for (String u : us) {
                    URL url = URL.valueOf(u);
                    if (StringUtils.isEmpty(url.getPath())) {
                         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 { // assemble URL from register center's configuration
               checkRegistry();
               List<URL> us = loadRegistries(false);
               if (CollectionUtils.isNotEmpty(us)) {
               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.isEmpty()) {
               throw new IllegalStateException("No such any registry to reference " + interfaceName + " on the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion() + ", please config <dubbo:registry address=\"...\" /> to your spring config.");
               }
          }

          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; // use last registry url
               }
               }
               if (registryURL != null) { // registry url is available
               // use RegistryAwareCluster only when register's cluster is available
               URL u = registryURL.addParameter(Constants.CLUSTER_KEY, RegistryAwareCluster.NAME);
               // The invoker wrap relation would be: RegistryAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, will execute route) -> Invoker
               invoker = cluster.join(new StaticDirectory(u, invokers));
               } else { // not a registry url, must be direct invoke.
               invoker = cluster.join(new StaticDirectory(invokers));
               }
          }
     }

     if (shouldCheck() && !invoker.isAvailable()) {
          // make it possible for consumer to retry later if provider is temporarily unavailable
          initialized = false;
          throw new IllegalStateException("Failed to check the status of the service " + interfaceName + ". No provider available for the service " + (group == null ? "" : group + "/") + interfaceName + (version == null ? "" : ":" + version) + " from the url " + invoker.getUrl() + " to the consumer " + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
     }
     if (logger.isInfoEnabled()) {
          logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
     }
     /**
     * @since 2.7.0
     * ServiceData Store
     */
     MetadataReportService metadataReportService = null;
     if ((metadataReportService = getMetadataReportService()) != null) {
          URL consumerURL = new URL(Constants.CONSUMER_PROTOCOL, map.remove(Constants.REGISTER_IP_KEY), 0, map.get(Constants.INTERFACE_KEY), map);
          metadataReportService.publishConsumer(consumerURL);
     }
     // create service proxy
     return (T) proxyFactory.getProxy(invoker);
}

  在Protocol层中，也根据SPI机制选择对应的Protocol实现服务调用。

直连引用服务

直连引用服务

  当在本地直连引用服务时，默认会进入DubboProtocol，也可以配置其他协议。暴露时进入的入口为export方法，引入的入口就是refer。这里会直接返回提供者的引用。

1
2
3
4
5
6
7
8
9
10
@Override
public <T> Invoker<T> refer(Class<T> serviceType, URL url) throws RpcException {
     optimizeSerialization(url);

     // create rpc invoker.
     DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
     invokers.add(invoker);

     return invoker;
}

  这里会调用getClients方法，进而调用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
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
private ExchangeClient[] getClients(URL url) {
     // whether to share connection

     boolean useShareConnect = false;

     int connections = url.getParameter(Constants.CONNECTIONS_KEY, 0);
     List<ReferenceCountExchangeClient> shareClients = null;
     // if not configured, connection is shared, otherwise, one connection for one service
     if (connections == 0) {
          useShareConnect = true;

          /**
          * The xml configuration should have a higher priority than properties.
          */
          String shareConnectionsStr = url.getParameter(Constants.SHARE_CONNECTIONS_KEY, (String) null);
          connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(Constants.SHARE_CONNECTIONS_KEY,
               Constants.DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr);
          shareClients = getSharedClient(url, connections);
     }

     ExchangeClient[] clients = new ExchangeClient[connections];
     for (int i = 0; i < clients.length; i++) {
          if (useShareConnect) {
               clients[i] = shareClients.get(i);

          } else {
               clients[i] = initClient(url);
          }
     }

     return clients;
}

private ExchangeClient initClient(URL url) {

     // client type setting.
     String str = url.getParameter(Constants.CLIENT_KEY, url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_CLIENT));

     url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);
     // enable heartbeat by default
     url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));

     // BIO is not allowed since it has severe performance issue.
     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 {
          // connection should be 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;
}

  接下来的流程与服务暴露相似，不同在于，暴露是一层一层通过bind方法暴露出去，而消费方是一层层通过connect方法建立netty连接。

从注册中心发现引用服务

从注册中心发现引用服务

  与暴露过程中相似，在调用Protocol获取invoker(提供者的引用)时，也会使用到SPI机制，ProtocolFilterWrapper和ProtocolListenerWrapper中：

1
2
3
4
5
6
7
8
// ProtocolFilterWrapper
@Override
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);
}
1
2
3
4
5
6
7
8
9
10
11
// ProtocolListenerWrapper
@Override
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)));
}

  与暴露过程相同，这里如果发现是需要去注册中心获取服务，会直接进入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
35
36
37
38
39
40
41
42
@Override
@SuppressWarnings("unchecked")
public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
     url = URLBuilder.from(url)
               .setProtocol(url.getParameter(REGISTRY_KEY, DEFAULT_REGISTRY))
               .removeParameter(REGISTRY_KEY)
               .build();
     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_KEY));
     String group = qs.get(Constants.GROUP_KEY);
     if (group != null && group.length() > 0) {
          if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
               return doRefer(getMergeableCluster(), registry, type, url);
          }
     }
     return doRefer(cluster, registry, type, url);
}

private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
     RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
     directory.setRegistry(registry);
     directory.setProtocol(protocol);
     // all attributes of REFER_KEY
     Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters());
     URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
     if (!ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true)) {
          directory.setRegisteredConsumerUrl(getRegisteredConsumerUrl(subscribeUrl, url));
          registry.register(directory.getRegisteredConsumerUrl());
     }
     directory.buildRouterChain(subscribeUrl);
     directory.subscribe(subscribeUrl.addParameter(CATEGORY_KEY,
               PROVIDERS_CATEGORY + "," + CONFIGURATORS_CATEGORY + "," + ROUTERS_CATEGORY));

     Invoker invoker = cluster.join(directory);
     ProviderConsumerRegTable.registerConsumer(invoker, url, subscribeUrl, directory);
     return invoker;
}

  这里最终调用到了doRefer方法，在这里在注册中心，注册节点，并且订阅通知，最终返回invoker对象。这里的invoker都是服务消费invoker。
  注册消费者节点后，注册目录会订阅通知。
  在注册中心注册节点的过程与暴露过程一样，只不过这里注册的是消费者节点，暴露注册的是提供者节点。
  在这个过程中，cluster的作用是，通过Cluster扩展点将多个多个提供者应用伪装组合成单个提供者引用。

相关文章