---

前言

文章基于3.1.0版本进行分析

		<dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo</artifactId>
            <version>3.1.0</version>
        </dependency>

---

一、RegistryDirectory

RegistryDirectory是Dubbo中的服务目录，服务目录中存储了一些和服务提供者有关的信息。通过服务目录，服务消费者可获取到服务提供者的信息，比如 ip、端口、服务协议等。
通过这些信息，服务消费者就可通过 Netty 等客户端进行远程调用。

创建RegistryDirectory会出初始化父类DynamicDirectory ，初始化接口、分组、失败策略等，从DynamicDirectory 开始分析

1、DynamicDirectory

创建服务目录与初始化的一些信息，比如接口、分组、失败策略

	public DynamicDirectory(Class<T> serviceType, URL url) {
        super(url, true);

        ModuleModel moduleModel = url.getOrDefaultModuleModel();

        // 容错适配器，Cluster$Adaptive 默认的容错机制是失效转移 failover
		this.cluster = moduleModel.getExtensionLoader(Cluster.class).getAdaptiveExtension();
        // 路由工厂适配器RouterFactory$Adaptive
		this.routerFactory = moduleModel.getExtensionLoader(RouterFactory.class).getAdaptiveExtension();

        if (serviceType == null) {
            throw new IllegalArgumentException("service type is null.");
        }

        if (StringUtils.isEmpty(url.getServiceKey())) {
            throw new IllegalArgumentException("registry serviceKey is null.");
        }

        this.shouldRegister = !ANY_VALUE.equals(url.getServiceInterface()) && url.getParameter(REGISTER_KEY, true);
        this.shouldSimplified = url.getParameter(SIMPLIFIED_KEY, false);

        // 接口类型 org.sjl.dubbo.AsyncProvider
		this.serviceType = serviceType;
		// 接口key org.sjl.dubbo.AsyncProvider
        this.serviceKey = super.getConsumerUrl().getServiceKey();

        this.directoryUrl = consumerUrl;
        String group = directoryUrl.getGroup("");
        this.multiGroup = group != null && (ANY_VALUE.equals(group) || group.contains(","));
		// 默认快速失败
        this.shouldFailFast = Boolean.parseBoolean(ConfigurationUtils.getProperty(moduleModel, Constants.SHOULD_FAIL_FAST_KEY, "true"));
    }

2、RegistryProtocol.doCreateInvoker

ClusterInvoker的创建过程

	protected <T> ClusterInvoker<T> doCreateInvoker(DynamicDirectory<T> directory, Cluster cluster, Registry registry, Class<T> type) {
        // 初始化服务目录的注册中心和协议
		directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY
        Map<String, String> parameters = new HashMap<>(directory.getConsumerUrl().getParameters());
		// urlToRegistry = consumer://192.168.0.101/org.sjl.dubbo.AsyncProvider?application=dubbo-springboot-start-consumer&background=false&dubbo=2.0.2
		// &interface=org.sjl.dubbo.AsyncProvider&methods=sayHiAsync,sayHello,sayHelloAsync&pid=15072&qos.enable=false&release=3.1.0
		// &side=consumer&sticky=false&timeout=30000&timestamp=1725462541548
		URL urlToRegistry = new ServiceConfigURL(
            parameters.get(PROTOCOL_KEY) == null ? CONSUMER : parameters.get(PROTOCOL_KEY),
            parameters.remove(REGISTER_IP_KEY),
            0,
            getPath(parameters, type),
            parameters
        );
        urlToRegistry = urlToRegistry.setScopeModel(directory.getConsumerUrl().getScopeModel());
        urlToRegistry = urlToRegistry.setServiceModel(directory.getConsumerUrl().getServiceModel());
        if (directory.isShouldRegister()) {
            directory.setRegisteredConsumerUrl(urlToRegistry);
			//  消费者配置到注册中心
            registry.register(directory.getRegisteredConsumerUrl());
        }
		// 路由调用链
        directory.buildRouterChain(urlToRegistry);
		// 服务发现并订阅的逻辑 核心逻辑
        directory.subscribe(toSubscribeUrl(urlToRegistry));

        return (ClusterInvoker<T>) cluster.join(directory, true);
    }

