Dubbo源码解析之服务发布与注册

客官°小女子只卖身不卖艺 2022-04-22 05:24 263阅读 0赞

# 准备 #

dubbo版本：2.5.4

# Spring自定义扩展 #

`dubbo` 是基于 `spring` 配置来实现服务发布，并基于 `spring` 的扩展机制定义了一套自定义标签，要实现自定义扩展， `spring` 中提供了 `NamespaceHandler` 、`BeanDefinitionParser` 两个类用于实现扩展

*  `NamespaceHandler` ：注册一系列 `BeanDefinitionParser` ，利用他们进行解析
 *  `BeanDefinitionParser` ：用于解析每个 `element` 内容
 *  `Spring` 默认会加载jar包下的 `META-INF/spring.handlers` 文件寻找对应的 `NamespaceHandler`（`Dubbo-config` 模块下 `dubbo-config-spring` ）。

![dubbo-publish-and-register-01.png][]

查看 `DubboNamespaceHandler` 类

public class DubboNamespaceHandler extends NamespaceHandlerSupport { 
    
    	static { 
    		Version.checkDuplicate(DubboNamespaceHandler.class);
    	}
    
    	public void init() { 
    	    registerBeanDefinitionParser("application", new DubboBeanDefinitionParser(ApplicationConfig.class, true));
            registerBeanDefinitionParser("module", new DubboBeanDefinitionParser(ModuleConfig.class, true));
            registerBeanDefinitionParser("registry", new DubboBeanDefinitionParser(RegistryConfig.class, true));
            registerBeanDefinitionParser("monitor", new DubboBeanDefinitionParser(MonitorConfig.class, true));
            registerBeanDefinitionParser("provider", new DubboBeanDefinitionParser(ProviderConfig.class, true));
            registerBeanDefinitionParser("consumer", new DubboBeanDefinitionParser(ConsumerConfig.class, true));
            registerBeanDefinitionParser("protocol", new DubboBeanDefinitionParser(ProtocolConfig.class, true));
            registerBeanDefinitionParser("service", new DubboBeanDefinitionParser(ServiceBean.class, true));
            registerBeanDefinitionParser("reference", new DubboBeanDefinitionParser(ReferenceBean.class, false));
            registerBeanDefinitionParser("annotation", new DubboBeanDefinitionParser(AnnotationBean.class, true));
        }
    
    }

通过 `DubboBeanDefinitionParser` 将 `dubbo` 自定义标签解析为相应 `Bean` 对象。同时为了 `spring` 启动时，相应地启动 `provider` 发布服务注册服务的过程，以及让客户端在启动时自动订阅发现服务，添加了两个 `bean` ，`ServiceBean` 及 `ReferenceBean` ，分别继承 `ServiceConfig` 和 `ReferenceConfig` 。

`ServiceBean` 实现了`InitializingBean` 、`DisposableBean` 、`ApplicationContextAware` 、`ApplicationListener` 、`BeanNameAware`

*  **InitializingBean**：`bean` 初始化过程中会调用 `afterPropertiesSet` 方法
 *  **DisposableBean**：`bean` 销毁时，`spring` 容器会自动执行 `destory` 方法，比如释放资源
 *  **ApplicationContextAware**：当 `spring` 容器初始化时会自动将 `ApplicationContext` 注入进来
 *  **ApplicationListener**：`ApplicationEvent` 事件监听，`spring` 容器启动后会发一个事件通知
 *  **BeanNameAware**：获得自身初始化时 `bean` 的 `id` 属性

# 服务发布过程 #

先上时序图，帮助跟踪过程走向。  
![Dubbo发布服务][Dubbo]

`serviceBean` 是服务发布入口，在 `bean` 初始化时会执行其 `afterPropertiesSet` 方法，在该方法最后会执行 `export` 方法进行发布

// ServiceConfig
    public synchronized void export() { 
        if (provider != null) { 
            if (export == null) { 
                export = provider.getExport();
            }
            if (delay == null) { 
                delay = provider.getDelay();
            }
        }
        if (export != null && ! export.booleanValue()) { 
            return;
        }
        if (delay != null && delay > 0) { 
            Thread thread = new Thread(new Runnable() { 
                public void run() { 
                    try { 
                        Thread.sleep(delay);
                    } catch (Throwable e) { 
                    }
                    doExport();
                }
            });
            thread.setDaemon(true);
            thread.setName("DelayExportServiceThread");
            thread.start();
        } else { 
            doExport();
        }
    }

