dubbo学习-服务引用源码剖析

<dubbo:reference id="demoService" check="false" interface="org.apache.dubbo.demo.DemoService" retries="0" init="true"/>


// 饿汉式，如果配置了init的参数为true，则直接进行服务引用
if (shouldInit()) {
  getObject();
}

 public synchronized T get() {
   checkAndUpdateSubConfigs(); // ①
   if (destroyed) {
     throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
   }
   if (ref == null) { // ②
     init();
   }
   return ref;
 }

private T createProxy(Map<String, String> map) {
   // 是否打开本地引用
   if (shouldJvmRefer(map)) {
     URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
     // 调用InjvmProtocol的refer方法生成InjvmInvoker实例
     invoker = REF_PROTOCOL.refer(interfaceClass, url);
     if (logger.isInfoEnabled()) {
       logger.info("Using injvm service " + interfaceClass.getName());
     }
   } else {
     urls.clear(); // reference retry init will add url to urls, lead to OOM
     // 用户是否指定服务提供方地址：可以使服务提供方的ip地址，也就是服务直连的方式
     if (url != null && url.length() > 0) { // ①
       String[] us = 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 (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
             urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
           } else { // ②
             urls.add(ClusterUtils.mergeUrl(url, map));
           }
         }
       }
     } else { // ③
       // if protocols not injvm checkRegistry
       if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())){
         // 检查注册中心的配置是否存在，然后将注册信息转换为Registry对象
         checkRegistry(); // ④ 
         List<URL> us = loadRegistries(false); // ⑤
         if (CollectionUtils.isNotEmpty(us)) {
           for (URL u : us) { // ⑥
             URL monitorUrl = loadMonitor(u);
             if (monitorUrl != null) {
               map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
             }
             urls.add(u.addParameterAndEncoded(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 = REF_PROTOCOL.refer(interfaceClass, urls.get(0));
     } else {
       // ⑧
       List<Invoker<?>> invokers = new ArrayList<Invoker<?>>();
       URL registryURL = null;
       for (URL url : urls) { // ⑨ 有多个提供者地址
         invokers.add(REF_PROTOCOL.refer(interfaceClass, url));
         if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {  // ⑩
           registryURL = url; // use last registry url
         }
       }
       if (registryURL != null) { 
         // ⑪
         URL u = registryURL.addParameter(CLUSTER_KEY, RegistryAwareCluster.NAME);
         // ⑫
         invoker = CLUSTER.join(new StaticDirectory(u, invokers));
       } else { // ⑬
         invoker = CLUSTER.join(new StaticDirectory(invokers));
       }
     }
   }
   // ⑭
   if (shouldCheck() && !invoker.isAvailable()) {
     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());
   }


   // ⑮
   return (T) PROXY_FACTORY.getProxy(invoker);
 }

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

public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
  if (REGISTRY_PROTOCOL.equals(url.getProtocol())) {
    return protocol.refer(type, url);
  }
  return buildInvokerChain(protocol.refer(type, url), REFERENCE_FILTER_KEY, CommonConstants.CONSUMER);
}

public <T> Invoker<T> protocolBindingRefer(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;
}