2、RegistryProtocol.subscribe

    public void subscribe(URL url) {
        setSubscribeUrl(url);
        registry.subscribe(url, this);
    }

3、ListenerRegistryWrapper.subscribe

进行包装，对订阅完成后的后置处理

	public void subscribe(URL url, NotifyListener listener) {
        try {
            if (registry != null) {
                registry.subscribe(url, listener);
            }
        } finally {
            if (CollectionUtils.isNotEmpty(listeners)) {
                RuntimeException exception = null;
                for (RegistryServiceListener registryListener : listeners) {
                    if (registryListener != null) {
                        try {
                            registryListener.onSubscribe(url, registry);
                        } catch (RuntimeException t) {
                            logger.error(t.getMessage(), t);
                            exception = t;
                        }
                    }
                }
                if (exception != null) {
                    throw exception;
                }
            }
        }
    }

4、FailbackRegistry.subscribe

failbackRegistry这个类，从名字就可以看出，它的主要作用就是实现具有故障恢复功能的服务订阅机制，简单来说就是如果在订阅服务注册中心时出现异常，会触发重试机制。

	public void subscribe(URL url, NotifyListener listener) {
        super.subscribe(url, listener);
		// 移除失效的监听
        removeFailedSubscribed(url, listener);
        try {
            // Sending a subscription request to the server side
			// 发送订阅到服务端
            doSubscribe(url, listener);
        } catch (Exception e) {
            Throwable t = e;

            // *** 异常处理

            // Record a failed registration request to a failed list, retry regularly
            addFailedSubscribed(url, listener);
        }
    }

5、ZookeeperRegistry.doSubscribe