接着调用 `doExportUrls` 方法

private void doExportUrls() { 
        List<URL> registryURLs = loadRegistries(true);// 是否获得注册中心配置
        for (ProtocolConfig protocolConfig : protocols) { 
            // 执行协议发布
            doExportUrlsFor1Protocol(protocolConfig, registryURLs);
        }
    }

doExportUrlsFor1Protocol

private void doExportUrlsFor1Protocol(ProtocolConfig protocolConfig, List<URL> registryURLs) { 
        /** * 只保留核心代码 */
        String scope = url.getParameter(Constants.SCOPE_KEY);
        if (! Constants.SCOPE_NONE.toString().equalsIgnoreCase(scope)) { 
            if (!Constants.SCOPE_REMOTE.toString().equalsIgnoreCase(scope)) { 
                exportLocal(url);
            }
            if (! Constants.SCOPE_LOCAL.toString().equalsIgnoreCase(scope) ){ 
                if (logger.isInfoEnabled()) { 
                    logger.info("Export dubbo service " + interfaceClass.getName() + " to url " + url);
                }
                // registryURLs
                // registrys://...
                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);
                        }
                        // 通过proxyFactory获取Invoker对象
                        Invoker<?> invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString()));
      					// 注册服务 
                        // protocol = ExtensionLoader.getExtensionLoader(Protocol.class).getAdaptiveExtension();
                        // 这里的protocol是Protocol$Adpative适配器对象
                        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$Adpative

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.RpcException { 
            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(Class arg0, com.alibaba.dubbo.common.URL arg1) throws com.alibaba.dubbo.rpc.RpcException { 
            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);
        }
    }

getExtension

public T getExtension(String name) { 
        if (name != null && name.length() != 0) { 
            if ("true".equals(name)) { 
                return this.getDefaultExtension();
            } else { 
                Holder<Object> holder = (Holder)this.cachedInstances.get(name);
                if (holder == null) { 
                    this.cachedInstances.putIfAbsent(name, new Holder());
                    holder = (Holder)this.cachedInstances.get(name);
                }
    
                Object instance = holder.get();
                if (instance == null) { 
                    synchronized(holder) { 
                        instance = holder.get();
                        if (instance == null) { 
                            // 创建指定名称的扩张点>>
                            instance = this.createExtension(name);
                            holder.set(instance);
                        }
                    }
                }
    
                return instance;
            }
        } else { 
            throw new IllegalArgumentException("Extension name == null");
        }
    }

createExtension

private T createExtension(String name) { 
        Class<?> clazz = (Class)this.getExtensionClasses().get(name);
        if (clazz == null) { 
            throw this.findException(name);
        } else { 
            try { 
                T instance = EXTENSION_INSTANCES.get(clazz);
                if (instance == null) { 
                    EXTENSION_INSTANCES.putIfAbsent(clazz, clazz.newInstance());
                    instance = EXTENSION_INSTANCES.get(clazz);
                }
    			// 获取实例进行依赖注入
                this.injectExtension(instance);
                // cachedWrapperClasses在loadFile中设置
                Set<Class<?>> wrapperClasses = this.cachedWrapperClasses;
                Class wrapperClass;
                if (wrapperClasses != null && wrapperClasses.size() > 0) { 
                    // 获取带有指定类型参数的构造方法创建实例，进行依赖注入
                    for(Iterator var5 = wrapperClasses.iterator(); var5.hasNext(); instance = this.injectExtension(wrapperClass.getConstructor(this.type).newInstance(instance))) { 
                        wrapperClass = (Class)var5.next();
                    }
                }
    
                return instance;
            } catch (Throwable var7) { 
                throw new IllegalStateException("Extension instance(name: " + name + ", class: " + this.type + ") could not be instantiated: " + var7.getMessage(), var7);
            }
        }
    }

上面方法主要步骤

*  根据name获取对应class
 *  根据class创建实例
 *  获取实例进行依赖注入
 *  获取带有指定类型参数的构造方法创建实例，进行依赖注入

以Protocol为例：

