深度解析dubbo服务远程暴露(一)

骑猪看日落 2023-02-18 14:08 46阅读 0赞

我们知道dubbo服务暴露分为本地（injvm）与远程（remote）两种方式，上篇[《深度解析dubbo服务本地暴露(injvm)》][dubbo_injvm]讲了dubbo服务本地暴露，接下来本篇将解析dubbo的远程暴露。  
我们接着看doExportUrlsFor1Protocol这个方法的后半部分

// don't export when none is configured
            if (!Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { 
    
                // export to local if the config is not remote (export to remote only when config is remote)
                if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {   // 本地服务暴露 不是remote就本地暴露，如果不配置scope也进行本地暴露
                    exportLocal(url);
                }
                // export to remote if the config is not local (export to local only when config is local)
                if (!Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope)) {   // 远程服务暴露 不是local
                    if (logger.isInfoEnabled()) { 
                        logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                    }
                    if (registryURLs != null && !registryURLs.isEmpty()) { 
                        //遍历注册中心
                        for (URL registryURL : registryURLs) { 
                            url = url.addParameterIfAbsent(Constants.DYNAMIC_KEY, registryURL.getParameter(Constants.DYNAMIC_KEY));
                            URL monitorUrl = loadMonitor(registryURL);//获取监控中心
                            if (monitorUrl != null) {   // 将监控中心添加到 url中
                                url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());
                            }
                            if (logger.isInfoEnabled()) { 
                                logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);
                            }
    
                            // For providers, this is used to enable custom proxy to generate invoker
                            String proxy = url.getParameter(Constants.PROXY_KEY);  // 配置中有proxy_key 的话就使用配置的
                            if (StringUtils.isNotEmpty(proxy)) {   // 设置配置的 proxy_key
                                registryURL = registryURL.addParameter(Constants.PROXY_KEY, proxy);
                            }
    
    
                            // invoker 使用ProxyFactory 生成 invoker对象，这里这个invoker其实是一个代理对象
                            Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
                            // 创建 DelegateProvoderMetaInvoker 对象
                            DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
                            // registryURL.getProtocol= registry
                            // filter ---->listener --->registryProtocol ( 这里使用了wapper 包装机制)
                            // filter ----> listener ----> dubboProtocol 服务暴露
                            Exporter<?> exporter = protocol.export(wrapperInvoker);
                            // 添加exporter
                            exporters.add(exporter);
                        }
                    } else {  // 没有注册中心
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);
                        DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
    
                        Exporter<?> exporter = protocol.export(wrapperInvoker);
                        exporters.add(exporter);
                    }
                }
            }

如果scope属性没有明确指出是local，这时候就进行远程暴露，如果有注册中心，然后遍历注册中心，首先是获取监控中心，将监控中心添加到url中，然后就是将proxy\_key属性放到registryUrl中，其实这个proxy\_key 的值是可以设置的，就是告诉dubbo我用什么来进行生成代理，这个对应的就是dubbo spi 自适应特性。接下来这几行就是非常重要的了，

// invoker 使用ProxyFactory 生成 invoker对象，这里这个invoker其实是一个代理对象
    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
    // 创建 DelegateProvoderMetaInvoker 对象
    DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);
    // registryURL.getProtocol= registry
    // filter ---->listener --->registryProtocol ( 这里使用了wapper 包装机制)
    // filter ----> listener ----> dubboProtocol 服务暴露
    Exporter<?> exporter = protocol.export(wrapperInvoker);
    // 添加exporter
    exporters.add(exporter);

首先`Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));`这行，与本地暴露一样，生成一个invoker对象，不熟悉的同学可以去看下本地暴露这篇文章。  
接着`DelegateProviderMetaDataInvoker wrapperInvoker = new DelegateProviderMetaDataInvoker(invoker, this);`，这行就是包装了一下这个invoker，主要是把原始的配置信息跟invoker绑在一块了。可以看下DelegateProviderMetaDataInvoker 这个类。

