源码分析Dubbo服务提供者启动流程-下篇

谁践踏了优雅 2022-05-22 10:43 177阅读 0赞

本文继续上文Dubbo服务提供者启动流程，在上篇文章中详细梳理了从dubbo spring文件开始，Dubbo是如何加载配置文件，服务提供者dubbo:service标签服务暴露全流程，本节重点关注RegistryProtocol\#export中调用doLocalExport方法，其实主要是根据各自协议，服务提供者建立网络服务器，在特定端口建立监听，监听来自消息消费端服务的请求。   
RegistryProtocol\#doLocalExport:

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) {
                        final Invoker<?> invokerDelegete = new InvokerDelegete<T>(originInvoker, getProviderUrl(originInvoker));   // @1
                        exporter = new ExporterChangeableWrapper<T>((Exporter<T>) protocol.export(invokerDelegete), originInvoker);    // @2
                        bounds.put(key, exporter);
                    }
                }
            }
            return exporter;
        }

代码@1：如果服务提供者以dubbo协议暴露服务，getProviderUrl(originInvoker)返回的URL将以dubbo://开头。   
代码@2：根据Dubbo内置的SPI机制，将调用DubboProtocol\#export方法。   
1、源码分析DubboProtocol\#export

public <T> Exporter<T> export(Invoker<T> invoker) throws RpcException {
            URL url = invoker.getUrl();     // @1
            // export service.
            String key = serviceKey(url);      // @2
            DubboExporter<T> exporter = new DubboExporter<T>(invoker, key, exporterMap);
            exporterMap.put(key, exporter);
    
            //export an stub service for dispatching event
            Boolean isStubSupportEvent = url.getParameter(Constants.STUB_EVENT_KEY, Constants.DEFAULT_STUB_EVENT);    //@3 start
            Boolean isCallbackservice = url.getParameter(Constants.IS_CALLBACK_SERVICE, false);                                                  
            if (isStubSupportEvent && !isCallbackservice) {                                                                                                                        
                String stubServiceMethods = url.getParameter(Constants.STUB_EVENT_METHODS_KEY);
                if (stubServiceMethods == null || stubServiceMethods.length() == 0) {
                    if (logger.isWarnEnabled()) {
                        logger.warn(new IllegalStateException("consumer [" + url.getParameter(Constants.INTERFACE_KEY) +
                                "], has set stubproxy support event ,but no stub methods founded."));
                    }
                } else {
                    stubServiceMethodsMap.put(url.getServiceKey(), stubServiceMethods);                                                                      
                }
            }   // @3 end
    
            openServer(url);   // @4
            optimizeSerialization(url);  // @5
            return exporter;                
        }

代码@1：获取服务提供者URL，以协议名称，这里是dubbo://开头。   
代码@2：从服务提供者URL中获取服务名，key: interface:port，例如：com.alibaba.dubbo.demo.DemoService:20880。   
代码@3：是否将转发事件导出成stub。   
代码@4：根据url打开服务，下面将详细分析其实现。   
代码@5：根据url优化器序列化方式。   
2、源码分析DubboProtocol\#openServer

private void openServer(URL url) {
            // find server.
            String key = url.getAddress();    // @1
            //client can export a service which's only for server to invoke
            boolean isServer = url.getParameter(Constants.IS_SERVER_KEY, true);
            if (isServer) {
                ExchangeServer server = serverMap.get(key);           // @2
                if (server == null) {
                    serverMap.put(key, createServer(url));                    //@3
                } else {
                    // server supports reset, use together with override
                    server.reset(url);                                                       //@4
                }
            }
        }

代码@1：根据url获取网络地址：ip:port，例如：192.168.56.1:20880，服务提供者IP与暴露服务端口号。   
代码@2：根据key从服务器缓存中获取，如果存在，则执行代码@4，如果不存在，则执行代码@3.   
代码@3：根据URL创建一服务器，Dubbo服务提供者服务器实现类为ExchangeServer。   
代码@4：如果服务器已经存在，用当前URL重置服务器，这个不难理解，因为一个Dubbo服务中，会存在多个dubbo:service标签，这些标签都会在服务台提供者的同一个IP地址、端口号上暴露服务。   
2.1 源码分析DubboProtocol\#createServer

