• Dubbo服务发布与消费过程概述

    1 Dubbo服务发布端和消费端启动过程图

    Dubbo服务发布端和消费端启动过程如下所示：

    （1）Dubbo服务发布端启动过程

    • ServiceConfig通过ProxyFactory将服务提供类（ref）转换成invoker；
    • 然后通过Protocol将invoker转为Exporter。在这个过程中会先创建并启动NettyServer监听服务连接，然后将服务注册到服务注册中心。

    （2）Dubbo服务消费端启动过程

    • ReferenceConfig通过Protocol将远程服务转换为Invoker。在创建Invoker的同时会创建并启动NettyClient对象，使服务消费端与服务提供端建立TCP连接。
    • 然后通过ProxyFactory将Invoker转换为客户端（即消费端）需要的接口。

    2 Dubbo服务发布端启动过程

    2.1 Dubbo服务发布端启动过程概述

    • ServiceConfig通过ProxyFactory将服务提供类（ref）转换成invoker；
    • 然后通过Protocol将invoker转为Exporter。在这个过程中会先启动NettyServer监听服务连接，然后将服务注册到服务注册中心。

    2.2 源码剖析

     2.2.1 服务发布端启动入口

    服务发布端启动入口为 org.apache.dubbo.config.ServiceConfigBase#export() 方法，其中ServiceConfigBase为ServiceConfig的父类。

    1.     /**
    2.      * export service and auto start application instance
    3.      */
    4.     public final void export() {
    5.         export(RegisterTypeEnum.AUTO_REGISTER);
    6.     }

    其中项目启动时，服务被注册到注册中心的方式有以下四种：

    1. public enum RegisterTypeEnum {
    2.  
    3.     /**
    4.      * 不注册-Never register. Cannot be registered by any command(like QoS-online).
    5.      */
    6.     NEVER_REGISTER,
    7.  
    8.     /**
    9.      * 手动注册-Manual register. Can be registered by command(like QoS-online), but not register by default.
    10.      */
    11.     MANUAL_REGISTER,
    12.  
    13.     /**
    14.      * 自动注册（某个服务的所有的服务提供者都启动之后才注册）
    15.      * (INTERNAL) Auto register by deployer. Will be registered after deployer started.
    16.      * (Delay publish when starting. Prevent service from being invoked before all services are started)
    17.      */
    18.     AUTO_REGISTER_BY_DEPLOYER,
    19.  
    20.     /**
    21.      * 自动注册（某个服务的其中一个服务提供者启动之后就注册）
    22.      * Auto register. Will be registered when one service is exported.
    23.      */
    24.     AUTO_REGISTER;
    25. }

    export方法中调用了doExport方法（org.apache.dubbo.config.ServiceConfig#doExport），内部实现如下所示。

    1.     protected synchronized void doExport(RegisterTypeEnum registerType) {
    2.         // ...
    3.         
    4.         // 服务注册
    5.         doExportUrls(registerType);
    6.         exported();
    7.     }
    8.  
    9.  
    10.     private void doExportUrls(RegisterTypeEnum registerType) {
    11.         // ...
    12.  
    13.         // 加载注册中心信息
    14.         List registryURLs = ConfigValidationUtils.loadRegistries(this, true);
    15.  
    16.         for (ProtocolConfig protocolConfig : protocols) {
    17.             // ...
    18.             
    19.             // 注册服务
    20.             doExportUrlsFor1Protocol(protocolConfig, registryURLs, registerType);
    21.         }
    22.  
    23.         providerModel.setServiceUrls(urls);
    24.     }
    25.  
    26.  
    27.     private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List registryURLs, RegisterTypeEnum registerType) {
    28.         Map map = buildAttributes(protocolConfig);
    29.  
    30.         // remove null key and null value
    31.         map.keySet().removeIf(key -> StringUtils.isEmpty(key) || StringUtils.isEmpty(map.get(key)));
    32.         // init serviceMetadata attachments
    33.         serviceMetadata.getAttachments().putAll(map);
    34.  
    35.         URL url = buildUrl(protocolConfig, map);
    36.  
    37.         processServiceExecutor(url);
    38.  
    39.         // 注册服务
    40.         exportUrl(url, registryURLs, registerType);
    41.  
    42.         initServiceMethodMetrics(url);
    43.     }
    44.  
    45.  
    46.     private void exportUrl(URL url, List registryURLs, RegisterTypeEnum registerType) {
    47.         String scope = url.getParameter(SCOPE_KEY);
    48.         // don't export when none is configured
    49.         if (!SCOPE_NONE.equalsIgnoreCase(scope)) {
    50.  
    51.             // export to local if the config is not remote (export to remote only when config is remote)
    52.             if (!SCOPE_REMOTE.equalsIgnoreCase(scope)) {
    53.                 // 注册本地服务
    54.                 exportLocal(url);
    55.             }
    56.  
    57.             // export to remote if the config is not local (export to local only when config is local)
    58.             if (!SCOPE_LOCAL.equalsIgnoreCase(scope)) {
    59.                 // ...
    60.  
    61.                 // 注册远程服务
    62.                 url = exportRemote(url, registryURLs, registerType);
    63.  
    64.                 // ...
    65.             }
    66.         }
    67.         this.urls.add(url);
    68.     }

    2.2.2 将服务提供类（ref）转化为Exporter

    ServiceConfig通过ProxyFactory将服务提供类（ref）转换成invoker，然后通过Protocol将invoker转为Exporter。主要的实现方法为doExportUrl()方法。

    其中注册本地服务和注册远程服务最终均调用了doExportUrl()方法（org.apache.dubbo.config.ServiceConfig#doExportUrl）。注册本地服务时，withMetaData为false；注册远程服务时，withMetaData为true。doExportUrl方法的具体实现如下所示。

    1.     private void doExportUrl(URL url, boolean withMetaData, RegisterTypeEnum registerType) {
    2.         if (!url.getParameter(REGISTER_KEY, true)) {
    3.             registerType = RegisterTypeEnum.MANUAL_REGISTER;
    4.         }
    5.         if (registerType == RegisterTypeEnum.NEVER_REGISTER ||
    6.             registerType == RegisterTypeEnum.MANUAL_REGISTER ||
    7.             registerType == RegisterTypeEnum.AUTO_REGISTER_BY_DEPLOYER) {
    8.             url = url.addParameter(REGISTER_KEY, false);
    9.         }
    10.  
    11.         // 1、ServiceConfig通过ProxyFactory将服务提供类（ref）转换成invoker
    12.         Invoker invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
    13.         if (withMetaData) {
    14.             // 转为invoker的包装类
    15.             // wrap the invoker and all the metadata (ServiceConfig)
    16.             invoker = new DelegateProviderMetaDataInvoker(invoker, this);
    17.         }
    18.  
    19.         // 2、通过Protocol将invoker转为Exporter
    20.         Exporter exporter = protocolSPI.export(invoker);
    21.         exporters.computeIfAbsent(registerType, k -> new CopyOnWriteArrayList<>()).add(exporter);
    22.     }

    2.2.3 创建并启动NettyServer

    （1）在创建Exporter的同时，会启动NettyServer监听服务连接，然后将服务注册到服务注册中心。

    如果为远程服务暴露，则其内部根据URL中Protocol的类型为registry，会选择RegistryProtocol。如果为本地服务暴露，则其内部根据URL中Protocol的类型为injvm，会选InjvmProtocol。

    （2）以RegistryProtocol为例的export方法如下所示。其中通过doLocalExport将Invoker转为Exporter，同时启动NettyServer。通过getRegistry获取注册中心，以及通过register将当前服务注册到服务注册中心。

    （3）Dubbo服务发布端的NettyServer主要负责监听客户端连接以及读写IO操作。

    Dubbo默认的底层网络通信使用的是Netty，服务提供方NettyServer使用两级线程池，其中boss线程池主要用来接受客户端的链接请求，并把完成Tcp三次握手的连接分发给worker来处理。

    1.     public  Exporter export(final Invoker originInvoker) throws RpcException {
    2.         URL registryUrl = getRegistryUrl(originInvoker);
    3.         // url to export locally
    4.         URL providerUrl = getProviderUrl(originInvoker);
    5.  
    6.         // Subscribe the override data
    7.         final URL overrideSubscribeUrl = getSubscribedOverrideUrl(providerUrl);
    8.         final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
    9.         Map> overrideListeners = getProviderConfigurationListener(overrideSubscribeUrl).getOverrideListeners();
    10.         overrideListeners.computeIfAbsent(overrideSubscribeUrl, k -> new ConcurrentHashSet<>())
    11.             .add(overrideSubscribeListener);
    12.  
    13.         providerUrl = overrideUrlWithConfig(providerUrl, overrideSubscribeListener);
    14.         //export invoker
    15.         // 1、启动NettyServer
    16.         final ExporterChangeableWrapper exporter = doLocalExport(originInvoker, providerUrl);
    17.  
    18.         // url to registry
    19.         // 2、获取注册中心
    20.         final Registry registry = getRegistry(registryUrl);
    21.         final URL registeredProviderUrl = getUrlToRegistry(providerUrl, registryUrl);
    22.  
    23.         // decide if we need to delay publish (provider itself and registry should both need to register)
    24.         boolean register = providerUrl.getParameter(REGISTER_KEY, true) && registryUrl.getParameter(REGISTER_KEY, true);
    25.         if (register) {
    26.             // 3、将当前服务注册到服务注册中心
    27.             register(registry, registeredProviderUrl);
    28.         }
    29.  
    30.         // register stated url on provider model
    31.         registerStatedUrl(registryUrl, registeredProviderUrl, register);
    32.  
    33.  
    34.         exporter.setRegisterUrl(registeredProviderUrl);
    35.         exporter.setSubscribeUrl(overrideSubscribeUrl);
    36.         exporter.setNotifyListener(overrideSubscribeListener);
    37.         exporter.setRegistered(register);
    38.  
    39.         ApplicationModel applicationModel = getApplicationModel(providerUrl.getScopeModel());
    40.         if (applicationModel.modelEnvironment().getConfiguration().convert(Boolean.class, ENABLE_26X_CONFIGURATION_LISTEN, true)) {
    41.             if (!registry.isServiceDiscovery()) {
    42.                 // Deprecated! Subscribe to override rules in 2.6.x or before.
    43.                 registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
    44.             }
    45.         }
    46.  
    47.         notifyExport(exporter);
    48.         //Ensure that a new exporter instance is returned every time export
    49.         return new DestroyableExporter<>(exporter);
    50.     }

    创建和启动NettyServer的入口：在doLocalExport中最终会调用DubboProtocol中的export方法，export方法之后的主要调用链路为export()->openServer()->createServer()，具体实现如下。

    1.     public  Exporter export(Invoker invoker) throws RpcException {
    2.         checkDestroyed();
    3.         URL url = invoker.getUrl();
    4.  
    5.         // 1、创建Exporter。export service.
    6.         String key = serviceKey(url);
    7.         DubboExporter exporter = new DubboExporter(invoker, key, exporterMap);
    8.  
    9.         //export a stub service for dispatching event
    10.         // ...
    11.  
    12.         // 2、创建并启动NettyServer
    13.         openServer(url);
    14.         optimizeSerialization(url);
    15.  
    16.         return exporter;
    17.     }
    18.  
    19.  
    20.     private void openServer(URL url) {
    21.         checkDestroyed();
    22.         // find server.
    23.         // 服务提供者机器的ip和端口
    24.         String key = url.getAddress();
    25.         // client can export a service which only for server to invoke
    26.         // 只有服务提供者才会启动NettyServer监听
    27.         boolean isServer = url.getParameter(IS_SERVER_KEY, true);
    28.  
    29.         if (isServer) {
    30.             ProtocolServer server = serverMap.get(key);
    31.             if (server == null) {
    32.                 synchronized (this) {
    33.                     server = serverMap.get(key);
    34.                     if (server == null) {
    35.                         // 该服务对应的NettyServer还没启动过，则创建并启动NettyServer
    36.                         // 同一个机器的不同服务发布只会开启一个NettyServer
    37.                         serverMap.put(key, createServer(url));
    38.                         return;
    39.                     }
    40.                 }
    41.             }
    42.  
    43.             // server supports reset, use together with override
    44.             server.reset(url);
    45.         }
    46.     }
    47.  
    48.     private ProtocolServer createServer(URL url) {
    49.         url = URLBuilder.from(url)
    50.             // send readonly event when server closes, it's enabled by default
    51.             .addParameterIfAbsent(CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString())
    52.             // enable heartbeat by default
    53.             .addParameterIfAbsent(HEARTBEAT_KEY, String.valueOf(DEFAULT_HEARTBEAT))
    54.             .addParameter(CODEC_KEY, DubboCodec.NAME)
    55.             .build();
    56.  
    57.         String transporter = url.getParameter(SERVER_KEY, DEFAULT_REMOTING_SERVER);
    58.         if (StringUtils.isNotEmpty(transporter) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(transporter)) {
    59.             throw new RpcException("Unsupported server type: " + transporter + ", url: " + url);
    60.         }
    61.  
    62.         ExchangeServer server;
    63.         try {
    64.             // 启动NettyServer
    65.             server = Exchangers.bind(url, requestHandler);
    66.         } catch (RemotingException e) {
    67.             throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
    68.         }
    69.  
    70.         transporter = url.getParameter(CLIENT_KEY);
    71.         if (StringUtils.isNotEmpty(transporter) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(transporter)) {
    72.             throw new RpcException("Unsupported client type: " + transporter);
    73.         }
    74.  
    75.         DubboProtocolServer protocolServer = new DubboProtocolServer(server);
    76.         loadServerProperties(protocolServer);
    77.         return protocolServer;
    78.     }
    79.  
    80.

    DubboProtocol的createServer()之后的调用链为：

    createServer()->Exchangers的bind()->HeaderExchanger的bind()->Transporters的bind()->NettyTransporter的bind()。具体实现如下。

    1.     // Exchangers的bind方法
    2.     public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
    3.         if (url == null) {
    4.             throw new IllegalArgumentException("url == null");
    5.         }
    6.         if (handler == null) {
    7.             throw new IllegalArgumentException("handler == null");
    8.         }
    9.         url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
    10.         return getExchanger(url).bind(url, handler);
    11.     }
    12.  
    13.     // HeaderExchanger的bind方法
    14.     public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException {
    15.         ExchangeServer server;
    16.         boolean isPuServerKey = url.getParameter(IS_PU_SERVER_KEY, false);
    17.         if(isPuServerKey) {
    18.             server = new HeaderExchangeServer(PortUnificationExchanger.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    19.         }else {
    20.             server = new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    21.         }
    22.         return server;
    23.     }
    24.  
    25.     // Transporters的bind方法
    26.     public static RemotingServer bind(URL url, ChannelHandler... handlers) throws RemotingException {
    27.         if (url == null) {
    28.             throw new IllegalArgumentException("url == null");
    29.         }
    30.         if (handlers == null || handlers.length == 0) {
    31.             throw new IllegalArgumentException("handlers == null");
    32.         }
    33.         ChannelHandler handler;
    34.         if (handlers.length == 1) {
    35.             handler = handlers[0];
    36.         } else {
    37.             handler = new ChannelHandlerDispatcher(handlers);
    38.         }
    39.         return getTransporter(url).bind(url, handler);
    40.     }
    41.  
    42.     // NettyTransporter的bind方法
    43.     public RemotingServer bind(URL url, ChannelHandler handler) throws RemotingException {
    44.         return new NettyServer(url, handler);
    45.     }

    以 org.apache.dubbo.remoting.transport.netty4.NettyTransporter 为例，可以看到，在NettyTransporter的bind方法中，新建了一个NettyServer。

    在NettyServer的构造函数中，调用了其父类AbstractServer的构造函数，其中的核心方法为doOpen()-初始化和启动 netty server。具体如下。

    1.     public NettyServer(URL url, ChannelHandler handler) throws RemotingException {
    2.         // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREAD_POOL_KEY in CommonConstants.
    3.         // the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler
    4.         // 父类的构造函数
    5.         super(url, ChannelHandlers.wrap(handler, url));
    6.  
    7.         // read config before destroy
    8.         serverShutdownTimeoutMills = ConfigurationUtils.getServerShutdownTimeout(getUrl().getOrDefaultModuleModel());
    9.     }
    10.  
    11.     public AbstractServer(URL url, ChannelHandler handler) throws RemotingException {
    12.         super(url, handler);
    13.         executorRepository = ExecutorRepository.getInstance(url.getOrDefaultApplicationModel());
    14.         localAddress = getUrl().toInetSocketAddress();
    15.  
    16.         String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost());
    17.         int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort());
    18.         if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) {
    19.             bindIp = ANYHOST_VALUE;
    20.         }
    21.         bindAddress = new InetSocketAddress(bindIp, bindPort);
    22.         this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS);
    23.         try {
    24.             // 初始化和启动 netty server
    25.             doOpen();
    26.             if (logger.isInfoEnabled()) {
    27.                 logger.info("Start " + getClass().getSimpleName() + " bind " + getBindAddress() + ", export " + getLocalAddress());
    28.             }
    29.         } catch (Throwable t) {
    30.             throw new RemotingException(url.toInetSocketAddress(), null, "Failed to bind " + getClass().getSimpleName()
    31.                 + " on " + bindAddress + ", cause: " + t.getMessage(), t);
    32.         }
    33.         executors.add(executorRepository.createExecutorIfAbsent(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME)));
    34.     }

    以 org.apache.dubbo.remoting.transport.netty4.NettyServer.doOpen 为例，doOpen()具体实现如下所示。

    1.  
    2.     /**
    3.      * Init and start netty server
    4.      *
    5.      * @throws Throwable
    6.      */
    7.     @Override
    8.     protected void doOpen() throws Throwable {
    9.         // 创建ServerBootstrap
    10.         bootstrap = new ServerBootstrap();
    11.  
    12.         // 创建Netty的boss线程池和worker线程池
    13.         bossGroup = createBossGroup();
    14.         workerGroup = createWorkerGroup();
    15.  
    16.         // 配置NettyServer，添加handler到管线
    17.         final NettyServerHandler nettyServerHandler = createNettyServerHandler();
    18.         channels = nettyServerHandler.getChannels();
    19.  
    20.         initServerBootstrap(nettyServerHandler);
    21.  
    22.         // bind
    23.         try {
    24.             // 绑定本地端口，并启动监听服务
    25.             ChannelFuture channelFuture = bootstrap.bind(getBindAddress());
    26.             channelFuture.syncUninterruptibly();
    27.             channel = channelFuture.channel();
    28.         } catch (Throwable t) {
    29.             closeBootstrap();
    30.             throw t;
    31.         }
    32.     }

    2.2.4 将服务注册到服务注册中心

    由上可知，以RegistryProtocol为例，在export方法中，通过doLocalExport将Invoker转为Exporter，同时启动NettyServer，监听服务连接。然后通过getRegistry方法获取注册中心，并通过register方法将当前服务注册到服务注册中心。

    2.2.4.1 获取注册中心

    通过getRegistry方法获取注册中心，最终是通过RegistryFactory子类AbstractRegistryFactory中的createRegistry方法创建注册中心。比如创建一个ZookeeperRegistry作为Zookeeper注册中心。具体实现如下所示。

    1.     /**
    2.      * Get an instance of registry based on the address of invoker
    3.      *
    4.      * @param registryUrl
    5.      * @return
    6.      */
    7.     protected Registry getRegistry(final URL registryUrl) {
    8.         RegistryFactory registryFactory = ScopeModelUtil.getExtensionLoader(RegistryFactory.class, registryUrl.getScopeModel()).getAdaptiveExtension();
    9.         return registryFactory.getRegistry(registryUrl);
    10.     }
    11.  
    12.  
    13.     // AbstractRegistryFactory 的 getRegistry
    14.     @Override
    15.     public Registry getRegistry(URL url) {
    16.         if (registryManager == null) {
    17.             throw new IllegalStateException("Unable to fetch RegistryManager from ApplicationModel BeanFactory. " +
    18.                 "Please check if `setApplicationModel` has been override.");
    19.         }
    20.  
    21.         Registry defaultNopRegistry = registryManager.getDefaultNopRegistryIfDestroyed();
    22.         if (null != defaultNopRegistry) {
    23.             return defaultNopRegistry;
    24.         }
    25.  
    26.         url = URLBuilder.from(url)
    27.             .setPath(RegistryService.class.getName())
    28.             .addParameter(INTERFACE_KEY, RegistryService.class.getName())
    29.             .removeParameter(TIMESTAMP_KEY)
    30.             .removeAttribute(EXPORT_KEY)
    31.             .removeAttribute(REFER_KEY)
    32.             .build();
    33.  
    34.         String key = createRegistryCacheKey(url);
    35.         Registry registry = null;
    36.         boolean check = url.getParameter(CHECK_KEY, true) && url.getPort() != 0;
    37.  
    38.         // Lock the registry access process to ensure a single instance of the registry
    39.         // 加锁，确保注册中心是单例对象
    40.         registryManager.getRegistryLock().lock();
    41.         try {
    42.             // double check
    43.             // fix https://github.com/apache/dubbo/issues/7265.
    44.             defaultNopRegistry = registryManager.getDefaultNopRegistryIfDestroyed();
    45.             if (null != defaultNopRegistry) {
    46.                 return defaultNopRegistry;
    47.             }
    48.  
    49.             // 从本地缓存中获取注册中心对象
    50.             registry = registryManager.getRegistry(key);
    51.             if (registry != null) {
    52.                 return registry;
    53.             }
    54.  
    55.             // create registry by spi/ioc
    56.             // 创建服务注册中心实例
    57.             registry = createRegistry(url);
    58.             if (check && registry == null) {
    59.                 throw new IllegalStateException("Can not create registry " + url);
    60.             }
    61.  
    62.             if (registry != null) {
    63.                 // 将注册中心对象存入本地缓存中
    64.                 registryManager.putRegistry(key, registry);
    65.             }
    66.         } catch (Exception e) {
    67.             if (check) {
    68.                 throw new RuntimeException("Can not create registry " + url, e);
    69.             } else {
    70.                 // 1-11 Failed to obtain or create registry (service) object.
    71.                 LOGGER.warn(REGISTRY_FAILED_CREATE_INSTANCE, "", "",
    72.                     "Failed to obtain or create registry ", e);
    73.             }
    74.         } finally {
    75.             // Release the lock
    76.             registryManager.getRegistryLock().unlock();
    77.         }
    78.  
    79.         return registry;
    80.     }
    81.  
    82.  
    83.     /**
    84.      * ZookeeperRegistryFactory 的 createRegistry
    85.      * 其中 ZookeeperRegistryFactory 为 AbstractRegistryFactory 的子类
    86.      */
    87.     @Override
    88.     public Registry createRegistry(URL url) {
    89.         return new ZookeeperRegistry(url, zookeeperTransporter);
    90.     }
    91.
    2.2.4.2 注册服务

    通过register方法将当前服务注册到服务注册中心。org.apache.dubbo.registry.integration.RegistryProtocol#register 实现如下所示。

    1.     private static void register(Registry registry, URL registeredProviderUrl) {
    2.         ApplicationDeployer deployer = registeredProviderUrl.getOrDefaultApplicationModel().getDeployer();
    3.         try {
    4.             deployer.increaseServiceRefreshCount();
    5.             String registryName = Optional.ofNullable(registry.getUrl())
    6.                 .map(u -> u.getParameter(RegistryConstants.REGISTRY_CLUSTER_KEY,
    7.                     UrlUtils.isServiceDiscoveryURL(u) ? u.getParameter(REGISTRY_KEY) : u.getProtocol()))
    8.                 .filter(StringUtils::isNotEmpty)
    9.                 .orElse("unknown");
    10.             MetricsEventBus.post(RegistryEvent.toRsEvent(registeredProviderUrl.getApplicationModel(), registeredProviderUrl.getServiceKey(), 1, Collections.singletonList(registryName)),
    11.                 () -> {
    12.                     // 注册服务
    13.                     registry.register(registeredProviderUrl);
    14.                     return null;
    15.                 });
    16.         } finally {
    17.             deployer.decreaseServiceRefreshCount();
    18.         }
    19.     }

    以ZooKeeper注册中心为例。将会调用ZookeeperRegistry的父类的方法

    org.apache.dubbo.registry.support.FailbackRegistry#register，在该方法中将调用ZookeeperRegistry的doRegister()方法，最终将调用ZookeeperClient的create()方法创建Zookeeper节点，将服务注册到Zookeeper中。具体实现如下。

    1.     public void register(URL url) {
    2.         if (!acceptable(url)) {
    3.             logger.info("URL " + url + " will not be registered to Registry. Registry " + this.getUrl() + " does not accept service of this protocol type.");
    4.             return;
    5.         }
    6.         super.register(url);
    7.         removeFailedRegistered(url);
    8.         removeFailedUnregistered(url);
    9.         try {
    10.             // 注册服务
    11.             // Sending a registration request to the server side
    12.             doRegister(url);
    13.         } catch (Exception e) {
    14.             Throwable t = e;
    15.  
    16.             // If the startup detection is opened, the Exception is thrown directly.
    17.             boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
    18.                 && url.getParameter(Constants.CHECK_KEY, true)
    19.                 && (url.getPort() != 0);
    20.             boolean skipFailback = t instanceof SkipFailbackWrapperException;
    21.             if (check || skipFailback) {
    22.                 if (skipFailback) {
    23.                     t = t.getCause();
    24.                 }
    25.                 throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
    26.             } else {
    27.                 logger.error(INTERNAL_ERROR, "unknown error in registry module", "", "Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
    28.             }
    29.  
    30.             // Record a failed registration request to a failed list, retry regularly
    31.             addFailedRegistered(url);
    32.         }
    33.     }
    34.  
    35.  
    36.     public void doRegister(URL url) {
    37.         try {
    38.             checkDestroyed();
    39.             zkClient.create(toUrlPath(url), url.getParameter(DYNAMIC_KEY, true), true);
    40.         } catch (Throwable e) {
    41.             throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
    42.         }
    43.     }

    其中ZookeeperClient的create()方法的代码如下所示。

    1.     /**
    2.      * Create ZNode in Zookeeper.
    3.      *
    4.      * @param path path to ZNode
    5.      * @param ephemeral specify create mode of ZNode creation. true - EPHEMERAL, false - PERSISTENT.
    6.      * @param faultTolerant specify fault tolerance of ZNode creation.
    7.      *                       true - ignore exception and recreate if is ephemeral, false - throw exception.
    8.      */
    9.     void create(String path, boolean ephemeral, boolean faultTolerant);
    10.  
    11.  
    12.     public void create(String path, boolean ephemeral, boolean faultTolerant) {
    13.         if (!ephemeral) {
    14.             if (persistentExistNodePath.contains(path)) {
    15.                 return;
    16.             }
    17.             if (checkExists(path)) {
    18.                 persistentExistNodePath.add(path);
    19.                 return;
    20.             }
    21.         }
    22.         int i = path.lastIndexOf('/');
    23.         if (i > 0) {
    24.             create(path.substring(0, i), false, true);
    25.         }
    26.         if (ephemeral) {
    27.             createEphemeral(path, faultTolerant);
    28.         } else {
    29.             createPersistent(path, faultTolerant);
    30.             persistentExistNodePath.add(path);
    31.         }
    32.     }

    create()方法的伪代码和打印结果如下所示。

    1.     public static void main(String[] args) {
    2.         // 原始url
    3.         String path = "/dubbo/com.hn.TestService/providers/dubbo://192.168.0.1:20880/com.hn.TestService?anyhost=true&default.dynamic=false&dubbo=3.2";
    4.         // toUrlPath() 转化后
    5.         path = "/dubbo/com.hn.TestService/providers/dubbo%3A%2F%2F192.168.0.1%3A20880%2Fcom.hn.TestService%3Fanyhost%3Dtrue%26default.dynamic%3Dfalse%26dubbo%3D3.2";
    6.         create(path, true, true);
    7.     }
    8.  
    9.     public static void create(String path, boolean ephemeral, boolean faultTolerant) {
    10.         int i = path.lastIndexOf('/');
    11.         if (i > 0) {
    12.             create(path.substring(0, i), false, true);
    13.         }
    14.         if (ephemeral) {
    15.             createEphemeral(path, faultTolerant);
    16.         } else {
    17.             createPersistent(path, faultTolerant);
    18.         }
    19.     }
    20.  
    21.     private static void createEphemeral(String path, boolean faultTolerant) {
    22.         System.out.println("createEphemeral:" + path);
    23.     }
    24.  
    25.     private static void createPersistent(String path, boolean faultTolerant) {
    26.         System.out.println("createPersistent:" + path);
    27.     }

    打印结果如下所示。

    1. createPersistent:/dubbo
    2. createPersistent:/dubbo/com.hn.TestService
    3. createPersistent:/dubbo/com.hn.TestService/providers
    4. createEphemeral:/dubbo/com.hn.TestService/providers/dubbo%3A%2F%2F192.168.0.1%3A20880%2Fcom.hn.TestService%3Fanyhost%3Dtrue%26default.dynamic%3Dfalse%26dubbo%3D3.2

    create()方法首先执行递归部分，调用createPersistent()方法依次创建ZooKeeper节点 /dubbo、/dubbo/com.hn.TestService 和 /dubbo/com.hn.TestService/providers；

    然后执行递归之后的部分，调用createEphemeral() 方法创建 

    /dubbo/com.hn.TestService/providers/dubbo%3A%2F%2F192.168.0.1%3A20880%2Fcom.hn.TestService%3Fanyhost%3Dtrue%26default.dynamic%3Dfalse%26dubbo%3D3.2。

    2.2.4.3 ZooKeeper服务端的树状结构图

    服务注册到ZooKeeper之后，ZooKeeper服务端的树状结构图如下所示。

    3 Dubbo服务消费端启动过程

    3.1 Dubbo服务消费端启动过程概述

    • ReferenceConfig通过Protocol将远程服务转换为Invoker。在创建Invoker的同时会创建NettyClient对象，使服务消费端与服务提供端建立TCP连接。
    • 然后通过ProxyFactory将Invoker转换为客户端（即消费端）需要的接口。

    3.2 源码剖析

    3.2.1 服务消费端启动入口

    服务消费端启动入口为org.apache.dubbo.config.ReferenceConfigBase#get()方法，其中ReferenceConfigBase为ReferenceConfig的父类，通过此方法创建一个对服务提供方的远程调用代理。代码如下所示。

    1.     @Transient
    2.     public final T get() {
    3.         return get(true);
    4.     }
    5.  
    6.  
    7.      public T get(boolean check) {
    8.         if (destroyed) {
    9.             throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
    10.         }
    11.  
    12.         if (ref == null) {
    13.             if (getScopeModel().isLifeCycleManagedExternally()) {
    14.                 // prepare model for reference
    15.                 getScopeModel().getDeployer().prepare();
    16.             } else {
    17.                 // ensure start module, compatible with old api usage
    18.                 getScopeModel().getDeployer().start();
    19.             }
    20.  
    21.             init(check);
    22.         }
    23.  
    24.         return ref;
    25.     }

    3.2.2 将远程服务转换为Invoker

    上述get()方法将调用init()方法，然后调用createProxy() 方法来创建远程服务代理类，将远程服务转换为客户端需要的接口。

    具体而言，首先通过Protocol将远程服务转换为Invoker；然后通过ProxyFactory将Invoker转换为客户端需要的接口。具体代码如下所示。

    1.     protected synchronized void init(boolean check) {
    2.         // ...
    3.  
    4.         Map referenceParameters = appendConfig();
    5.  
    6.         ModuleServiceRepository repository = getScopeModel().getServiceRepository();
    7.         ServiceDescriptor serviceDescriptor;
    8.         if (CommonConstants.NATIVE_STUB.equals(getProxy())) {
    9.             serviceDescriptor = StubSuppliers.getServiceDescriptor(interfaceName);
    10.             repository.registerService(serviceDescriptor);
    11.         } else {
    12.             serviceDescriptor = repository.registerService(interfaceClass);
    13.         }
    14.         consumerModel = new ConsumerModel(serviceMetadata.getServiceKey(), proxy, serviceDescriptor,
    15.             getScopeModel(), serviceMetadata, createAsyncMethodInfo(), interfaceClassLoader);
    16.  
    17.         // Compatible with dependencies on ServiceModel#getReferenceConfig() , and will be removed in a future version.
    18.         consumerModel.setConfig(this);
    19.  
    20.         repository.registerConsumer(consumerModel);
    21.  
    22.         serviceMetadata.getAttachments().putAll(referenceParameters);
    23.  
    24.         // 创建远程服务代理类
    25.         ref = createProxy(referenceParameters);
    26.  
    27.         // ...
    28.     }
    29.  
    30.  
    31.     private T createProxy(Map referenceParameters) {
    32.         urls.clear();
    33.  
    34.         meshModeHandleUrl(referenceParameters);
    35.  
    36.         if (StringUtils.isNotEmpty(url)) {
    37.             // user specified URL, could be peer-to-peer address, or register center's address.
    38.             parseUrl(referenceParameters);
    39.         } else {
    40.             // if protocols not in jvm checkRegistry
    41.             aggregateUrlFromRegistry(referenceParameters);
    42.         }
    43.  
    44.         // 1、通过Protocol将远程服务转换为Invoker
    45.         createInvoker();
    46.  
    47.         if (logger.isInfoEnabled()) {
    48.             logger.info("Referred dubbo service: [" + referenceParameters.get(INTERFACE_KEY) + "]." +
    49.                     (ProtocolUtils.isGeneric(referenceParameters.get(GENERIC_KEY)) ?
    50.                             " it's GenericService reference" : " it's not GenericService reference"));
    51.         }
    52.  
    53.         URL consumerUrl = new ServiceConfigURL(CONSUMER_PROTOCOL, referenceParameters.get(REGISTER_IP_KEY), 0,
    54.                 referenceParameters.get(INTERFACE_KEY), referenceParameters);
    55.         consumerUrl = consumerUrl.setScopeModel(getScopeModel());
    56.         consumerUrl = consumerUrl.setServiceModel(consumerModel);
    57.         MetadataUtils.publishServiceDefinition(consumerUrl, consumerModel.getServiceModel(), getApplicationModel());
    58.  
    59.         // 通过ProxyFactory将Invoker转换为客户端需要的接口
    60.         // create service proxy
    61.         return (T) proxyFactory.getProxy(invoker, ProtocolUtils.isGeneric(generic));
    62.     }
    63.

    在createInvoker() 方法中，因为服务注册和发现使用是register协议，因此将调用 RegistryProtocol#refer方法创建invoker。代码如下所示。

    1.     private void createInvoker() {
    2.         if (urls.size() == 1) {
    3.             URL curUrl = urls.get(0);
    4.             // urls - 消费端准备消费的远程服务
    5.             // 将将远程服务转换为Invoker
    6.             invoker = protocolSPI.refer(interfaceClass, curUrl);
    7.             // registry url, mesh-enable and unloadClusterRelated is true, not need Cluster.
    8.             if (!UrlUtils.isRegistry(curUrl) &&
    9.                     !curUrl.getParameter(UNLOAD_CLUSTER_RELATED, false)) {
    10.                 List> invokers = new ArrayList<>();
    11.                 invokers.add(invoker);
    12.                 invoker = Cluster.getCluster(getScopeModel(), Cluster.DEFAULT).join(new StaticDirectory(curUrl, invokers), true);
    13.             }
    14.         } else {
    15.             List> invokers = new ArrayList<>();
    16.             URL registryUrl = null;
    17.             for (URL url : urls) {
    18.                 // For multi-registry scenarios, it is not checked whether each referInvoker is available.
    19.                 // Because this invoker may become available later.
    20.                 invokers.add(protocolSPI.refer(interfaceClass, url));
    21.  
    22.                 if (UrlUtils.isRegistry(url)) {
    23.                     // use last registry url
    24.                     registryUrl = url;
    25.                 }
    26.             }
    27.  
    28.             if (registryUrl != null) {
    29.                 // registry url is available
    30.                 // for multi-subscription scenario, use 'zone-aware' policy by default
    31.                 String cluster = registryUrl.getParameter(CLUSTER_KEY, ZoneAwareCluster.NAME);
    32.                 // The invoker wrap sequence would be: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker
    33.                 // (RegistryDirectory, routing happens here) -> Invoker
    34.                 invoker = Cluster.getCluster(registryUrl.getScopeModel(), cluster, false).join(new StaticDirectory(registryUrl, invokers), false);
    35.             } else {
    36.                 // not a registry url, must be direct invoke.
    37.                 if (CollectionUtils.isEmpty(invokers)) {
    38.                     throw new IllegalArgumentException("invokers == null");
    39.                 }
    40.                 URL curUrl = invokers.get(0).getUrl();
    41.                 String cluster = curUrl.getParameter(CLUSTER_KEY, Cluster.DEFAULT);
    42.                 invoker = Cluster.getCluster(getScopeModel(), cluster).join(new StaticDirectory(curUrl, invokers), true);
    43.             }
    44.         }
    45.     }
    46.  
    47.  
    48.     public  Invoker refer(Class type, URL url) throws RpcException {
    49.         url = getRegistryUrl(url);
    50.         Registry registry = getRegistry(url);
    51.         if (RegistryService.class.equals(type)) {
    52.             return proxyFactory.getInvoker((T) registry, type, url);
    53.         }
    54.  
    55.         // group="a,b" or group="*"
    56.         Map qs = (Map) url.getAttribute(REFER_KEY);
    57.         String group = qs.get(GROUP_KEY);
    58.         if (StringUtils.isNotEmpty(group)) {
    59.             if ((COMMA_SPLIT_PATTERN.split(group)).length > 1 || "*".equals(group)) {
    60.                 return doRefer(Cluster.getCluster(url.getScopeModel(), MergeableCluster.NAME), registry, type, url, qs);
    61.             }
    62.         }
    63.  
    64.         Cluster cluster = Cluster.getCluster(url.getScopeModel(), qs.get(CLUSTER_KEY));
    65.  
    66.         // 创建invoker
    67.         return doRefer(cluster, registry, type, url, qs);
    68.     }
    69.  
    70.  
    71.     protected  Invoker doRefer(Cluster cluster, Registry registry, Class type, URL url, Map parameters) {
    72.         // ...
    73.  
    74.         // 创建invoker
    75.         ClusterInvoker migrationInvoker = getMigrationInvoker(this, cluster, registry, type, url, consumerUrl);
    76.         return interceptInvoker(migrationInvoker, url, consumerUrl);
    77.     }
    78.

    因为默认的rpc协议为dubbo协议，通过interceptInvoker()方法，最终将调用DubboProtocol的方法创建invoker。创建代码如下所示。

    1. // InterfaceCompatibleRegistryProtocol.getInvoker
    2. public  ClusterInvoker getInvoker(Cluster cluster, Registry registry, Class type, URL url) {
    3.     DynamicDirectory directory = new RegistryDirectory<>(type, url);
    4.     return doCreateInvoker(directory, cluster, registry, type);
    5. }
    6.  
    7. // RegistryProtocol#doCreateInvoker
    8. protected  ClusterInvoker doCreateInvoker(DynamicDirectory directory, Cluster cluster, Registry registry, Class type) {
    9. 	directory.setRegistry(registry);
    10. 	directory.setProtocol(protocol);
    11. 	// all attributes of REFER_KEY
    12. 	Map parameters = new HashMap(directory.getConsumerUrl().getParameters());
    13. 	URL urlToRegistry = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
    14. 	if (directory.isShouldRegister()) {
    15. 		directory.setRegisteredConsumerUrl(urlToRegistry);
    16. 		registry.register(directory.getRegisteredConsumerUrl());
    17. 	}
    18. 	
    19. 	// 1、建立路由规则链
    20. 	directory.buildRouterChain(urlToRegistry);
    21. 	
    22. 	// 2、订阅服务提供者地址，生成invoker
    23. 	directory.subscribe(toSubscribeUrl(urlToRegistry));
    24.  
    25. 	// 3、包装机器容错策略到invoker
    26. 	return (ClusterInvoker) cluster.join(directory);
    27. }
    28.  
    29. // org.apache.dubbo.rpc.protocol.AbstractProtocol#refer
    30. public  Invoker refer(Class type, URL url) throws RpcException {
    31. 	return new AsyncToSyncInvoker<>(protocolBindingRefer(type, url));
    32. }
    33.  
    34. // org.apache.dubbo.rpc.protocol.dubbo.DubboProtocol#protocolBindingRefer
    35. public  Invoker protocolBindingRefer(Class serviceType, URL url) throws RpcException {
    36. 	optimizeSerialization(url);
    37.  
    38. 	// create rpc invoker.
    39. 	DubboInvoker invoker = new DubboInvoker(serviceType, url, getClients(url), invokers);
    40. 	invokers.add(invoker);
    41.  
    42. 	return invoker;
    43. }

    3.2.3 创建并启动NettyClient

    在创建Invoker的同时，通过getClients() 方法创建并启动服务消费端的NettyClient对象，与服务提供端建立TCP连接。

    （1）每个服务消费端与服务提供端集群中的所有机器都建立了连接，因为服务消费端会将所有服务提供者的URL转化为Invoker。

    （2）默认情况下，针对一个服务提供者机器只创建一个NettyClient对象。即服务消费端引用同一个服务提供者机器上的多个服务时，这些服务复用同一个Netty连接。代码如下所示。

    1.     private ClientsProvider getClients(URL url) {
    2.         int connections = url.getParameter(CONNECTIONS_KEY, 0);
    3.         // whether to share connection
    4.         // if not configured, connection is shared, otherwise, one connection for one service
    5.         // 默认情况下，针对一个服务提供者机器只创建一个NettyClient对象，
    6.         // 即服务消费端引用同一个服务提供者机器上的多个服务时，这些服务复用同一个Netty连接
    7.         if (connections == 0) {
    8.             /*
    9.              * The xml configuration should have a higher priority than properties.
    10.              */
    11.             String shareConnectionsStr = StringUtils.isBlank(url.getParameter(SHARE_CONNECTIONS_KEY, (String) null))
    12.                 ? ConfigurationUtils.getProperty(url.getOrDefaultApplicationModel(), SHARE_CONNECTIONS_KEY, DEFAULT_SHARE_CONNECTIONS)
    13.                 : url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
    14.             connections = Integer.parseInt(shareConnectionsStr);
    15.  
    16.             return getSharedClient(url, connections);
    17.         }
    18.  
    19.         List clients = IntStream.range(0, connections)
    20.             .mapToObj((i) -> initClient(url))
    21.             .collect(Collectors.toList());
    22.         return new ExclusiveClientsProvider(clients);
    23.     }
    24.  
    25.  
    26.     private SharedClientsProvider getSharedClient(URL url, int connectNum) {
    27.         String key = url.getAddress();
    28.  
    29.         // connectNum must be greater than or equal to 1
    30.         int expectedConnectNum = Math.max(connectNum, 1);
    31.         return referenceClientMap.compute(key, (originKey, originValue) -> {
    32.             if (originValue != null && originValue.increaseCount()) {
    33.                 return originValue;
    34.             } else {
    35.                 return new SharedClientsProvider(this, originKey, buildReferenceCountExchangeClientList(url, expectedConnectNum));
    36.             }
    37.         });
    38.     }
    39.  
    40.     private List buildReferenceCountExchangeClientList(URL url, int connectNum) {
    41.         List clients = new ArrayList<>();
    42.  
    43.         for (int i = 0; i < connectNum; i++) {
    44.             clients.add(buildReferenceCountExchangeClient(url));
    45.         }
    46.  
    47.         return clients;
    48.     }
    49.  
    50.  
    51.     /**
    52.      * Build a single client
    53.      *
    54.      * @param url
    55.      * @return
    56.      */
    57.     private ReferenceCountExchangeClient buildReferenceCountExchangeClient(URL url) {
    58.         ExchangeClient exchangeClient = initClient(url);
    59.         ReferenceCountExchangeClient client = new ReferenceCountExchangeClient(exchangeClient, DubboCodec.NAME);
    60.         // read configs
    61.         int shutdownTimeout = ConfigurationUtils.getServerShutdownTimeout(url.getScopeModel());
    62.         client.setShutdownWaitTime(shutdownTimeout);
    63.         return client;
    64.     }

    其中创建NettyClient的关键方法为initClient()方法。具体如下所示。

    1.     /**
    2.      * Create new connection
    3.      *
    4.      * @param url
    5.      */
    6.     private ExchangeClient initClient(URL url) {
    7.         /*
    8.          * Instance of url is InstanceAddressURL, so addParameter actually adds parameters into ServiceInstance,
    9.          * which means params are shared among different services. Since client is shared among services this is currently not a problem.
    10.          */
    11.         String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));
    12.  
    13.         // BIO is not allowed since it has severe performance issue.
    14.         if (StringUtils.isNotEmpty(str) && !url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).hasExtension(str)) {
    15.             throw new RpcException("Unsupported client type: " + str + "," +
    16.                 " supported client type is " + StringUtils.join(url.getOrDefaultFrameworkModel().getExtensionLoader(Transporter.class).getSupportedExtensions(), " "));
    17.         }
    18.  
    19.         try {
    20.             ScopeModel scopeModel = url.getScopeModel();
    21.             int heartbeat = UrlUtils.getHeartbeat(url);
    22.             // Replace InstanceAddressURL with ServiceConfigURL.
    23.             url = new ServiceConfigURL(DubboCodec.NAME, url.getUsername(), url.getPassword(), url.getHost(), url.getPort(), url.getPath(), url.getAllParameters());
    24.             url = url.addParameter(CODEC_KEY, DubboCodec.NAME);
    25.             // enable heartbeat by default
    26.             url = url.addParameterIfAbsent(HEARTBEAT_KEY, Integer.toString(heartbeat));
    27.             url = url.setScopeModel(scopeModel);
    28.  
    29.             // connection should be lazy
    30.             return url.getParameter(LAZY_CONNECT_KEY, false)
    31.                 ? new LazyConnectExchangeClient(url, requestHandler)
    32.                 : Exchangers.connect(url, requestHandler);
    33.         } catch (RemotingException e) {
    34.             throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
    35.         }
    36.     }

    （3）当消费端启动时就与服务提供端建立了TCP连接。

    由initClient()方法实现可知，默认情况下，LAZY_CONNECT_KEY为false，即此处是创建及时连接 —— Exchangers.connect(url, requestHandler)。

     Exchangers.connect() 之后的主要调用链路为：Exchangers的connect()->HeaderExchanger的connect()->Transporters的connect()->NettyTransporter的connect()，最终通过NettyTransporter的connect()创建NettyClient。具体代码如下所示。

    1.     public static ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
    2.         if (url == null) {
    3.             throw new IllegalArgumentException("url == null");
    4.         }
    5.         if (handler == null) {
    6.             throw new IllegalArgumentException("handler == null");
    7.         }
    8.         return getExchanger(url).connect(url, handler);
    9.     }
    10.  
    11.  
    12.     public ExchangeClient connect(URL url, ExchangeHandler handler) throws RemotingException {
    13.         return new HeaderExchangeClient(Transporters.connect(url, new DecodeHandler(new HeaderExchangeHandler(handler))), true);
    14.     }
    15.  
    16.     public static Client connect(URL url, ChannelHandler... handlers) throws RemotingException {
    17.         if (url == null) {
    18.             throw new IllegalArgumentException("url == null");
    19.         }
    20.         ChannelHandler handler;
    21.         if (handlers == null || handlers.length == 0) {
    22.             handler = new ChannelHandlerAdapter();
    23.         } else if (handlers.length == 1) {
    24.             handler = handlers[0];
    25.         } else {
    26.             handler = new ChannelHandlerDispatcher(handlers);
    27.         }
    28.         return getTransporter(url).connect(url, handler);
    29.     }
    30.  
    31.     public Client connect(URL url, ChannelHandler handler) throws RemotingException {
    32.         return new NettyClient(url, handler);
    33.     }
    34.  
    35.

    （4）创建并启动服务消费端的NettyClient对象，并与服务提供端建立TCP连接

    调用父类 AbstractClient 的构造函数，创建并启动服务消费端的NettyClient对象，并与服务提供端建立TCP连接。

    1.     public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
    2.         super(url, wrapChannelHandler(url, handler));
    3.     }
    4.  
    5.     public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
    6.         super(url, handler);
    7.         // set default needReconnect true when channel is not connected
    8.         needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, true);
    9.  
    10.         initExecutor(url);
    11.  
    12.         try {
    13.             // 1、启动 NettyClient
    14.             doOpen();
    15.         } catch (Throwable t) {
    16.             close();
    17.             throw new RemotingException(url.toInetSocketAddress(), null,
    18.                 "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
    19.                     + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
    20.         }
    21.  
    22.         try {
    23.             // connect.
    24.             // 2、发起远程连接，与服务提供者建立TCP连接
    25.             connect();
    26.             if (logger.isInfoEnabled()) {
    27.                 logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress());
    28.             }
    29.         } catch (RemotingException t) {
    30.             // If lazy connect client fails to establish a connection, the client instance will still be created,
    31.             // and the reconnection will be initiated by ReconnectTask, so there is no need to throw an exception
    32.             if (url.getParameter(LAZY_CONNECT_KEY, false)) {
    33.                 logger.warn(TRANSPORT_FAILED_CONNECT_PROVIDER, "", "", "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() +
    34.                     " connect to the server " + getRemoteAddress() +
    35.                     " (the connection request is initiated by lazy connect client, ignore and retry later!), cause: " +
    36.                     t.getMessage(), t);
    37.                 return;
    38.             }
    39.  
    40.             if (url.getParameter(Constants.CHECK_KEY, true)) {
    41.                 close();
    42.                 throw t;
    43.             } else {
    44.                 logger.warn(TRANSPORT_FAILED_CONNECT_PROVIDER, "", "", "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
    45.                     + " connect to the server " + getRemoteAddress() + " (check == false, ignore and retry later!), cause: " + t.getMessage(), t);
    46.             }
    47.         } catch (Throwable t) {
    48.             close();
    49.             throw new RemotingException(url.toInetSocketAddress(), null,
    50.                 "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
    51.                     + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
    52.         }
    53.     }
    54.  
    55.  
    56.     /**
    57.      * Init bootstrap
    58.      *
    59.      * @throws Throwable
    60.      */
    61.     @Override
    62.     protected void doOpen() throws Throwable {
    63.         // 创建业务Handler
    64.         final NettyClientHandler nettyClientHandler = createNettyClientHandler();
    65.         // 创建启动器并配置
    66.         bootstrap = new Bootstrap();
    67.         initBootstrap(nettyClientHandler);
    68.     }
    69.  
    70.  
    71.     protected void connect() throws RemotingException {
    72.         connectLock.lock();
    73.  
    74.         try {
    75.             // ...
    76.             doConnect();
    77.             // ...
    78.         } catch (RemotingException e) {
    79.             throw e;
    80.  
    81.         } catch (Throwable e) {
    82.             throw new RemotingException(this, "Failed to connect to server " + getRemoteAddress() + " from " + getClass().getSimpleName() + " "
    83.                 + NetUtils.getLocalHost() + " using dubbo version " + Version.getVersion()
    84.                 + ", cause: " + e.getMessage(), e);
    85.  
    86.         } finally {
    87.             connectLock.unlock();
    88.         }
    89.     }
    90.  
    91.     private void doConnect(InetSocketAddress serverAddress) throws RemotingException {
    92.         long start = System.currentTimeMillis();
    93.         // 发起链接
    94.         ChannelFuture future = bootstrap.connect(serverAddress);
    95.         // ...
    96.     }

    3.2.4 将Invoker转换为客户端需要的接口（ref）

    通过ProxyFactory的getProxy()方法创建代理类，将Invoker转换为客户端需要的接口。以JavassistProxyFactory为例，当使用JavassistProxyFactory创建失败时，会使用JdkProxyFactory重试。代码如下所示。

    1.     public  T getProxy(Invoker invoker, Class[] interfaces) {
    2.         try {
    3.             return (T) Proxy.getProxy(interfaces).newInstance(new InvokerInvocationHandler(invoker));
    4.         } catch (Throwable fromJavassist) {
    5.             // try fall back to JDK proxy factory
    6.             try {
    7.                 T proxy = jdkProxyFactory.getProxy(invoker, interfaces);
    8.                 logger.error(PROXY_FAILED, "", "", "Failed to generate proxy by Javassist failed. Fallback to use JDK proxy success. " +
    9.                     "Interfaces: " + Arrays.toString(interfaces), fromJavassist);
    10.                 return proxy;
    11.             } catch (Throwable fromJdk) {
    12.                 logger.error(PROXY_FAILED, "", "", "Failed to generate proxy by Javassist failed. Fallback to use JDK proxy is also failed. " +
    13.                     "Interfaces: " + Arrays.toString(interfaces) + " Javassist Error.", fromJavassist);
    14.                 logger.error(PROXY_FAILED, "", "", "Failed to generate proxy by Javassist failed. Fallback to use JDK proxy is also failed. " +
    15.                     "Interfaces: " + Arrays.toString(interfaces) + " JDK Error.", fromJdk);
    16.                 throw fromJavassist;
    17.             }
    18.         }
    19.     }
  • 相关阅读:
    nestjs使用rabbitMQ
    STC89C52定时器/中断快速上手示例讲解
    黑马mysql教程笔记（mysql8教程）基础篇——数据库相关概念、mysql安装及卸载、数据模型、SQL通用语法及分类（DDL、DML、DQL、DCL）
    操作系统实验四：文件管理(2学时)
    如何通过DBeaver 连接 TDengine？
    int的存储和解释
    【每日一题】CF1680C. Binary String | 双指针 | 简单
    当老板一定要学会花钱去赚钱
    vue3异步加载js文件
    mysql为什么使用B+树
  • 原文地址：https://blog.csdn.net/J_bean/article/details/133894304