private ExchangeClient[] getClients(URL url) {
  // 是否使用共享连接
  boolean useShareConnect = false;
  // 获取连接数，默认为0。表示未配置
  int connections = url.getParameter(CONNECTIONS_KEY, 0);


  List<ReferenceCountExchangeClient> shareClients = null;
  // 如果没有配置connections，则共享连接，否则一个连接供一个service使用
  if (connections == 0) {
    useShareConnect = true;


    String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
    // 如果没有配置shareconnections的值，默认为connections的值默认为1
    connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY, 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 List<ReferenceCountExchangeClient> getSharedClient(URL url, int connectNum) {
   String key = url.getAddress();
   // 从换从中获取
   List<ReferenceCountExchangeClient> clients = referenceClientMap.get(key); // ①


   // 检查客户端是否可用，只要有一个不可用，就需要将不可用的客户端替换成可以用的客户端
   if (checkClientCanUse(clients)) { // ②
     // 因为创建的共享客户端是带有引用计数功能的ReferenceCountExchangeClient，所以这里会对所有的客户端引用计数+1
     batchClientRefIncr(clients);
     return clients;
   }


   locks.putIfAbsent(key, new Object());
   synchronized (locks.get(key)) { // ③
     clients = referenceClientMap.get(key);
     // dubbo check
     if (checkClientCanUse(clients)) {
       batchClientRefIncr(clients);
       return clients;
     }


     // 连接数必须大于等于1
     connectNum = Math.max(connectNum, 1);


     // 如果clients集合为空，则进行初始化，并放入到缓存
     if (CollectionUtils.isEmpty(clients)) { // ④
       clients = buildReferenceCountExchangeClientList(url, connectNum);
       referenceClientMap.put(key, clients);


     } else {
       for (int i = 0; i < clients.size(); i++) { // ⑤
         ReferenceCountExchangeClient referenceCountExchangeClient = clients.get(i);
         // 如果客户端集合中某个客户端不可用，创建一个新的客户端来替代
         // If there is a client in the list that is no longer available, create a new one to replace him.
         if (referenceCountExchangeClient == null || referenceCountExchangeClient.isClosed()) {
           clients.set(i, buildReferenceCountExchangeClient(url));
           continue;
         }


         referenceCountExchangeClient.incrementAndGetCount();
       }
     }


     locks.remove(key);


     return clients;
   }
 }

private ExchangeClient initClient(URL url) {
   // 获取配置的客户端类型配置，默认为netty
   String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));


   url = url.addParameter(CODEC_KEY, DubboCodec.NAME);
   // 默认开启心跳机制
   url = url.addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(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 {
     // 是否进行懒加载，也就是当有请求发生时，再去创建客户端
     if (url.getParameter(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;
 }

protected void doOpen() throws Throwable {
   // 实例化处理客户端业务的Handler
   final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this);
   // 实例化bootstrap
   bootstrap = new Bootstrap();
   bootstrap.group(nioEventLoopGroup)
     .option(ChannelOption.SO_KEEPALIVE, true)
     .option(ChannelOption.TCP_NODELAY, true)
     .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
     .channel(NioSocketChannel.class);


   // 设置超时时间，默认3s
   if (getConnectTimeout() < 3000) {
     bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
   } else {
     bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getConnectTimeout());
   }


   bootstrap.handler(new ChannelInitializer() {
     @Override
     protected void initChannel(Channel ch) throws Exception {
       int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
       NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
       ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
         .addLast("decoder", adapter.getDecoder()) // 添加解码器Handler到管道
         .addLast("encoder", adapter.getEncoder()) // 添加编码码器Handler到管道
         .addLast("client-idle-handler", new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
         .addLast("handler", nettyClientHandler); // 添加处理业务的handler到管道
       String socksProxyHost = ConfigUtils.getProperty(SOCKS_PROXY_HOST);
       if(socksProxyHost != null) {
         int socksProxyPort = Integer.parseInt(ConfigUtils.getProperty(SOCKS_PROXY_PORT, DEFAULT_SOCKS_PROXY_PORT));
         Socks5ProxyHandler socks5ProxyHandler = new Socks5ProxyHandler(new InetSocketAddress(socksProxyHost, socksProxyPort));
         ch.pipeline().addFirst(socks5ProxyHandler);
       }
     }
   });
 }

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(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并设置注册中心实例和协议
  RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url); // ①
  directory.setRegistry(registry);
  directory.setProtocol(protocol);
  // 从directory中获取所有和refer相关的参数
  Map<String, String> parameters = new HashMap<String, String>(directory.getUrl().getParameters()); // ②
  // 构建服务消费者的URL对象
  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);
  // 订阅providers、configurators、routers等节点数据
  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;
}

public void doRegister(URL url) {
  try {
    // 通过调用zkClient.create方法注册服务节点信息到zookeeper
    zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true));
  } catch (Throwable e) {
    throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
  }
}