private ExchangeServer createServer(URL url) {
            // send readonly event when server closes, it's enabled by default
            url = url.addParameterIfAbsent(Constants.CHANNEL_READONLYEVENT_SENT_KEY, Boolean.TRUE.toString());    // @1
            // enable heartbeat by default
            url = url.addParameterIfAbsent(Constants.HEARTBEAT_KEY, String.valueOf(Constants.DEFAULT_HEARTBEAT));     // @2
            String str = url.getParameter(Constants.SERVER_KEY, Constants.DEFAULT_REMOTING_SERVER);  // @3
    
            if (str != null && str.length() > 0 && !ExtensionLoader.getExtensionLoader(Transporter.class).hasExtension(str))    // @4
                throw new RpcException("Unsupported server type: " + str + ", url: " + url);
    
            url = url.addParameter(Constants.CODEC_KEY, DubboCodec.NAME);       // @5
            ExchangeServer server;
            try {
                server = Exchangers.bind(url, requestHandler);    // @6
            } catch (RemotingException e) {
                throw new RpcException("Fail to start server(url: " + url + ") " + e.getMessage(), e);
            }
            str = url.getParameter(Constants.CLIENT_KEY);     //@7
            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;
        }

代码@1：为服务提供者url增加channel.readonly.sent属性，默认为true，表示在发送请求时，是否等待将字节写入socket后再返回，默认为true。   
代码@2：为服务提供者url增加heartbeat属性，表示心跳间隔时间，默认为60\*1000，表示60s。   
代码@3：为服务提供者url增加server属性，可选值为netty,mina等等，默认为netty。   
代码@4：根据SPI机制，判断server属性是否支持。   
代码@5：为服务提供者url增加codec属性，默认值为dubbo，协议编码方式。   
代码@6：根据服务提供者URI,服务提供者命令请求处理器requestHandler构建ExchangeServer实例。requestHandler的实现具体在以后详细分析Dubbo服务调用时再详细分析。   
代码@7：验证客户端类型是否可用。   
2.1.1 源码分析Exchangers.bind 根据URL、ExchangeHandler构建服务器

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");
            return getExchanger(url).bind(url, handler);
        }

上述代码不难看出，首先根据url获取Exchanger实例，然后调用bind方法构建ExchangeServer，Exchanger接口如下   
![这里写图片描述][70]   
1)ExchangeServer bind(URL url, ExchangeHandler handler) : 服务提供者调用。   
2)ExchangeClient connect(URL url, ExchangeHandler handler):服务消费者调用。   
dubbo提供的实现类为：HeaderExchanger，其bind方法如下：   
HeaderExchanger\#bind

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

从此次可以看到，端口的绑定由Transporters的bind方法实现。   
2.1.2 源码分析Transporters.bind方法

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() {
            return ExtensionLoader.getExtensionLoader(Transporter.class).getAdaptiveExtension();
    }

从这里得知，Dubbo网络传输的接口有Transporter接口实现，其继承类图所示：   
![这里写图片描述][70 1]   
本文以netty版本来查看一下Transporter实现。   
NettyTransporter源码如下：

public class NettyTransporter implements Transporter { 
    
        public static final String NAME = "netty";
    
        @Override
        public Server bind(URL url, ChannelHandler listener) throws RemotingException {
            return new NettyServer(url, listener);
        }
    
        @Override
        public Client connect(URL url, ChannelHandler listener) throws RemotingException {
            return new NettyClient(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);      // @1
            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() {
                @Override
                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);     // @2
                    return pipeline;
                }
            });
            // bind
            channel = bootstrap.bind(getBindAddress());
        }

