Dubbo Consumer參照服務範例程式碼詳解

2023-03-02 18:01:22

Consumer消費者Demo範例

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:dubbo="http://dubbo.apache.org/schema/dubbo"
       xmlns="http://www.springframework.org/schema/beans" xmlns:context="http://www.springframework.org/schema/context"
       xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
       http://dubbo.apache.org/schema/dubbo http://dubbo.apache.org/schema/dubbo/dubbo.xsd http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd">
    <context:property-placeholder/>
    <dubbo:application name="serialization-java-consumer">
        <dubbo:parameter key="qos.enable" value="true" />
        <dubbo:parameter key="qos.accept.foreign.ip" value="false" />
        <dubbo:parameter key="qos.port" value="33333" />
    </dubbo:application>
    <dubbo:registry address="zookeeper://${zookeeper.address:127.0.0.1}:2181"/>
    <dubbo:reference id="demoService" check="true" interface="org.apache.dubbo.samples.serialization.api.DemoService"/>
</beans>

在之前的章節中已經知道，Dubbo基於Spring自定義標籤規範實現了自定義標籤，通過自定義標籤完成了bean的載入，並且通過實現監聽Spring容器重新整理完畢事件啟動dubbo使用者端。啟動使用者端伴隨著服務釋出和服務的訂閱。

public class DubboConsumer {
    public static void main(String[] args) {
        ClassPathXmlApplicationContext context = new ClassPathXmlApplicationContext("spring/dubbo-demo-consumer.xml");
        context.start();
        DemoService demoService = context.getBean("demoService", DemoService.class);
        String hello = demoService.sayHello("world");
        System.out.println(hello);
    }
}

dubbo通過<dubbo:reference標籤參照服務，之後在程式中通過Spring的Context依賴查詢(getBean)的方式獲取參照的服務的代理範例。<dubbo:reference載入的Bean是ReferenceBean，它實現了FactoryBean介面，getBean時會呼叫ReferenceBean的getObject()方法，這是獲取參照的入口。getBean方法會判斷Reference物件是否是空的，如果是空的，呼叫init方法。程式碼如下：

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

ReferenceConfig#getObject()獲取應用Bean

