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

本文继续上文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 &amp;&amp; !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 &amp;&amp; str.length() &gt; 0 &amp;&amp; !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 &amp;&amp; str.length() &gt; 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接口如下

• ExchangeServer bind(URL url, ExchangeHandler handler) : 服务提供者调用。
• 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接口实现，其继承类图所示： 本文以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系列文章，这里不对每一行代码进行解读，对于与网络相关的参数，将在后续文章中详细讲解，本方法@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服务网络实现原理将在后续章节中详细分析，敬请期待。

---

作者介绍：丁威，《RocketMQ技术内幕》作者，RocketMQ 社区布道师，公众号：中间件兴趣圈 维护者，目前已陆续发表源码分析Java集合、Java 并发包(JUC)、Netty、Mycat、Dubbo、RocketMQ、Mybatis等源码专栏。可以点击链接：中间件知识星球，一起探讨高并发、分布式服务架构，交流源码。

</string></t></t></t></t></t></t></t></t></t></t></t></t></t></t>