public class DelegateProviderMetaDataInvoker<T> implements Invoker { 
        protected final Invoker<T> invoker;
        private ServiceConfig metadata;
        public DelegateProviderMetaDataInvoker(Invoker<T> invoker,ServiceConfig metadata) { 
            this.invoker = invoker;
            this.metadata = metadata;
        }
       	...
        public ServiceConfig getMetadata() { 
            return metadata;
        }
    }

在接下来就是`Exporter<?> exporter = protocol.export(wrapperInvoker);`这行了。  
我们可以看下这个protocol：

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

获得的是一个自适应的扩展。

@SPI("dubbo")
    public interface Protocol { 
        int getDefaultPort();
        @Adaptive
        <T> Exporter<T> export(Invoker<T> invoker) throws RpcException;
        @Adaptive
        <T> Invoker<T> refer(Class<T> type, URL url) throws RpcException;
        void destroy();
    }

我们可以看到Protocol这个扩展点export方法上是有@Adaptive注解的，然后没有value，根据自适应的规则，没有value则value=类名小写。  
这样的话就是从url中获取protocol的值，咱们invoker里面包的url是registryURL，然后对应的protocol也就是注册中心的protocol，RegistryProtocol这个类。但是在创建扩展实现类的时候dubbo会给我们setter注入与wrapper包装，所以我们拿到的RegistryProtocol最终样子是这样的  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]  
Qos —> Filter —> Listener ----> RegistryProtocol  
接下来我们先看下QosProtocolWrapper，这里我们只挑与本篇有关的内容

@Override
        public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
            //判断是registry
            if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { 
                // 启动Qos服务器？
                startQosServer(invoker.getUrl());
                return protocol.export(invoker);
            }
            return protocol.export(invoker);
        }

判断是registry协议的话，就启动Qos服务器，然后调用下一个export 方法，也就是ProtocolFilterWrapper的export方法。

@Override
        public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
            if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { /// registry protocol 就到下一个wapper 包装对象中就可以了
                return protocol.export(invoker);
            }
            return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
        }

如果是registry协议的话就可以进入下一个了，然后不是registry协议就需要绑定一堆filter了，这个我们在说dubbo协议的时候会讲到，这里我们直接进入下一个类ProtocolListenerWrapper的export方法。

@Override
        public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
            if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { // registry 是否是注册中心
                return protocol.export(invoker);
            }
            return new ListenerExporterWrapper<T>(protocol.export(invoker),// dubbo export dubbo --->暴露服务 生成exporter
                    Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
                            .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));// exporter listener
        }

这里也是判断了一下是否是registry协议，如果是的话就直接进入下一个RegistryProtocol。

@Override
        public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { 
            //export invoker 暴露服务 doLocalExport表示本地启动服务不包括去注册中心注册
            final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
            // 获得注册中心URL
            URL registryUrl = getRegistryUrl(originInvoker);
    
            //registry provider 获得注册中心对象
            final Registry registry = getRegistry(originInvoker);
    
            // 获得服务提供者URL
            final URL registeredProviderUrl = getRegisteredProviderUrl(originInvoker);
    
            //to judge to delay publish whether or not
            boolean register = registeredProviderUrl.getParameter("register", true);
            // 向本地服务注册表注册服务
            ProviderConsumerRegTable.registerProvider(originInvoker, registryUrl, registeredProviderUrl);
    
            if (register) {   // 向注册中心注册自己
                register(registryUrl, registeredProviderUrl);
                ProviderConsumerRegTable.getProviderWrapper(originInvoker).setReg(true);  // 设置注册标志
            }
    
            // Subscribe the override data
            // FIXME When the provider subscribes, it will affect the scene : a certain JVM exposes the service and call the same service. Because the subscribed is cached key with the name of the service, it causes the subscription information to cover.
            final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registeredProviderUrl);
            final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl, originInvoker);
            overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
            registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
            //Ensure that a new exporter instance is returned every time export
            return new DestroyableExporter<T>(exporter, originInvoker, overrideSubscribeUrl, registeredProviderUrl);
        }