ReferenceBean繼承了ReferenceConfig，當呼叫ReferenceBean的getObject()方法會呼叫ReferenceBean的get()方法。

    public synchronized T get() {
        if (destroyed) {
            throw new IllegalStateException("The invoker of ReferenceConfig(" + url + ") has already destroyed!");
        }
        // 代理參照如果是空的，呼叫init
        if (ref == null) {
            init();
        }
        return ref;
    }
    public synchronized void init() {
        if (initialized) {
            return;
        }
        if (bootstrap == null) {
            bootstrap = DubboBootstrap.getInstance();
            bootstrap.init();
        }
        // 1. 檢查設定ConsumerConfig，有的話檢查設定，沒有就新建一個ConsumerConfig
        // 2. 反射建立呼叫的API
        // 3. 初始化ServiceMetadata
        // 4. 註冊Consumer
        // 5. 檢查ReferenceConfig，RegistryConfig，ConsumerConfig
        checkAndUpdateSubConfigs();
        checkStubAndLocal(interfaceClass);
        // 檢查參照的介面是否mock
        ConfigValidationUtils.checkMock(interfaceClass, this);
        // consumer的資訊
        Map<String, String> map = new HashMap<String, String>();
        map.put(SIDE_KEY, CONSUMER_SIDE);
        // 新增執行時引數到map，包括：dubbo，release，timestamp，pid
        ReferenceConfigBase.appendRuntimeParameters(map);
        // 是不是泛化，不是的話進入條件
        if (!ProtocolUtils.isGeneric(generic)) {
            String revision = Version.getVersion(interfaceClass, version);
            if (revision != null && revision.length() > 0) {
                map.put(REVISION_KEY, revision);
            }
            // 獲取方法，生成包裝類，使用javassist，將生成的類放到WRAPPER_MAP中，key是org.apache.dubbo.samples.serialization.api.DemoService類物件，value是包裝類的範例
            String[] methods = Wrapper.getWrapper(interfaceClass).getMethodNames();
            if (methods.length == 0) {
                logger.warn("No method found in service interface " + interfaceClass.getName());
                map.put(METHODS_KEY, ANY_VALUE);
            } else {
                // method放到map，這裡method是sayHello
                map.put(METHODS_KEY, StringUtils.join(new HashSet<String>(Arrays.asList(methods)), COMMA_SEPARATOR));
            }
        }
        // interface org.apache.dubbo.samples.serialization.api.DemoService
        map.put(INTERFACE_KEY, interfaceName);
        // 追加其他引數到map中
        AbstractConfig.appendParameters(map, getMetrics());
        AbstractConfig.appendParameters(map, getApplication());
        AbstractConfig.appendParameters(map, getModule());
        // remove 'default.' prefix for configs from ConsumerConfig
        // appendParameters(map, consumer, Constants.DEFAULT_KEY);
        AbstractConfig.appendParameters(map, consumer);
        AbstractConfig.appendParameters(map, this);
        MetadataReportConfig metadataReportConfig = getMetadataReportConfig();
        if (metadataReportConfig != null && metadataReportConfig.isValid()) {
            map.putIfAbsent(METADATA_KEY, REMOTE_METADATA_STORAGE_TYPE);
        }
        Map<String, AsyncMethodInfo> attributes = null;
        if (CollectionUtils.isNotEmpty(getMethods())) {
            attributes = new HashMap<>();
            for (MethodConfig methodConfig : getMethods()) {
                AbstractConfig.appendParameters(map, methodConfig, methodConfig.getName());
                String retryKey = methodConfig.getName() + ".retry";
                if (map.containsKey(retryKey)) {
                    String retryValue = map.remove(retryKey);
                    if ("false".equals(retryValue)) {
                        map.put(methodConfig.getName() + ".retries", "0");
                    }
                }
                AsyncMethodInfo asyncMethodInfo = AbstractConfig.convertMethodConfig2AsyncInfo(methodConfig);
                if (asyncMethodInfo != null) {
//                    consumerModel.getMethodModel(methodConfig.getName()).addAttribute(ASYNC_KEY, asyncMethodInfo);
                    attributes.put(methodConfig.getName(), asyncMethodInfo);
                }
            }
        }
        String hostToRegistry = ConfigUtils.getSystemProperty(DUBBO_IP_TO_REGISTRY);
        if (StringUtils.isEmpty(hostToRegistry)) {
            hostToRegistry = NetUtils.getLocalHost();
        } else if (isInvalidLocalHost(hostToRegistry)) {
            throw new IllegalArgumentException("Specified invalid registry ip from property:" + DUBBO_IP_TO_REGISTRY + ", value:" + hostToRegistry);
        }
        map.put(REGISTER_IP_KEY, hostToRegistry);
        // 所有資料在存到serviceMetadata的attachments中
        serviceMetadata.getAttachments().putAll(map);
        // 建立Service代理
        ref = createProxy(map);
        // 設定ServiceMetadata的Service代理參照
        serviceMetadata.setTarget(ref);
        serviceMetadata.addAttribute(PROXY_CLASS_REF, ref);
        ConsumerModel consumerModel = repository.lookupReferredService(serviceMetadata.getServiceKey());
        consumerModel.setProxyObject(ref);
        consumerModel.init(attributes);
        // 標記初始化完畢
        initialized = true;
        // dispatch a ReferenceConfigInitializedEvent since 2.7.4
        dispatch(new ReferenceConfigInitializedEvent(this, invoker));
    }