在dubbo-rpc-api的resources路径下，可以找到com.alibaba.dubbo.rcp.Protocol文件中有存在filter/listener

> filter=com.alibaba.dubbo.rpc.protocol.ProtocolFilterWrapper  
> listener=com.alibaba.dubbo.rpc.protocol.ProtocolListenerWrapper

而ProtocolFilterWrapper、ProtocolListenerWrapper都存在以Protocol为参数的构造方法

public ProtocolFilterWrapper(Protocol protocol){ }

public ProtocolListenerWrapper(Protocol protocol){ }

遍历cachedWrapperClass对DubboProtocol 进行包装，会通过ProtocolFilterWrapper、ProtocolListenerWrapper进行包装 。

因为在方法doExportUrlsFor1Protocol中传入的是registryURL，因此Protocol$Adpative的export方法最终会调用RegistryProtocol的export方法

RegistryProtocol

public <T> Exporter<T> export(Invoker<T> originInvoker) throws RpcException { 
        // 执行本地发布>>
        final RegistryProtocol.ExporterChangeableWrapper<T> exporter = this.doLocalExport(originInvoker);
        final Registry registry = this.getRegistry(originInvoker);
        final URL registedProviderUrl = this.getRegistedProviderUrl(originInvoker);
        registry.register(registedProviderUrl);
        final URL overrideSubscribeUrl = this.getSubscribedOverrideUrl(registedProviderUrl);
        final RegistryProtocol.OverrideListener overrideSubscribeListener = new RegistryProtocol.OverrideListener(overrideSubscribeUrl);
        this.overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        return new Exporter<T>() { 
            public Invoker<T> getInvoker() { 
                return exporter.getInvoker();
            }
    
            public void unexport() { 
                try { 
                    exporter.unexport();
                } catch (Throwable var4) { 
                    RegistryProtocol.logger.warn(var4.getMessage(), var4);
                }
    
                try { 
                    registry.unregister(registedProviderUrl);
                } catch (Throwable var3) { 
                    RegistryProtocol.logger.warn(var3.getMessage(), var3);
                }
    
                try { 
                    RegistryProtocol.this.overrideListeners.remove(overrideSubscribeUrl);
                    registry.unsubscribe(overrideSubscribeUrl, overrideSubscribeListener);
                } catch (Throwable var2) { 
                    RegistryProtocol.logger.warn(var2.getMessage(), var2);
                }
    
            }
        };
    }

private <T> RegistryProtocol.ExporterChangeableWrapper<T> doLocalExport(Invoker<T> originInvoker) { 
        String key = this.getCacheKey(originInvoker);
        RegistryProtocol.ExporterChangeableWrapper<T> exporter = (RegistryProtocol.ExporterChangeableWrapper)this.bounds.get(key);
        if (exporter == null) { 
            Map var4 = this.bounds;
            synchronized(this.bounds) { 
                exporter = (RegistryProtocol.ExporterChangeableWrapper)this.bounds.get(key);
                if (exporter == null) { 
                    Invoker<?> invokerDelegete = new RegistryProtocol.InvokerDelegete(originInvoker, this.getProviderUrl(originInvoker));
                    // 这里的protocol不仅仅是Procotol$Adaptive，而是经过Wrapper包装
                    // ProtocolFilterWrapper、ProtocolListenerWrapper
                    exporter = new RegistryProtocol.ExporterChangeableWrapper(this.protocol.export(invokerDelegete), originInvoker);
                    this.bounds.put(key, exporter);
                }
            }
        }
    
        return exporter;
    }

ProtocolFilterWrapper

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
        if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { 
            return protocol.export(invoker);
        }
        return protocol.export(buildInvokerChain(invoker, Constants.SERVICE_FILTER_KEY, Constants.PROVIDER));
    }
    
    private static <T> Invoker<T> buildInvokerChain(final Invoker<T> invoker, String key, String group) { 
        Invoker<T> last = invoker;
        List<Filter> filters = ExtensionLoader.getExtensionLoader(Filter.class).getActivateExtension(invoker.getUrl(), key, group);
        if (filters.size() > 0) { 
            for (int i = filters.size() - 1; i >= 0; i --) { 
                final Filter filter = filters.get(i);
                final Invoker<T> next = last;
                last = new Invoker<T>() { 
    
                    public Class<T> getInterface() { 
                        return invoker.getInterface();
                    }
    
                    public URL getUrl() { 
                        return invoker.getUrl();
                    }
    
                    public boolean isAvailable() { 
                        return invoker.isAvailable();
                    }
    
                    public Result invoke(Invocation invocation) throws RpcException { 
                        return filter.invoke(next, invocation);
                    }
    
                    public void destroy() { 
                        invoker.destroy();
                    }
    
                    @Override
                    public String toString() { 
                        return invoker.toString();
                    }
                };
            }
        }
        return last;
    }