public void subscribe(URL url) {
    setConsumerUrl(url);
    CONSUMER_CONFIGURATION_LISTENER.addNotifyListener(this); // 添加consumer端配置的监听器
    serviceConfigurationListener = new ReferenceConfigurationListener(this, url); // 添加服务引用相关配置的监听器
    registry.subscribe(url, this); // 执行订阅操作
}

 public void doSubscribe(final URL url, final NotifyListener listener) {
   try {
     // 如果配置的是*，那就会为root节点以及子节点添加监听器
     if (ANY_VALUE.equals(url.getServiceInterface())) {
       String root = toRootPath();
       ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
       if (listeners == null) {
         zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
         listeners = zkListeners.get(url);
       }
       ChildListener zkListener = listeners.get(listener);
       if (zkListener == null) {
         listeners.putIfAbsent(listener, (parentPath, currentChilds) -> {
           for (String child : currentChilds) {
             child = URL.decode(child);
             if (!anyServices.contains(child)) {
               anyServices.add(child);
               subscribe(url.setPath(child).addParameters(INTERFACE_KEY, child, Constants.CHECK_KEY, String.valueOf(false)), listener);
             }
           }
         });
         zkListener = listeners.get(listener);
       }
       zkClient.create(root, false);
       List<String> services = zkClient.addChildListener(root, zkListener);
       if (CollectionUtils.isNotEmpty(services)) {
         for (String service : services) {
           service = URL.decode(service);
           anyServices.add(service);
           subscribe(url.setPath(service).addParameters(INTERFACE_KEY, service,
                                                        Constants.CHECK_KEY, String.valueOf(false)), listener);
         }
       }
     } else {
       // 如果配置的不是*，那么就会获取providers、configurators、routes这几个节点的url，然后为这几个节点及子节点添加监听器
       List<URL> urls = new ArrayList<>();
       for (String path : toCategoriesPath(url)) {
         ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.get(url);
         if (listeners == null) {
           zkListeners.putIfAbsent(url, new ConcurrentHashMap<>());
           listeners = zkListeners.get(url);
         }
         ChildListener zkListener = listeners.get(listener);
         if (zkListener == null) {
           listeners.putIfAbsent(listener, (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, listener, toUrlsWithEmpty(url, parentPath, currentChilds)));
           zkListener = listeners.get(listener);
         }
         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);
   }
 }

// 包装类，对Cluster进行包装
public class MockClusterWrapper implements Cluster {


  private Cluster cluster;


  public MockClusterWrapper(Cluster cluster) {
    this.cluster = cluster;
  }


  @Override
  public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
    // 在方法内部，先调用被包装的cluster实例的join方法获取Invoker实例，然后再通过MockClusterInvoker进行封装
    return new MockClusterInvoker<T>(directory, this.cluster.join(directory));
  }


}


// 默认扩展实现类 FailoverCluster
public class FailoverCluster implements Cluster {


  public final static String NAME = "failover";


  @Override
  public <T> Invoker<T> join(Directory<T> directory) throws RpcException {
    // 返回一个FailoverClusterInvoker对象
    return new FailoverClusterInvoker<T>(directory);
  }


}

public <T> T getProxy(Invoker<T> invoker, boolean generic) throws RpcException {
  Class<?>[] interfaces = null;
  // 获取url中的interfaces参数值
  String config = invoker.getUrl().getParameter(INTERFACES);
  if (config != null && config.length() > 0) {
    // 解析配置，将服务接口类和EchoService.class方法如到interfaces数组
    String[] types = 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 + 2] = ReflectUtils.forName(types[i]);
      }
    }
  }
  if (interfaces == null) {
    interfaces = new Class<?>[]{invoker.getInterface(), EchoService.class};
  }
  
  // 泛化调用相关配置
  if (!GenericService.class.isAssignableFrom(invoker.getInterface()) && generic) {
    int len = interfaces.length;
    Class<?>[] temp = interfaces;
    interfaces = new Class<?>[len + 1];
    System.arraycopy(temp, 0, interfaces, 0, len);
    interfaces[len] = com.alibaba.dubbo.rpc.service.GenericService.class;
  }
  // 获取代理类，由子类实现
  return getProxy(invoker, interfaces);
}

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

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