ReferenceConfig#createProxy()建立服務代理

    @SuppressWarnings({"unchecked", "rawtypes", "deprecation"})
    private T createProxy(Map<String, String> map) {
        // 是否是InJvm，協定是InJvm
        if (shouldJvmRefer(map)) {
            URL url = new URL(LOCAL_PROTOCOL, LOCALHOST_VALUE, 0, interfaceClass.getName()).addParameters(map);
            invoker = REF_PROTOCOL.refer(interfaceClass, url);
            if (logger.isInfoEnabled()) {
                logger.info("Using injvm service " + interfaceClass.getName());
            }
        } else {
            urls.clear();
            if (url != null && url.length() > 0) { // user specified URL, could be peer-to-peer address, or register center's address.
                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 (UrlUtils.isRegistry(url)) {
                            urls.add(url.addParameterAndEncoded(REFER_KEY, StringUtils.toQueryString(map)));
                        } else {
                            urls.add(ClusterUtils.mergeUrl(url, map));
                        }
                    }
                }
            } else { // assemble URL from register center's configuration
                // 如果protocols不是injvm檢查註冊中心
                if (!LOCAL_PROTOCOL.equalsIgnoreCase(getProtocol())) {
                    checkRegistry();
                    // url列表，將zookeeper://改成registry://
                    List<URL> us = ConfigValidationUtils.loadRegistries(this, false);
                    if (CollectionUtils.isNotEmpty(us)) {
                        for (URL u : us) {
                            // 監控URL
                            URL monitorUrl = ConfigValidationUtils.loadMonitor(this, u);
                            // 如果有監控設定放到map中
                            if (monitorUrl != null) {
                                map.put(MONITOR_KEY, URL.encode(monitorUrl.toFullString()));
                            }
                            // 根據map中的資訊生成url，這裡生成的結果是
                            /*
                            registry://127.0.0.1:2181/org.apache.dubbo.registry.RegistryService?application=serialization-java-consumer&dubbo=2.0.2&pid=66793&qos.accept.foreign.ip=false&qos.enable=true&qos.port=33333&refer=application%3Dserialization-java-consumer%26check%3Dtrue%26dubbo%3D2.0.2%26init%3Dfalse%26interface%3Dorg.apache.dubbo.samples.serialization.api.DemoService%26methods%3DsayHello%26pid%3D66793%26qos.accept.foreign.ip%3Dfalse%26qos.enable%3Dtrue%26qos.port%3D33333%26register.ip%3D192.168.58.45%26release%3D2.7.7%26side%3Dconsumer%26sticky%3Dfalse%26timestamp%3D1636532992568&registry=zookeeper&release=2.7.7&timestamp=1636533091883
                            */
                            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.");
                    }
                }
            }
            // 如果參照的URL就1個直接通過refer參照服務，這裡和export相似，通過呼叫鏈實現的
            // - ProtocolListenerWrapper
            // - - ProtocolFilterWrapper
            // - - - RegistryProtocol
            if (urls.size() == 1) {
                // 通過RegistryProtocol#refer參照服務
                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 (UrlUtils.isRegistry(url)) {
                        registryURL = url; // use last registry url
                    }
                }
                if (registryURL != null) { // registry url is available
                    // for multi-subscription scenario, use 'zone-aware' policy by default
                    // 叢集處理
                    URL u = registryURL.addParameterIfAbsent(CLUSTER_KEY, ZoneAwareCluster.NAME);
                    // The invoker wrap relation would be like: ZoneAwareClusterInvoker(StaticDirectory) -> FailoverClusterInvoker(RegistryDirectory, routing happens here) -> Invoker
                    // 加入到叢集中，這裡包含集中叢集處理模式，分別是：
                    invoker = CLUSTER.join(new StaticDirectory(u, invokers));
                } else { // not a registry url, must be direct invoke.
                    invoker = CLUSTER.join(new StaticDirectory(invokers));
                }
            }
        }
        // invoer不可用處理
        if (shouldCheck() && !invoker.isAvailable()) {
            invoker.destroy();
            throw new IllegalStateException("Failed to check the status of the service "
                    + interfaceName
                    + ". No provider available for the service "
                    + (group == null ? "" : group + "/")
                    + interfaceName +
                    (version == null ? "" : ":" + version)
                    + " from the url "
                    + invoker.getUrl()
                    + " to the consumer "
                    + NetUtils.getLocalHost() + " use dubbo version " + Version.getVersion());
        }
        if (logger.isInfoEnabled()) {
            logger.info("Refer dubbo service " + interfaceClass.getName() + " from url " + invoker.getUrl());
        }
        /**
         * @since 2.7.0
         * ServiceData儲存，SPI機制，支援記憶體和遠端兩種方式
         */
        String metadata = map.get(METADATA_KEY);
        WritableMetadataService metadataService = WritableMetadataService.getExtension(metadata == null ? DEFAULT_METADATA_STORAGE_TYPE : metadata);
        if (metadataService != null) {
            URL consumerURL = new URL(CONSUMER_PROTOCOL, map.remove(REGISTER_IP_KEY), 0, map.get(INTERFACE_KEY), map);
            metadataService.publishServiceDefinition(consumerURL);
        }
        // create service proxy
        return (T) PROXY_FACTORY.getProxy(invoker, ProtocolUtils.isGeneric(generic));
    }