所有Service层发起订阅，类似于监控中心发起的订阅
指定的Service层发起的订阅，服务消费者的订阅

	@Override
    public void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            checkDestroyed();
			// 针对通用服务接口
            if (ANY_VALUE.equals(url.getServiceInterface())) {
                // ***
            }
			// 针对指定服务，正常走到这里
			else {
                CountDownLatch latch = new CountDownLatch(1);

                try {
                    List<URL> urls = new ArrayList<>();

                    // ["/dubbo/org.sjl.dubbo.GreetingsProvider/providers", ".../configurators", ".../routers"]
					// 需要订阅以上三个节点下数据变化
					for (String path : toCategoriesPath(url)) {
						// 构建一个listeners集合，其中key是NotifyListener，
						// value表示针对这个RegistryDirectory注册的子节点监听。
                        ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
                        ChildListener zkListener = listeners.computeIfAbsent(listener, k -> new RegistryChildListenerImpl(url, k, latch));

                        if (zkListener instanceof RegistryChildListenerImpl) {
                            ((RegistryChildListenerImpl) zkListener).setLatch(latch);
                        }

                        // 创建非临时节点，不存在时候会创建 
						// /dubbo/org.sjl.dubbo.GreetingsProvider/providers
                        // /dubbo/org.sjl.dubbo.GreetingsProvider/configurators
						// /dubbo/org.sjl.dubbo.GreetingsProvider/routers
						zkClient.create(path, false);
						// 服务目录创建完毕之后创建一个监听器用来监听子目录，同时要返回一个path目录的子子节点，
						// 比如providers下面的提供者节点列表，如果有多个可以返回多个
                        List<String> children = zkClient.addChildListener(path, zkListener);
                        if (children != null) {
                            urls.addAll(toUrlsWithEmpty(url, path, children));
                        }
                    }

                    // 调用notify方法触发回调通知
					notify(url, listener, urls);
                } finally {
                    // tells the listener to run only after the sync notification of main thread finishes.
                    latch.countDown();
                }
            }
        } catch (Throwable e) {
            throw new RpcException("Failed to subscribe " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

children里面值实例，包括ip、port、接口等参数

dubbo://192.168.0.101:20880/org.sjl.dubbo.GreetingsProvider?anyhost=true
&application=dubbo-springboot-start-provider
&background=false
&deprecated=false
&dubbo=2.0.2
&dynamic=true
&generic=false
&interface=org.sjl.dubbo.GreetingsProvider
&methods=sayHello,sayHi
&pid=30316&release=3.1.0
&service-name-mapping=true
&side=provider&timeout=30000
&timestamp=1725293330893

notify回调通知，链路结构为

FailbackRegistry -> AbstractRegistry -> RegistryDirectory

6、RegistryDirectory.notify

服务地址发生变化和更新时，调用Directory.notify来更新，意味着更新后的信息，会同步到Directory

	public synchronized void notify(List<URL> urls) {
        if (isDestroyed()) {
            return;
        }
		// 过滤不需要的数据
        Map<String, List<URL>> categoryUrls = urls.stream()
            .filter(Objects::nonNull)
            .filter(this::isValidCategory)
            .filter(this::isNotCompatibleFor26x)
            .collect(Collectors.groupingBy(this::judgeCategory));

        // 假设当前进来的通知是 providers节点
		// 判断configurator是否为空，这个节点下的配置，是在dubbo-admin控制台上修改配置时，会先创建一个配置节点到这个路径下，
		// 注册中心收到这个变化时会通知服务消费者，服务消费者会根据新的配置重新构建Invoker
		List<URL> configuratorURLs = categoryUrls.getOrDefault(CONFIGURATORS_CATEGORY, Collections.emptyList());
        this.configurators = Configurator.toConfigurators(configuratorURLs).orElse(this.configurators);

        // 判断路由规则配置是否为空，如果不为空，同样将路由规则添加到url中。
		List<URL> routerURLs = categoryUrls.getOrDefault(ROUTERS_CATEGORY, Collections.emptyList());
        toRouters(routerURLs).ifPresent(this::addRouters);

        // providers
		// 得到服务提供者的地址列表
        List<URL> providerURLs = categoryUrls.getOrDefault(PROVIDERS_CATEGORY, Collections.emptyList());

        // 3.x added for extend URL address
        ExtensionLoader<AddressListener> addressListenerExtensionLoader = getUrl().getOrDefaultModuleModel().getExtensionLoader(AddressListener.class);
        List<AddressListener> supportedListeners = addressListenerExtensionLoader.getActivateExtension(getUrl(), (String[]) null);
        if (supportedListeners != null && !supportedListeners.isEmpty()) {
            for (AddressListener addressListener : supportedListeners) {
                providerURLs = addressListener.notify(providerURLs, getConsumerUrl(), this);
            }
        }
		// 刷新并覆盖注册中心里面的配置、url、Invoker，
        refreshOverrideAndInvoker(providerURLs);
    }

7、RegistryDirectory.refreshOverrideAndInvoker

路由调用refreshInvoker方法

当注册中心的服务地址发生变化时，会触发更新。而更新之后并不是直接把url地址存储到内存，而是把
url转化为invoker进行存储，这个invoker是作为通信的调用器来构建的领域对象，所以如果地址发生变
化，那么需要把老的invoker销毁，然后用心的invoker替代。

    private synchronized void refreshOverrideAndInvoker(List<URL> urls) {
        // mock zookeeper://xxx?mock=return null
        refreshInvoker(urls);
    }
	
	private void refreshInvoker(List<URL> invokerUrls) {
        Assert.notNull(invokerUrls, "invokerUrls should not be null");
		// 如果只有一个服务提供者，并且如果是空协议，那么这个时候直接返回进制访问，并且销毁所有的invokers
        if (invokerUrls.size() == 1
            && invokerUrls.get(0) != null
            && EMPTY_PROTOCOL.equals(invokerUrls.get(0).getProtocol())) {
            this.forbidden = true; // Forbid to access
            routerChain.setInvokers(BitList.emptyList());
            destroyAllInvokers(); // Close all invokers
        } else {
			this.forbidden = false; // Allow to access

            //  第一次初始化的时候，invokerUrls为空。
			if (invokerUrls == Collections.<URL>emptyList()) {
                invokerUrls = new ArrayList<>();
            }
            // use local reference to avoid NPE as this.cachedInvokerUrls will be set null by destroyAllInvokers().
            // 获取invoker缓存
			Set<URL> localCachedInvokerUrls = this.cachedInvokerUrls;
            if (invokerUrls.isEmpty() && localCachedInvokerUrls != null) {

                // 1-4 Empty address.
                logger.warn("1-4", "configuration ", "",
                    "Service" + serviceKey + " received empty address list with no EMPTY protocol set, trigger empty protection.");

                invokerUrls.addAll(localCachedInvokerUrls);

            } else {
                localCachedInvokerUrls = new HashSet<>();
                localCachedInvokerUrls.addAll(invokerUrls);//Cached invoker urls, convenient for comparison
                this.cachedInvokerUrls = localCachedInvokerUrls;
            }
            if (invokerUrls.isEmpty()) {
                return;
            }

            // use local reference to avoid NPE as this.urlInvokerMap will be set null concurrently at destroyAllInvokers().
            Map<URL, Invoker<T>> localUrlInvokerMap = this.urlInvokerMap;
            // can't use local reference as oldUrlInvokerMap's mappings might be removed directly at toInvokers().
            Map<URL, Invoker<T>> oldUrlInvokerMap = null;
            if (localUrlInvokerMap != null) {
                // the initial capacity should be set greater than the maximum number of entries divided by the load factor to avoid resizing.
                oldUrlInvokerMap = new LinkedHashMap<>(Math.round(1 + localUrlInvokerMap.size() / DEFAULT_HASHMAP_LOAD_FACTOR));
                localUrlInvokerMap.forEach(oldUrlInvokerMap::put);
            }
			// 将URL转换为Invoker  这里会做一些协议的指定过滤操作
            Map<URL, Invoker<T>> newUrlInvokerMap = toInvokers(oldUrlInvokerMap, invokerUrls);// Translate url list to Invoker map

            /*
             * If the calculation is wrong, it is not processed.
             *
             * 1. The protocol configured by the client is inconsistent with the protocol of the server.
             *    eg: consumer protocol = dubbo, provider only has other protocol services(rest).
             * 2. The registration center is not robust and pushes illegal specification data.
             *
             */
            if (CollectionUtils.isEmptyMap(newUrlInvokerMap)) {

                // 3-1 - Failed to convert the URL address into Invokers.

                logger.error(
                    "3-1", "inconsistency between the client protocol and the protocol of the server",
                    "", "urls to invokers error",
                    new IllegalStateException(
                        "urls to invokers error. invokerUrls.size :" +
                            invokerUrls.size() + ", invoker.size :0. urls :" + invokerUrls.toString()));

                return;
            }

            List<Invoker<T>> newInvokers = Collections.unmodifiableList(new ArrayList<>(newUrlInvokerMap.values()));
            this.setInvokers(multiGroup ? new BitList<>(toMergeInvokerList(newInvokers)) : new BitList<>(newInvokers));
            // pre-route and build cache
            routerChain.setInvokers(this.getInvokers());
            this.urlInvokerMap = newUrlInvokerMap;

            try {
				// 销毁不使用或者已经被删除的 Invoker。例如某个提供者的代码删除了
                destroyUnusedInvokers(oldUrlInvokerMap, newUrlInvokerMap); // Close the unused Invoker
            } catch (Exception e) {
                logger.warn("destroyUnusedInvokers error. ", e);
            }

            // notify invokers refreshed
            this.invokersChanged();
        }
    }

将invokerURL列表转换为调用器映射。转换规则如下：

• 如果URL已转换为invoker，它将不再被重新引用并直接从缓存中获取，请注意，URL中的任何参数更改都将被重新引用。
• 如果传入调用器列表不为空，则表示它是最新的调用器列表。
• 如果传入invokerUrl的列表为空，则意味着该规则只是一个覆盖规则或路由规则，需要重新对比以决定是否重新引用。

8、RegistryDirectory.toInvokers

协议过滤，配置合并都会在这里面去做。url转换为调用程序，如果url已被引用，则不会重新引用。将放入newUrlInvokeMap的项将从OldUrlInvoceMap中删除。

    private Map<URL, Invoker<T>> toInvokers(Map<URL, Invoker<T>> oldUrlInvokerMap, List<URL> urls) {
        Map<URL, Invoker<T>> newUrlInvokerMap = new ConcurrentHashMap<>(urls == null ? 1 : (int) (urls.size() / 0.75f + 1));
        if (urls == null || urls.isEmpty()) {
            return newUrlInvokerMap;
        }
		// 获取指定的协议
        String queryProtocols = this.queryMap.get(PROTOCOL_KEY);
		// 遍历所有提供者列表进行转换
        for (URL providerUrl : urls) {
            // If protocol is configured at the reference side, only the matching protocol is selected
            if (queryProtocols != null && queryProtocols.length() > 0) {
                boolean accept = false;
                String[] acceptProtocols = queryProtocols.split(",");
                for (String acceptProtocol : acceptProtocols) {
                    if (providerUrl.getProtocol().equals(acceptProtocol)) {
                        accept = true;
                        break;
                    }
                }
                if (!accept) {
                    continue;
                }
            }

            // 空协议 直接跳过
			if (EMPTY_PROTOCOL.equals(providerUrl.getProtocol())) {
                continue;
            }
			//  查询扩展是否存在，这个要注意如果是自定义扩展或者像webservice这样的扩展一般需要额外引入扩展包才可以
            if (!getUrl().getOrDefaultFrameworkModel().getExtensionLoader(Protocol.class).hasExtension(providerUrl.getProtocol())) {

                // 4-1 - Unsupported protocol

                logger.error("4-1", "typo in URL", "", "Unsupported protocol.",
                    new IllegalStateException("Unsupported protocol " + providerUrl.getProtocol() +
                    " in notified url: " + providerUrl + " from registry " + getUrl().getAddress() +
                    " to consumer " + NetUtils.getLocalHost() + ", supported protocol: " +
                    getUrl().getOrDefaultFrameworkModel().getExtensionLoader(Protocol.class).getSupportedExtensions()));

                continue;
            }
			// 合并url参数。顺序是：覆盖协议配置（比如禁用服务有时候就这样做）> -D(JVM参数) > 消费者 > 提供者
            URL url = mergeUrl(providerUrl);

            // Cache key is url that does not merge with consumer side parameters,
            // regardless of how the consumer combines parameters,
            // if the server url changes, then refer again
            Invoker<T> invoker = oldUrlInvokerMap == null ? null : oldUrlInvokerMap.remove(url);
            if (invoker == null) { // Not in the cache, refer again
                try {
                    boolean enabled = true;
                    if (url.hasParameter(DISABLED_KEY)) {
                        enabled = !url.getParameter(DISABLED_KEY, false);
                    } else {
                        enabled = url.getParameter(ENABLED_KEY, true);
                    }
                    if (enabled) {
						// 消费者引用服务提供者
                        invoker = protocol.refer(serviceType, url);
                    }
                } catch (Throwable t) {

                    // Thrown by AbstractProtocol.optimizeSerialization()
                    if (t instanceof RpcException && t.getMessage().contains("serialization optimizer")) {
                        // 4-2 - serialization optimizer class initialization failed.
                        logger.error("4-2", "typo in optimizer class", "",
                            "Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);

                    } else {
                        // 4-3 - Failed to refer invoker by other reason.
                        logger.error("4-3", "", "",
                            "Failed to refer invoker for interface:" + serviceType + ",url:(" + url + ")" + t.getMessage(), t);
                    }
                }
                if (invoker != null) { // Put new invoker in cache
                    newUrlInvokerMap.put(url, invoker);
                }
            } else {
                newUrlInvokerMap.put(url, invoker);
            }
        }
        return newUrlInvokerMap;
    }

到这里我们可以看到核心功能已经展示出来了

• 指定注册中心的三个路径的子节点变更事件
• 触发时间变更之后，把变更的节点信息转化为Invoker
  这时候得到的Invoker是最新的服务提供者地址，远程访问得到我们实际的结果

二、Protocol

根据协议创建通信连接
调用链
QosProtocolWrapper -> ProtocolFilterWrapper -> ProtocolListenerWrapper -> DubboProtocol

1、DubboProtocol.refer

    @Override
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        checkDestroyed();
        return protocolBindingRefer(type, url);
    }

    @Override
    public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
        checkDestroyed();
        optimizeSerialization(url);

        // create rpc invoker.
		// 创建通信client,封装成DubboInvoker
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        invokers.add(invoker);

        return invoker;
    }

2、DubboProtocol.getClients

创建通信客户端

• 判断是否为共享连接，默认是共享同一个连接进行通信
• 是否配置了多个连接通道 connections，默认只有一个共享连接

	private ExchangeClient[] getClients(URL url) {
        int connections = url.getParameter(CONNECTIONS_KEY, 0);
        // whether to share connection
        // if not configured, connection is shared, otherwise, one connection for one service
        // 如果没有配置的情况下，默认是采用共享连接，否则，就是针对一个服务提供一个连接。
		// 共享长连接，多路复用
		if (connections == 0) {
            /*
             * The xml configuration should have a higher priority than properties.
             */
            String shareConnectionsStr = StringUtils.isBlank(url.getParameter(SHARE_CONNECTIONS_KEY, (String) null))
                ? ConfigurationUtils.getProperty(url.getOrDefaultApplicationModel(), SHARE_CONNECTIONS_KEY, DEFAULT_SHARE_CONNECTIONS)
                : url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
            connections = Integer.parseInt(shareConnectionsStr);

            // 返回共享连接
			List<ReferenceCountExchangeClient> shareClients = getSharedClient(url, connections);
            ExchangeClient[] clients = new ExchangeClient[connections];
            Arrays.setAll(clients, shareClients::get);
            return clients;
        }
		// 如果不是使用共享连接，则初始化一个新的客户端连接进行返回
        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) {
            clients[i] = initClient(url);
        }
        return clients;
    }

