dubbo框架揭秘之服务发布

通常情况下是通过Spring配置的方式去实现服务的发布，为了方便调试，我就不采用Spring配置的方式。

        DemoService demo = new DemoServiceImpl();

        ApplicationConfig config=new ApplicationConfig("hello-world-app");

        RegistryConfig reg=new RegistryConfig("127.0.0.1:2181");

        reg.setProtocol("zookeeper");

        ProtocolConfig protocol = new ProtocolConfig();

        protocol.setName("dubbo");

        protocol.setPort(20880);

        ServiceConfig<DemoService> service=new ServiceConfig<DemoService>();

        service.setApplication(config);

        service.setRegistry(reg);

        service.setProtocol(protocol);

        service.setInterface(DemoService.class);

        service.setRef(demo);

        service.setVersion("1.0");

        service.export();

        Thread.sleep(Integer.MAX_VALUE);

上面是服务发布的简短代码，其中最重要的就是export()方法。

public synchronized void export() {

           ...

            doExport()

    }

protected synchronized void doExport() {

       ...

        doExportUrls();

    }

private void doExportUrls() {
　　    //装配地址 registry://127.0.0.1:2181/com.alibaba.dubbo.registry.RegistryService?application=hello-world-app&dubbo=2.5.3&pid=5556&registry=zookeeper&timestamp=1484484518418　　

        List<URL> registryURLs = loadRegistries(true);

        for (ProtocolConfig protocolConfig : protocols) {

            doExportUrlsFor1Protocol(protocolConfig, registryURLs);

        }

    }

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) {

        //配置为none不暴露

        if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) {

            //配置不是remote的情况下做本地暴露 (配置为remote，则表示只暴露远程服务)

            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) {

                exportLocal(url);

            }

            //如果配置不是local则暴露为远程服务.(配置为local，则表示只暴露远程服务)

            if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){

                if (logger.isInfoEnabled()) {

                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);

                }

                if (registryURLs != null && registryURLs.size() > 0

                        && url.getParameter("register", true)) {

                    for (URL registryURL : registryURLs) {

                        url = url.addParameterIfAbsent("dynamic", registryURL.getParameter("dynamic"));

                        URL monitorUrl = loadMonitor(registryURL);

                        if (monitorUrl != null) {

                            url = url.addParameterAndEncoded(Constants.MONITOR_KEY, monitorUrl.toFullString());

                        }

                        if (logger.isInfoEnabled()) {

                            logger.info("Register dubbo service " + interfaceClass.getName() + " url " + url + " to registry " + registryURL);

                        }

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

                        Exporter<?> exporter = protocol.export(invoker);

                        exporters.add(exporter);

                    }

                } else {

                    Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, url);

                    Exporter<?> exporter = protocol.export(invoker);

                    exporters.add(exporter);

                }

            }

        }

        this.urls.add(url);

    }

最终调用的是protocol.export(invoker)方法，protocol是通过扩展点机制动态生成的类：

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

public class Protocol$Adpative implements com.alibaba.dubbo.rpc.Protocol {

        public void destroy() {

            throw new UnsupportedOperationException(

                    "method public abstract void com.alibaba.dubbo.rpc.Protocol.destroy() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");

        }

        public int getDefaultPort() {

            throw new UnsupportedOperationException(

                    "method public abstract int com.alibaba.dubbo.rpc.Protocol.getDefaultPort() of interface com.alibaba.dubbo.rpc.Protocol is not adaptive method!");

        }

　　　　 //调用该方法

        public com.alibaba.dubbo.rpc.Exporter export(com.alibaba.dubbo.rpc.Invoker arg0)

                throws com.alibaba.dubbo.rpc.Invoker {

            if (arg0 == null)

                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument == null");

            if (arg0.getUrl() == null)

                throw new IllegalArgumentException("com.alibaba.dubbo.rpc.Invoker argument getUrl() == null");

            com.alibaba.dubbo.common.URL url = arg0.getUrl();

            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());