建立代理服務會建立Invoker，在參照服務過程中，會判斷協定是否為injvm，會根據協定做不同的處理，不是injvm協定會根據構造的設定資訊（map）生成url並將url協定有zookeeper://改成registry://，然後通過Protocol介面refer方法參照服務，與釋出服務相似，參照服務的過程也會包裝方法的呼叫鏈，如下：

- ProtocolListenerWrapper
- - ProtocolFilterWrapper
- - - RegistryProtocol

在refer的過程中會對一個伺服器端的參照和一個服務多個伺服器端的服務進行區分處理，對於有多個伺服器端的服務會進行叢集處理（cluster），會講invoker列表加入到叢集中，在呼叫過程中會根據叢集策略來選擇不同的策略進行呼叫，叢集策略實現也實現了SPI機制，

RegistryProtocol#refer參照服務

    @Override
    @SuppressWarnings("unchecked")
    public <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException {
        // 註冊中心url 將 registry:// 轉成 zookeeper://
        url = getRegistryUrl(url);
        // 獲取註冊中心
        Registry registry = registryFactory.getRegistry(url);
        if (RegistryService.class.equals(type)) {
            return proxyFactory.getInvoker((T) registry, type, url);
        }
        // 從url中解析出參照服務資訊
        Map<String, String> qs = StringUtils.parseQueryString(url.getParameterAndDecoded(REFER_KEY));
        // 獲取分組
        String group = qs.get(GROUP_KEY);
        // 如果設定分組了，那麼使用MergeableCluster策略
        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);
    }

RegistryProtocol#doRefer參照服務

    private <T> Invoker<T> doRefer(Cluster cluster, Registry registry, Class<T> type, URL url) {
        // 獲取叢集目錄Directory，代表Invoker的集合
        RegistryDirectory<T> directory = new RegistryDirectory<T>(type, url);
        directory.setRegistry(registry);
        directory.setProtocol(protocol);
        // all attributes of REFER_KEY
        Map<String, String> parameters = new HashMap<String, String>(directory.getConsumerUrl().getParameters());
        URL subscribeUrl = new URL(CONSUMER_PROTOCOL, parameters.remove(REGISTER_IP_KEY), 0, type.getName(), parameters);
        if (directory.isShouldRegister()) {
            directory.setRegisteredConsumerUrl(subscribeUrl);
            // 註冊消費者url
            registry.register(directory.getRegisteredConsumerUrl());
        }
        directory.buildRouterChain(subscribeUrl);
        // 向ZK發起訂閱服務，並設定監聽，這裡比較複雜，最終會在ZookeeperRegistry#doSubscribe做服務訂閱
        directory.subscribe(toSubscribeUrl(subscribeUrl));
        // 加入叢集
        Invoker<T> invoker = cluster.join(directory);
        List<RegistryProtocolListener> listeners = findRegistryProtocolListeners(url);
        if (CollectionUtils.isEmpty(listeners)) {
            return invoker;
        }
        RegistryInvokerWrapper<T> registryInvokerWrapper = new RegistryInvokerWrapper<>(directory, cluster, invoker, subscribeUrl);
        for (RegistryProtocolListener listener : listeners) {
            listener.onRefer(this, registryInvokerWrapper);
        }
        return registryInvokerWrapper;
    }

RegistryDirectory#subscribe訂閱服務

RegistryProtocol#doRefer方法，directory.subscribe會按照下面的呼叫鏈進行處理，最後呼叫ZookeeperRegistry#doSubscribe方法向zk註冊資料訂閱介面，並設定監聽。

- RegistryDirectory
- - ListenerRegistryWrapper
- - - FailbackRegistry
- - - - ZookeeperRegistry

    @Override
    public void doSubscribe(final URL url, final NotifyListener listener) {
        try {
            if (ANY_VALUE.equals(url.getServiceInterface())) {
                String root = toRootPath();
                // 初始化監聽
                ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
                ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (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)), k);
                        }
                    }
                });
                // 建立zk節點
                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 {
                List<URL> urls = new ArrayList<>();
                for (String path : toCategoriesPath(url)) {
                    ConcurrentMap<NotifyListener, ChildListener> listeners = zkListeners.computeIfAbsent(url, k -> new ConcurrentHashMap<>());
                    ChildListener zkListener = listeners.computeIfAbsent(listener, k -> (parentPath, currentChilds) -> ZookeeperRegistry.this.notify(url, k, toUrlsWithEmpty(url, parentPath, currentChilds)));
                    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);
        }
    }

