Dubbo源碼學習之-通過源碼看看dubbo對netty的使用
前言
前段時間，從頭開始將netty源碼了解了個大概，但都是原理上理解。剛好部落客對dubbo框架了解過一些，這次就以dubbo框架為例，詳細看看dubbo這種出色的開源框架是如何使用netty的，又是如何與框架本身邏輯進行融合的。
本文分成兩大部分，一部分是dubbo服務端對netty的封裝，一部分是dubbo客戶端對netty的封裝，而每部分都分netty初始化和調用兩個階段，下面進入正題。
一、Dubbo服務端
Dubbo服務端對netty的調用始於服務導出，服務導出的流程之前文章中有介紹，詳見【//www.cnblogs.com/zzq6032010/p/11275478.html】，在服務導出的最後，會調用DubboProtocol#openServer方法，就是在此方法中完成的netty服務端的初始化（本文均以配置了netty通訊為前提），下面就以該處作為起點探尋。
1、服務端初始化
openServer方法源碼如下，主體邏輯是先獲取了address作為key—ip:port格式的字元串，然後做了一個雙重檢查，server不存在則調createServer創建一個放入serverMap中。到這裡我們可以知道，dubbo服務提供者中一個ip+埠對應一個nettyServer，所有的nettyServer統一放在一個ConcurrentHashMap中維護了起來。但其實通常情況下，一個服務提供者的伺服器，只會暴露一個埠給dubbo用，故雖然用Map存起來，但一般只會有一個nettyServer。此處還要注意，dubbo中是暴露一個服務提供者執行一次export方法，即一個服務提供者介面觸發一次openServer方法、對應一個nettyServer，下面跟進server的創建過程。
1 private void openServer(URL url) { 2 // find server. 3 String key = url.getAddress(); 4 //client can export a service which's only for server to invoke 5 boolean isServer = url.getParameter(IS_SERVER_KEY, true); 6 if (isServer) { 7 ProtocolServer server = serverMap.get(key); 8 if (server == null) { 9 synchronized (this) { 10 server = serverMap.get(key); 11 if (server == null) { 12 serverMap.put(key, createServer(url)); 13 } 14 } 15 } else { 16 // server supports reset, use together with override 17 server.reset(url); 18 } 19 } 20 }
openServer調用的方法棧如下所示：
進入NettyTransporter的bind方法，NettyTransporter一共有兩個方法-bind和connect，前者初始化服務端時調用，後者初始化客戶端時觸發，源碼如下：
1 public class NettyTransporter implements Transporter { 2 public static final String NAME = "netty"; 3 @Override 4 public RemotingServer bind(URL url, ChannelHandler listener) throws RemotingException { 5 return new NettyServer(url, listener); 6 } 7 @Override 8 public Client connect(URL url, ChannelHandler listener) throws RemotingException { 9 return new NettyClient(url, listener); 10 } 11 }
下面看NettyServer如何與netty關聯起來的。先看下NettyServer的類圖：
有經驗的園友看到類圖估計就能猜到，此處是源碼框架中常用的分層抽象，AbstractServer作為一個模板的抽象，繼承它之後可以擴展出其他類型的通訊，比如MinaServer、GrizzlyServer。下面回到本文的主角NettyServer，看看其構造器：
1 public NettyServer(URL url, ChannelHandler handler) throws RemotingException { 2 super(ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME), ChannelHandlers.wrap(handler, url)); 3 }
設置了一下url中的執行緒名參數，將handler和url進行了封裝，然後調用了父類AbstractServer的構造器。
到這裡，需要確定好入參的handler類型和傳給父類構造器的handler類型。NettyServer構造器入參ChannelHandler是在HeaderExchanger#bind中封裝的，方式如下：
1 public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { 2 return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); 3 }
再進一步，bind方法入參ExchangeHandler的實現類要追溯到DubboProtocol，是其成員變數requestHandler如下：
private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() { // 省略若干個重寫的方法邏輯 }
至此，NettyServer構造器入參ChannelHandler的類型已經確認了，其內部最終實現是DubboProtocol中的ExchangeHandlerAdapter，外部封裝了一層HeaderExchangeHandler，又封裝了一層DecodeHandler。簡圖如下：
搞清楚NettyServer構造器入參的ChannelHandler之後，下面跟進ChannelHandlers.wrap方法，最終封裝方法如下：
1 protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) { 2 return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class) 3 .getAdaptiveExtension().dispatch(handler, url))); 4 }
而Dispatcher默認是AllDispatcher，其dispatch方法如下：
1 public ChannelHandler dispatch(ChannelHandler handler, URL url) { 2 return new AllChannelHandler(handler, url); 3 }
至此，ChannelHandlers.wrap方法執行完後得到的ChannelHandler結構如下，採用的是裝飾器模式，層層裝飾。
了解清楚了wrap方法，下面回到主線，進入AbstractServer的構造器：
1 public AbstractServer(URL url, ChannelHandler handler) throws RemotingException { 2 // 1、調用父類構造器將這兩個變數存起來，最終是存在了AbstractPeer中 3 super(url, handler); 4 // 2、設置兩個address 5 localAddress = getUrl().toInetSocketAddress(); 6 String bindIp = getUrl().getParameter(Constants.BIND_IP_KEY, getUrl().getHost()); 7 int bindPort = getUrl().getParameter(Constants.BIND_PORT_KEY, getUrl().getPort()); 8 if (url.getParameter(ANYHOST_KEY, false) || NetUtils.isInvalidLocalHost(bindIp)) { 9 bindIp = ANYHOST_VALUE; 10 } 11 bindAddress = new InetSocketAddress(bindIp, bindPort); 12 this.accepts = url.getParameter(ACCEPTS_KEY, DEFAULT_ACCEPTS); 13 this.idleTimeout = url.getParameter(IDLE_TIMEOUT_KEY, DEFAULT_IDLE_TIMEOUT); 14 try { 15 // 3、完成netty源碼的調用-開啟netty服務端 16 doOpen(); 17 } catch (Throwable t) { 18 // 省略異常處理 19 } 20 // 4、獲取/創建執行緒池 21 executor = executorRepository.createExecutorIfAbsent(url); 22 }
1/2的邏輯較簡單，3和4才是重點，下面進入3處的doOpen方法，doOpen方法在AbstractServer中是抽象方法，所以要到其子類NettyServer中看：
1 protected void doOpen() throws Throwable { 2 // 這裡可以看到熟悉的netty程式碼了 3 bootstrap = new ServerBootstrap(); 4 // bossGroup一個執行緒 5 bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory("NettyServerBoss", true)); 6 // workerGroup執行緒數取的CPU核數+1與32的小值 7 workerGroup = new NioEventLoopGroup(getUrl().getPositiveParameter(IO_THREADS_KEY, Constants.DEFAULT_IO_THREADS), 8 new DefaultThreadFactory("NettyServerWorker", true)); 9 // ***1、此處將NettyServer封裝進NettyServerHandler中，實現了netty和dubbo的連接 10 final NettyServerHandler nettyServerHandler = new NettyServerHandler(getUrl(), this); 11 channels = nettyServerHandler.getChannels(); 12 // netty封裝 13 bootstrap.group(bossGroup, workerGroup) 14 .channel(NioServerSocketChannel.class) 15 .childOption(ChannelOption.TCP_NODELAY, Boolean.TRUE) 16 .childOption(ChannelOption.SO_REUSEADDR, Boolean.TRUE) 17 .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) 18 .childHandler(new ChannelInitializer<NioSocketChannel>() { 19 @Override 20 protected void initChannel(NioSocketChannel ch) throws Exception { 21 int idleTimeout = UrlUtils.getIdleTimeout(getUrl()); 22 NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyServer.this); 23 if (getUrl().getParameter(SSL_ENABLED_KEY, false)) { 24 ch.pipeline().addLast("negotiation", 25 SslHandlerInitializer.sslServerHandler(getUrl(), nettyServerHandler)); 26 } 27 ch.pipeline() 28 .addLast("decoder", adapter.getDecoder()) 29 .addLast("encoder", adapter.getEncoder()) 30 .addLast("server-idle-handler", new IdleStateHandler(0, 0, idleTimeout, MILLISECONDS)) 31 .addLast("handler", nettyServerHandler); 32 } 33 }); 34 // 綁定IP和埠，此處用到的就是AbstractServer中的bindAddress變數 35 ChannelFuture channelFuture = bootstrap.bind(getBindAddress()); 36 channelFuture.syncUninterruptibly(); 37 channel = channelFuture.channel(); 38 }
標星號***的1處就是關鍵點，看下面的pipeline.addLast可知，存放著dubbo邏輯的NettyServer被封裝進了NettyServerHandler中，進而放入了pipeline裡面，當有客戶端連接的時候就會觸發這個nettyServerHandler中的對應方法，進入dubbo的介面調用邏輯。從dubbo功能到netty框架之間的連接者就是這個NettyServerHandler類。NettyServer中封裝了一個執行緒池，即一個客戶端連接過來之後，服務端用一個執行緒池來接收處理這個客戶端的一系列請求，即在netty原有執行緒模型基礎上又加了一層執行緒池。
4中的executorRepository.createExecutorIfAbsent(url)用於生成執行緒池，此處為服務端，點進去源碼可以發現在dubbo的服務端，一個port埠對應一個執行緒池，而且此處未設置特殊的參數，故走ThreadPool的默認類型fixed，即FixedThreadPool的getExecutor方法：
1 public Executor getExecutor(URL url) { 2 String name = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME); 3 int threads = url.getParameter(THREADS_KEY, DEFAULT_THREADS); 4 int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES); 5 return new ThreadPoolExecutor(threads, threads, 0, TimeUnit.MILLISECONDS, 6 queues == 0 ? new SynchronousQueue<Runnable>() : 7 (queues < 0 ? new LinkedBlockingQueue<Runnable>() 8 : new LinkedBlockingQueue<Runnable>(queues)), 9 new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url)); 10 }
該方法可關注兩點：執行緒數默認為200個，阻塞隊列由於queues==0採用的是SynchronousQueue。這個執行緒池初始化之後幹啥用？搜遍了調用關係，發現只有在NettyServer進行重置或者關閉時才會操作這個執行緒池。但理論上講不通啊，總不能創建了一個執行緒池之後只是為了關閉它。且往下看。
2、服務端調用
其實服務端的執行緒池這裡給部落客看源碼一個啟發，注意，此處是去倉庫獲取一個執行緒池的引用（即executorRepository.createExecutorIfAbsent(url)），而倉庫創建了執行緒池是將其快取了起來，而快取之後的執行緒池引用還可以暴露給其他地方，在其他地方執行執行緒池的execute方法。具體在這裡，最終是在AllChannelHandler中調用的執行緒池，比如connected方法，如下所示，getExecutorService方法就是去倉庫中獲取了服務端的這個執行緒池，封裝出一個ChannelEventRunnable丟給執行緒池執行。而服務端接收到請求時的received方法也是同樣的處理流程。
1 public void connected(Channel channel) throws RemotingException { 2 ExecutorService executor = getExecutorService(); 3 try { 4 executor.execute(new ChannelEventRunnable(channel, handler, ChannelState.CONNECTED)); 5 } catch (Throwable t) { 6 throw new ExecutionException("connect event", channel, getClass() + " error when process connected event .", t); 7 } 8 }
下面結合netty的調用流程對服務調用時的處理流程做一個梳理：在之前講解netty的run方法一文【//www.cnblogs.com/zzq6032010/p/13122483.html】中有過介紹，netty的ServerBootstrap啟動後，會開啟bossGroup中的那個執行緒（即Reactor執行緒），一直執行run方法。而當有客戶端要連接時，select方法會從作業系統獲取到一個連接事件，Reactor執行緒會為該連接方創建一個NioSocketChannel，並從workerGroup中挑選一個執行緒，運行run方法，該執行緒用於處理服務端與這個客戶端的後續通訊。而此處添加進pipeline中的nettyServerHandler會在客戶端傳來讀寫請求時觸發對應的方法。最終調用到上述AllChannelHandler中的對應方法，用執行緒池執行後續業務邏輯。
二、Dubbo的客戶端
1、客戶端初始化
dubbo客戶端初始化時會調用RegistryProtocol的refer方法，幾經周折，最後到了DubboProtocol的protocolBindingRefer方法，如下，其中第5行調用的getClients方法是與netty整合的重點，即生成連接服務端的客戶端。注意此處是在客戶端中每一個引入的服務介面對應一個DubboInvoker。
1 public <T> Invoker<T> protocolBindingRefer(Class<T> serviceType, URL url) throws RpcException { 2 optimizeSerialization(url); 3 4 // create rpc invoker. 5 DubboInvoker<T> invoker = new DubboInvoker<T>(serviceType, url, getClients(url), invokers); // 為每個invoker生成對應的nettyClient 6 invokers.add(invoker); 7 8 return invoker; 9 }
繼續跟進，會進入NettyTransporter的connect方法，到這裡應該會很熟悉，因為服務端初始化時調用的是該類下面的bind方法。bind方法初始化的是NettyServer對象，而connect初始化的是NettyClient對象。
public class NettyTransporter implements Transporter { public static final String NAME = "netty"; @Override public RemotingServer 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); } }
NettyClient的類圖結構與NettyServer類似：
下面看NettyClient的構造器：
1 public NettyClient(final URL url, final ChannelHandler handler) throws RemotingException { 2 // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants. 3 // the handler will be warped: MultiMessageHandler->HeartbeatHandler->handler 4 super(url, wrapChannelHandler(url, handler)); 5 }
直接調用了父類構造器，其中wrapChannelHandler方法與NettyServer中的一樣，不再贅述。下面看父類構造器：
1 public AbstractClient(URL url, ChannelHandler handler) throws RemotingException { 2 super(url, handler); 3 4 needReconnect = url.getParameter(Constants.SEND_RECONNECT_KEY, false); 5 // 1、初始化客戶端執行緒池 6 initExecutor(url); 7 8 try { 9 doOpen(); // 2、創建客戶端的Bootstrap 10 } catch (Throwable t) { 11 // 省略異常處理 12 } 13 try { 14 // 3、連接Netty服務端 15 connect(); 16 } catch (RemotingException t) { 17 // 省略異常處理 18 } catch (Throwable t) { 19 close(); 20 // 拋異常 21 } 22 }
主要有三步，已經在上面標出，下面分別跟進這三個方法。
1)、initExecutor方法直接先將執行緒池類型添加進url中，客戶端默認是Cached類型，所以在調用executorRepository.createExecutorIfAbsent(url)時會進入CachedThreadPool中。
1 private void initExecutor(URL url) { 2 url = ExecutorUtil.setThreadName(url, CLIENT_THREAD_POOL_NAME); 3 url = url.addParameterIfAbsent(THREADPOOL_KEY, DEFAULT_CLIENT_THREADPOOL); 4 executor = executorRepository.createExecutorIfAbsent(url); 5 }
CachedThreadPool程式碼如下，可見是創建的核心執行緒數為0最大執行緒數無上限的執行緒池，阻塞隊列默認SynchronousQueue。
1 public class CachedThreadPool implements ThreadPool { 2 3 @Override 4 public Executor getExecutor(URL url) { 5 String name = url.getParameter(THREAD_NAME_KEY, DEFAULT_THREAD_NAME); 6 int cores = url.getParameter(CORE_THREADS_KEY, DEFAULT_CORE_THREADS); 7 int threads = url.getParameter(THREADS_KEY, Integer.MAX_VALUE); 8 int queues = url.getParameter(QUEUES_KEY, DEFAULT_QUEUES); 9 int alive = url.getParameter(ALIVE_KEY, DEFAULT_ALIVE); 10 return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS, 11 queues == 0 ? new SynchronousQueue<Runnable>() : 12 (queues < 0 ? new LinkedBlockingQueue<Runnable>() 13 : new LinkedBlockingQueue<Runnable>(queues)), 14 new NamedInternalThreadFactory(name, true), new AbortPolicyWithReport(name, url)); 15 } 16 }
2)、doOpen方法
實現邏輯在NettyClient中，都是正常的封裝，變化的地方是將NettyClientHandler放入pipeline中。注意此處只是將Bootstrap初始化，但並未觸發與服務端的連接。
3)、connect方法
該方法最終在NettyClient的doConnect方法中調用了bootstrap的connect方法，完成與服務端的連接。
2、客戶端調用
在消費端調用服務端介面或者接收到服務端的返回結果時，均會觸發NettyClientHandler中對應的方法，而此處跟NettyServerHandler類似，最終都是在AllChannelHandler中獲取之前創建的客戶端執行緒池（Cached類型的），用該執行緒池進行後續操作。
最後，來一張示意圖做個調用的總結：