熟悉本方法需要具备Netty的知识，有关源码阅读Netty系列文章：[https://blog.csdn.net/prestigeding/article/details/53977445][https_blog.csdn.net_prestigeding_article_details_53977445]，这里不对每一行代码进行解读，对于与网络相关的参数，将在《源码梳理Dubbo网络模块实现原理（基于Netty)》章节中详细关注，本方法@1、@2引起了我的注意，首先创建NettyServer必须传入一个服务提供者URL，但从DubboProtocol\#createServer中可以看出，Server是基于网络套接字（ip:port）缓存的，一个JVM应用中，必然会存在多个dubbo:server标签，就会有多个URL，这里为什么可以这样做呢？从DubboProtocol\#createServer中可以看出，在解析第二个dubbo:service标签时并不会调用createServer,而是会调用Server\#reset方法，是不是这个方法有什么魔法，在reset方法时能将URL也注册到Server上，那接下来分析NettyServer\#reset方法是如何实现的。   
2.2源码分析DdubboProtocol\#reset   
reset方法最终将用Server的reset方法，同样还是以netty版本的NettyServer为例，查看reset方法的实现原理。   
NettyServer\#reset—>父类(AbstractServer)   
AbstractServer\#reset

public void reset(URL url) {
            if (url == null) {
                return;
            }
            try {                                                                                                       // @1 start
                if (url.hasParameter(Constants.ACCEPTS_KEY)) {
                    int a = url.getParameter(Constants.ACCEPTS_KEY, 0);
                    if (a > 0) {
                        this.accepts = a;
                    }
                }
            } catch (Throwable t) {
                logger.error(t.getMessage(), t);
            }
            try {
                if (url.hasParameter(Constants.IDLE_TIMEOUT_KEY)) {
                    int t = url.getParameter(Constants.IDLE_TIMEOUT_KEY, 0);
                    if (t > 0) {
                        this.idleTimeout = t;
                    }
                }
            } catch (Throwable t) {
                logger.error(t.getMessage(), t);
            }
            try {
                if (url.hasParameter(Constants.THREADS_KEY)
                        && executor instanceof ThreadPoolExecutor && !executor.isShutdown()) {
                    ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executor;
                    int threads = url.getParameter(Constants.THREADS_KEY, 0);
                    int max = threadPoolExecutor.getMaximumPoolSize();
                    int core = threadPoolExecutor.getCorePoolSize();
                    if (threads > 0 && (threads != max || threads != core)) {
                        if (threads < core) {
                            threadPoolExecutor.setCorePoolSize(threads);
                            if (core == max) {
                                threadPoolExecutor.setMaximumPoolSize(threads);
                            }
                        } else {
                            threadPoolExecutor.setMaximumPoolSize(threads);
                            if (core == max) {
                                threadPoolExecutor.setCorePoolSize(threads);
                            }
                        }
                    }
                }
            } catch (Throwable t) {
                logger.error(t.getMessage(), t);
            }              // @1 end
            super.setUrl(getUrl().addParameters(url.getParameters()));    // @2
        }

代码@1：首先是调整线程池的相关线程数量，这个好理解。、   
代码@2：然后设置调用setUrl覆盖原先NettyServer的private volatile URL url的属性，那为什么不会影响原先注册的dubbo:server呢？   
原来NettyHandler上加了注解：@Sharable，由该注解去实现线程安全。

Dubbo服务提供者启动流程将分析到这里了，本文并未对网络细节进行详细分析，旨在梳理出启动流程，有关Dubbo服务网络实现原理将在后续章节中详细分析，敬请期待。

转载自[https://blog.csdn.net/prestigeding/article/details/80588629][https_blog.csdn.net_prestigeding_article_details_80588629]

[70]: /images/20220522/262896d304364b908844b454a3ea5405.png
[70 1]: /images/20220522/7cac97848f8b4ec88f39831e3c251d2e.png
[https_blog.csdn.net_prestigeding_article_details_53977445]: https://blog.csdn.net/prestigeding/article/details/53977445
[https_blog.csdn.net_prestigeding_article_details_80588629]: https://blog.csdn.net/prestigeding/article/details/80588629