DubboProtocol#protocolBindingRefer建立Invoker

refer服務建立Invoker時會呼叫該方法，該方法會通過getClients建立網路使用者端，建立使用者端是會判斷使用者端是否為共用連結，根據connections建立使用者端ExchangeClient，然後通過initClient初始化使用者端，初始化過程中會判斷是否為延遲的使用者端LazyConnectExchangeClient，不是延遲使用者端，就會通過connect連線服務提供者，與服務提供者連線，具體建立連線流程不在這裡說明，會在網路通訊介紹。

    @Override
    public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException {
        optimizeSerialization(url);
        // 建立RPC Invoker，通過getClients(url)建立網路使用者端
        DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers);
        // 將invoker新增到invokers中
        invokers.add(invoker);
        return invoker;
    }
    // 建立使用者端陣列
    private ExchangeClient[] getClients(URL url) {
        // whether to share connection
        // 是否共用連結
        boolean useShareConnect = false;
        int connections = url.getParameter(CONNECTIONS_KEY, 0);
        List<ReferenceCountExchangeClient> shareClients = null;
        // if not configured, connection is shared, otherwise, one connection for one service
        if (connections == 0) {
            useShareConnect = true;
            /*
             * The xml configuration should have a higher priority than properties.
             * xml設定優先順序高於properties設定
             */
            String shareConnectionsStr = url.getParameter(SHARE_CONNECTIONS_KEY, (String) null);
            connections = Integer.parseInt(StringUtils.isBlank(shareConnectionsStr) ? ConfigUtils.getProperty(SHARE_CONNECTIONS_KEY,
                    DEFAULT_SHARE_CONNECTIONS) : shareConnectionsStr);
            // 共用連結使用者端
            shareClients = getSharedClient(url, connections);
        }
        // 建立ExchangeClient
        ExchangeClient[] clients = new ExchangeClient[connections];
        for (int i = 0; i < clients.length; i++) {
            if (useShareConnect) {
                // 從共用使用者端獲取
                clients[i] = shareClients.get(i);
            } else {
                // 初始化使用者端
                clients[i] = initClient(url);
            }
        }
        return clients;
    }
    // 初始化使用者端
    private ExchangeClient initClient(URL url) {
        // client type setting.
        String str = url.getParameter(CLIENT_KEY, url.getParameter(SERVER_KEY, DEFAULT_REMOTING_CLIENT));
        url = url.addParameter(CODEC_KEY, DubboCodec.NAME);
        // enable heartbeat by default
        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 {
            // connection should be lazy
            if (url.getParameter(LAZY_CONNECT_KEY, false)) {
                // 延遲載入使用者端
                client = new LazyConnectExchangeClient(url, requestHandler);
            } else {
                // 通過NettyTransporter connect建立使用者端 與provider建立連線
                client = Exchangers.connect(url, requestHandler);
            }
        } catch (RemotingException e) {
            throw new RpcException("Fail to create remoting client for service(" + url + "): " + e.getMessage(), e);
        }
        return client;
    }

總結

Dubbo Consumer整個的啟動，參照服務的流程大致就看完了，ReferenceBean實現了Spring的FactoryBean介面，在使用Spring上下文getBean時就會呼叫到ReferenceBean的getObject方法，這時就會通過建立代理（createProxy），然後通過Protocol的refer方法參照服務；

參照服務的流程大致可以理解為，通過DubboProtocol的refer方法建立DubboInvoker呼叫getClients方法建立ExchangeClient，然後通過initClient方法初始化網路使用者端，初始化使用者端過程中會通過Exchangers的connect與服務提供者建立連線，後續就是網路使用者端與伺服器端建立連線這裡會和服務提供者相似，通過doOpen初始化網路使用者端，然後呼叫doConnect建立連線。

到此這篇關於Dubbo Consumer參照服務範例程式碼詳解的文章就介紹到這了,更多相關Dubbo Consumer參照服務內容請搜尋it145.com以前的文章或繼續瀏覽下面的相關文章希望大家以後多多支援it145.com！