ProtocolListenerWrapper

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
        if (Constants.REGISTRY_PROTOCOL.equals(invoker.getUrl().getProtocol())) { 
            return protocol.export(invoker);
        }
        return new ListenerExporterWrapper<T>(protocol.export(invoker), 
                Collections.unmodifiableList(ExtensionLoader.getExtensionLoader(ExporterListener.class)
                        .getActivateExtension(invoker.getUrl(), Constants.EXPORTER_LISTENER_KEY)));
    }

最后会调用DubboProtocol的export方法

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException { 
        URL url = invoker.getUrl();
        String key = serviceKey(url);
        DubboExporter<T> exporter = new DubboExporter(invoker, key, this.exporterMap);
        this.exporterMap.put(key, exporter);
        Boolean isStubSupportEvent = url.getParameter("dubbo.stub.event", false);
        Boolean isCallbackservice = url.getParameter("is_callback_service", false);
        if (isStubSupportEvent && !isCallbackservice) { 
            String stubServiceMethods = url.getParameter("dubbo.stub.event.methods");
            if (stubServiceMethods != null && stubServiceMethods.length() != 0) { 
                this.stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);
            } else if (this.logger.isWarnEnabled()) { 
                this.logger.warn(new IllegalStateException("consumer [" + url.getParameter("interface") + "], has set stubproxy support event ,but no stub methods founded."));
            }
        }
    	// 暴露服务
        this.openServer(url);
        return exporter;
    }

openServer

// DubboProtocol
    private void openServer(URL url) { 
        String key = url.getAddress();
        boolean isServer = url.getParameter(Constants.IS_SERVER_KEY,true);
        if (isServer) { 
           ExchangeServer server = serverMap.get(key);
           if (server == null) { 
              // 如果没有则创建服务并放入serverMap
              serverMap.put(key, createServer(url));
           } else { 
              server.reset(url);
           }
        }
    }

// DubboProtocol
    private ExchangeServer createServer(URL url) { 
        //默认开启server关闭时发送readonly事件
        url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());
        //默认开启heartbeat
        url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));
        String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);
    
        if (str != null && str.length() > 0 && ! ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))
            throw new RpcException("Unsupported server type: " + str + ", url: " + url);
    
        url = url.addParameter(Constants.CODEC_KEY, Version.isCompatibleVersion() ? COMPATIBLE_CODEC_NAME : DubboCodec.NAME);
        ExchangeServer server;
        try { 
            server = Exchangers.bind(url, requestHandler);
        } catch (RemotingException e) { 
            throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
        }
        str = url.getParameter(Constants.CLIENT_KEY);
        if (str != null && str.length() > 0) { 
            Set<String> supportedTypes = ExtensionLoader.getExtensionLoader(Transporter.class).getSupportedExtensions();
            if (!supportedTypes.contains(str)) { 
                throw new RpcException("Unsupported client type: " + str);
            }
        }
        return server;
    }

Exchangers

public static ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { 
        if (url == null) { 
            throw new IllegalArgumentException("url == null");
        }
        if (handler == null) { 
            throw new IllegalArgumentException("handler == null");
        }
        url = url.addParameterIfAbsent(Constants.CODEC_KEY, "exchange");
        // 调用HeaderExchanger的bind方法
        return getExchanger(url).bind(url, handler);
    }
    
    public static Exchanger getExchanger(URL url) { 
        String type = url.getParameter(Constants.EXCHANGER_KEY, Constants.DEFAULT_EXCHANGER);
        return getExchanger(type);
    }
    
    public static Exchanger getExchanger(String type) { 
        // 获取指定名称扩展点，默认名称为header，因此会返回HeaderExchanger实例
        return ExtensionLoader.getExtensionLoader(Exchanger.class).getExtension(type);
    }