我们可以看到第一行`final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);`  
我们来看下doLocalExport方法，

// 暴露服务
        @SuppressWarnings("unchecked")
        private <T> ExporterChangeableWrapper<T> doLocalExport(final Invoker<T> originInvoker) { 
            String key = getCacheKey(originInvoker);
            ExporterChangeableWrapper<T> exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
            if (exporter == null) { //之前没有暴露过
                synchronized (bounds) { 
                    exporter = (ExporterChangeableWrapper<T>) bounds.get(key);
                    if (exporter == null) { 
                        // 封装 InvokerDelegete 将url封装起来了
                        final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));
                        exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);/// dubbo protocol
                        // 缓存起来
                        bounds.put(key, exporter);
                    }
                }
            }
            return exporter;
        }

首先获取一个cacheKey，这个cacheKey其实就是我们在服务暴露之前塞进去的一个export属性值（这里是把dynamic，enabled这两个属性移除掉了）：

Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));

接下来就是根据cacheKey去bounds里面找，如果没有找到，就说明之前没有暴露过，我们可以看下bounds这个成员：

//providerurl <--> exporter
        private final Map<String, ExporterChangeableWrapper<?>> bounds = new ConcurrentHashMap<String, ExporterChangeableWrapper<?>>();

如果没有暴露过的话，就将invoker封装起来，我们可以看下InvokerDelegete这个静态类与它的父类InvokerWrapper类

// InvokerDelegete 委托类 将provider 的url 进行缓存
        public static class InvokerDelegete<T> extends InvokerWrapper<T> { 
            private final Invoker<T> invoker;
            /** * @param invoker * @param url invoker.getUrl return this value */
            public InvokerDelegete(Invoker<T> invoker, URL url) { 
                super(invoker, url);
                this.invoker = invoker;
            }
            // 获取invoker
            public Invoker<T> getInvoker() { 
                if (invoker instanceof InvokerDelegete) {   // 如果是 委托类的话，就获取委托类里面的那个
                    return ((InvokerDelegete<T>) invoker).getInvoker();
                } else { 
                    return invoker;
                }
            }
        }

public class InvokerWrapper<T> implements Invoker<T> { 
        private final Invoker<T> invoker;
        private final URL url;
    	public InvokerWrapper(Invoker<T> invoker, URL url) { this.invoker = invoker;this.url = url;}
        @Override
        public Class<T> getInterface() { return invoker.getInterface();}
        @Override
        public URL getUrl() { return url;}
        @Override
        public boolean isAvailable() { return invoker.isAvailable();}
        @Override
        public Result invoke(Invocation invocation) throws RpcException { return invoker.invoke(invocation);}
        @Override
        public void destroy() { 
            invoker.destroy();
        }
    }

可以看出来这个委托类的作用就是把替换了url。  
接下来就是将protocol.export返回的exporter包装起来缓存到bounds中。  
我们看下ExporterChangeableWrapper这个成员内部包装类

private class ExporterChangeableWrapper<T> implements Exporter<T> { 
            private final Invoker<T> originInvoker;
            private Exporter<T> exporter;
            public ExporterChangeableWrapper(Exporter<T> exporter, Invoker<T> originInvoker) { 
                this.exporter = exporter;
                this.originInvoker = originInvoker;
            }
            public Invoker<T> getOriginInvoker() { 
                return originInvoker;
            }
            @Override
            public Invoker<T> getInvoker() { 
                return exporter.getInvoker();
            }
            public void setExporter(Exporter<T> exporter) { 
                this.exporter = exporter;
            }
            @Override
            public void unexport() { 
                String key = getCacheKey(this.originInvoker);
                bounds.remove(key);
                exporter.unexport();
            }
        }

这个作用就是将原始的invoker与exporter绑在了一起。  
下篇我们就看看这个exporter是怎么得到的。

[dubbo_injvm]: https://blog.csdn.net/yuanshangshenghuo/article/details/106462009
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200618160742121.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3l1YW5zaGFuZ3NoZW5naHVv,size_16,color_FFFFFF,t_70