最后创建的共享客户端是netty客户端
org.apache.dubbo.remoting.transport.netty4.NettyTransporter，外层由HeaderExchanger包装

3、NettyTransporter.connect

    @Override
    public Client connect(URL url, ChannelHandler handler) throws RemotingException {
        return new NettyClient(url, handler);
    }

4、NettyClient

    public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException {
        // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
        // the handler will be wrapped: MultiMessageHandler->HeartbeatHandler->handler
        super(url, wrapChannelHandler(url, handler));
    }
	public AbstractClient(URL url, ChannelHandler handler) throws RemotingException {
        // 初始化父类构造器，指定编码、超时时间等参数
		super(url, handler);
        // set default needReconnect true when channel is not connected
        needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, true);
		// 初始化线程池 默认为FixedThreadPool
        initExecutor(url);

        try {
			// 启动netty的核心代码初始化Bootstrap
            // 默认走的是NioSocketChannel
            // 初始化默认连接超时时间为3秒
            doOpen();
        } catch (Throwable t) {
            close();
            throw new RemotingException(url.toInetSocketAddress(), null,
                "Failed to start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress()
                    + " connect to the server " + getRemoteAddress() + ", cause: " + t.getMessage(), t);
        }

        try {
            // connect.
            // 前面是初始化netty的客户端启动类Bootstrap 这里是执行连接的代码：bootstrap.connect(getConnectAddress());
            // 等待3秒连接失败则抛出异常
            connect();
            if (logger.isInfoEnabled()) {
                logger.info("Start " + getClass().getSimpleName() + " " + NetUtils.getLocalAddress() + " connect to the server " + getRemoteAddress());
            }
        } catch (RemotingException t) {
            // ** 异常处理
        } catch (Throwable t) {
            // ** 异常处理
        }
    }

RegistryProtocol.refer过程中有一个关键步骤，即在监听到服务提供者url时触发RegistryDirectory.notify()方法。
RegistryDirectory.notify()方法调用 refreshInvoker()方法将服务提供者urls转换为对应的远程invoker，最终调用到 DubboProtocol.refer()方法生成对应的 DubboInvoker。
DubboInvoker的构造方法中有一项入参 ExchangeClient[] clients，对应网络客户端 Client，通过调用 client.request()方法完成网络通信的请求发送和响应接收功能。