HeaderExchanger

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { 
        return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler))));
    }

Transporters

public static Server bind(URL url, ChannelHandler... handlers) throws RemotingException { 
        if (url == null) { 
            throw new IllegalArgumentException("url == null");
        }
        if (handlers == null || handlers.length == 0) { 
            throw new IllegalArgumentException("handlers == null");
        }
        ChannelHandler handler;
        if (handlers.length == 1) { 
            handler = handlers[0];
        } else { 
            handler = new ChannelHandlerDispatcher(handlers);
        }
        return getTransporter().bind(url, handler);
    }
    
    public static Transporter getTransporter() { 
        // 获取自适应的Transporter实现
        return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
    }

NettyTransporter

public Server bind(URL url, ChannelHandler listener) throws RemotingException {
        return new NettyServer(url, listener);
    }

NettyServer

protected void doOpen() throws Throwable { 
        NettyHelper.setNettyLoggerFactory();
        ExecutorService boss = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerBoss", true));
        ExecutorService worker = Executors.newCachedThreadPool(new NamedThreadFactory("NettyServerWorker", true));
        ChannelFactory channelFactory = new NioServerSocketChannelFactory(boss, worker, getUrl().getPositiveParameter(Constants.IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS));
        bootstrap = new ServerBootstrap(channelFactory);
        
        final NettyHandler nettyHandler = new NettyHandler(getUrl(), this);
        channels = nettyHandler.getChannels();
        // https://issues.jboss.org/browse/NETTY-365
        // https://issues.jboss.org/browse/NETTY-379
        // final Timer timer = new HashedWheelTimer(new NamedThreadFactory("NettyIdleTimer", true));
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() { 
            public ChannelPipeline getPipeline() { 
                NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec() ,getUrl(), NettyServer.this);
                ChannelPipeline pipeline = Channels.pipeline();
                /*int idleTimeout = getIdleTimeout(); if (idleTimeout > 10000) { pipeline.addLast("timer", new IdleStateHandler(timer, idleTimeout / 1000, 0, 0)); }*/
                pipeline.addLast("decoder", adapter.getDecoder());
                pipeline.addLast("encoder", adapter.getEncoder());
                pipeline.addLast("handler", nettyHandler);
                return pipeline;
            }
        });
        // bind
        channel = bootstrap.bind(getBindAddress());
    }

# 服务注册过程 #

![Dubbo发布服务][Dubbo 1]

回到RegistryProtocol的export方法

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { 
        // 本地发布
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        // 获取注册中心
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
        // 注册服务
        registry.register(registedProviderUrl);
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        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);
                }
            }
        };
    }

根据invoker地址获取registry实例

private Registry getRegistry(final Invoker<?> originInvoker){ 
        // registryUrl = registry://172.16.105.128:2181/......
        URL registryUrl = originInvoker.getUrl();
        if (Constants.REGISTRY_PROTOCOL.equals(registryUrl.getProtocol())) { 
            String protocol = registryUrl.getParameter(Constants.REGISTRY_KEY, Constants.DEFAULT_DIRECTORY);
            // registryUrl转换为 zookeeper://172.16.105.128:2181/......
            registryUrl = registryUrl.setProtocol(protocol).removeParameter(Constants.REGISTRY_KEY);
        }
        return registryFactory.getRegistry(registryUrl);
    }

RegistryFactory是一个扩展点且在getRegistry方法上存在一个@Adaptive注解，说明会动态生成一个自适应的适配器，所以就是RegistryFactory$Adpative

@SPI("dubbo")
    public interface RegistryFactory { 
        @Adaptive({ "protocol"})
        Registry getRegistry(URL var1);
    }

RegistryFactory$Adpative

public class RegistryFactory$Adpative implements com.alibaba.dubbo.registry.RegistryFactory { 
        
