1. 程式人生 > >Dubbo原始碼學習之-通過原始碼看看dubbo對netty的使用

Dubbo原始碼學習之-通過原始碼看看dubbo對netty的使用

前言

    前段時間,從頭開始將netty原始碼瞭解了個大概,但都是原理上理解。剛好博主對dubbo框架了解過一些,這次就以dubbo框架為例,詳細看看dubbo這種出色的開源框架是如何使用netty的,又是如何與框架本身邏輯進行融合的。

    本文分成兩大部分,一部分是dubbo服務端對netty的封裝,一部分是dubbo客戶端對netty的封裝,而每部分都分netty初始化和呼叫兩個階段,下面進入正題。

一、Dubbo服務端

    Dubbo服務端對netty的呼叫始於服務匯出,服務匯出的流程之前文章中有介紹,詳見【https://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方法一文【https://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型別的),用該執行緒池進行後續操作。

     最後,來一張示意圖做個呼叫的總結:

&n