            if (extName == null)

                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url("

                        + url.toString() + ") use keys([protocol])");
           //扩展点机制加载扩展类：

            com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader

                    .getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);

            return extension.export(arg0);

        }

        public com.alibaba.dubbo.rpc.Invoker refer(java.lang.Class arg0, com.alibaba.dubbo.common.URL arg1)

                throws java.lang.Class {

            if (arg1 == null)

                throw new IllegalArgumentException("url == null");

            com.alibaba.dubbo.common.URL url = arg1;

            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());

            if (extName == null)

                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.rpc.Protocol) name from url("

                        + url.toString() + ") use keys([protocol])");

            com.alibaba.dubbo.rpc.Protocol extension = (com.alibaba.dubbo.rpc.Protocol) ExtensionLoader

                    .getExtensionLoader(com.alibaba.dubbo.rpc.Protocol.class).getExtension(extName);

            return extension.refer(arg0, arg1);

        }

    }

加载扩展类的核心方法：

public T getExtension(String name) {

        ...

        Object instance = holder.get();

        if (instance == null) {

            synchronized (holder) {

                instance = holder.get();

                if (instance == null) {

                    instance = createExtension(name);

                    holder.set(instance);

                }

            }

        }

        return (T) instance;

    }

private T createExtension(String name) {

        try {

            T instance = (T) EXTENSION_INSTANCES.get(clazz);

            if (instance == null) {

                EXTENSION_INSTANCES.putIfAbsent(clazz, (T) clazz.newInstance());

                instance = (T) EXTENSION_INSTANCES.get(clazz);

            }

            injectExtension(instance);
            //cachedWrapperClasses在getAdaptiveExtension方法中被初始化，值为：ProtocolListenerWrapper、ProtocolFilterWrapper
            //这2个包装类通过装饰器模式，构造器实例化，最终返回ProtocolFilterWrapper(构造函数参数：ProtocolListenerWrapper(构造函数参数：RegistryProtocol))

            Set<Class<?>> wrapperClasses = cachedWrapperClasses;

            if (wrapperClasses != null && wrapperClasses.size() > 0) {

                for (Class<?> wrapperClass : wrapperClasses) {

                    instance = injectExtension((T) wrapperClass.getConstructor(type).newInstance(instance));

                }

            }

            return instance;

        } catch (Throwable t) {

            throw new IllegalStateException("Extension instance(name: " + name + ", class: " +

                    type + ")  could not be instantiated: " + t.getMessage(), t);

        }

    }

所以，最终层层包装，调用RegistryProtocol的export(Invoker<T>)方法：

 public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException {

        //export invoker

        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);

        //registry provider

        final Registry registry = getRegistry(originInvoker);

        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
　　　　 //向注册中心注册地址：/dubbo/com.selrain.dubbodemo.DemoService/providers

        registry.register(registedProviderUrl);

        // 订阅override数据

        // FIXME 提供者订阅时，会影响同一JVM即暴露服务，又引用同一服务的的场景，因为subscribed以服务名为缓存的key，导致订阅信息覆盖。

        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);

        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);

        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);

        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);

        //保证每次export都返回一个新的exporter实例

        return new Exporter<T>() {

            public Invoker<T> getInvoker() {

                return exporter.getInvoker();

            }

            public void unexport() {

                try {

                    exporter.unexport();

                } catch (Throwable t) {

                    logger.warn(t.getMessage(), t);

                }

                try {

                    registry.unregister(registedProviderUrl);

                } catch (Throwable t) {

                    logger.warn(t.getMessage(), t);

                }

                try {

                    overrideListeners.remove(overrideSubscribeUrl);

                    registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);

                } catch (Throwable t) {

                    logger.warn(t.getMessage(), t);

                }

            }

        };

    }

参考：

Dubbo原理解析-服务发布

巴特西

dubbo框架揭秘之服务发布

最新文章

热门文章