        public com.alibaba.dubbo.registry.Registry getRegistry(com.alibaba.dubbo.common.URL arg0) { 
            if (arg0 == null) throw new IllegalArgumentException("url == null");
            com.alibaba.dubbo.common.URL url = arg0;
            String extName = (url.getProtocol() == null ? "dubbo" : url.getProtocol());
            if (extName == null)
                throw new IllegalStateException("Fail to get extension(com.alibaba.dubbo.registry.RegistryFactory) name from url(" + url.toString() + ") use keys([protocol])");
            // 这里extName为zookeeper
            com.alibaba.dubbo.registry.RegistryFactory extension = (com.alibaba.dubbo.registry.RegistryFactory) ExtensionLoader.getExtensionLoader(com.alibaba.dubbo.registry.RegistryFactory.class).getExtension(extName);
            // extension为ZookeeperRegistryFactory
            return extension.getRegistry(arg0);
        }
    }

公共逻辑在父类AbstractRegistryFactory中实现，并定义模板方法

public Registry getRegistry(URL url) { 
        url = url.setPath(RegistryService.class.getName()).addParameter("interface", RegistryService.class.getName()).removeParameters(new String[]{ "export", "refer"});
        String key = url.toServiceString();
        LOCK.lock();
    
        Registry var4;
        try { 
            // 从缓存中获取指定key的Registry，没有则创建
            Registry registry = (Registry)REGISTRIES.get(key);
            if (registry == null) { 
                // 创建Registry
                registry = this.createRegistry(url);
                if (registry == null) { 
                    throw new IllegalStateException("Can not create registry " + url);
                }
    
                REGISTRIES.put(key, registry);
                var4 = registry;
                return var4;
            }
    
            var4 = registry;
        } finally { 
            LOCK.unlock();
        }
    
        return var4;
    }
    
    // 模板方法
    protected abstract Registry createRegistry(URL var1);

ZookeeperRegistryFactory

public Registry createRegistry(URL url) { 
        return new ZookeeperRegistry(url, this.zookeeperTransporter);
    }

回到RegistryProtocol的export方法，查看register注册服务方法

public <T> Exporter<T> export(final Invoker<T> originInvoker) throws RpcException { 
        // 本地发布
        final ExporterChangeableWrapper<T> exporter = doLocalExport(originInvoker);
        // 获取注册中心
        final Registry registry = getRegistry(originInvoker);
        final URL registedProviderUrl = getRegistedProviderUrl(originInvoker);
        // 注册服务>>
        registry.register(registedProviderUrl);
        final URL overrideSubscribeUrl = getSubscribedOverrideUrl(registedProviderUrl);
        final OverrideListener overrideSubscribeListener = new OverrideListener(overrideSubscribeUrl);
        overrideListeners.put(overrideSubscribeUrl, overrideSubscribeListener);
        registry.subscribe(overrideSubscribeUrl, overrideSubscribeListener);
        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);
                }
            }
        };
    }

跟踪到ZookeeperRegistry，不存在register方法，而是在其父类FailbackRegistry中

public void register(URL url) { 
        super.register(url); // 将其添加到已注册map中
        failedRegistered.remove(url);
        failedUnregistered.remove(url);
        try { 
            // 执行注册
            doRegister(url);
        } catch (Exception e) { 
            Throwable t = e;
            // 如果开启了启动时检测，则直接抛出异常
            boolean check = getUrl().getParameter(Constants.CHECK_KEY, true)
                    && url.getParameter(Constants.CHECK_KEY, true)
                    && ! Constants.CONSUMER_PROTOCOL.equals(url.getProtocol());
            boolean skipFailback = t instanceof SkipFailbackWrapperException;
            if (check || skipFailback) { 
                if(skipFailback) { 
                    t = t.getCause();
                }
                throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t);
            } else { 
                logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t);
            }
            // 将失败的注册请求记录到失败列表，定时重试
            failedRegistered.add(url);
        }
    }

ZookeeperRegistry

protected void doRegister(URL url) { 
        try { 
            // 调用zkclient在zookeeper中创建节点
           zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true));
        } catch (Throwable e) { 
            throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e);
        }
    }

至此，整个服务发布和注册过程完成。

[dubbo-publish-and-register-01.png]: /images/20220422/26297a96e5aa4ffcb7a71d6fbaa76b80.png
[Dubbo]: /images/20220422/c1f6e3fff5dd452c9dcaeb0258a60a2f.png
[Dubbo 1]: /images/20220422/ac67f422f9f549f7a5be6db91a213115.png