Dubbo學習筆記8：Dubbo的線程模型與線程池策略

Dubbo默認的底層網絡通訊使用的是Netty，服務提供方NettyServer使用兩級線程池，其中 EventLoopGroup(boss) 主要用來接受客戶端的鏈接請求，並把接受的請求分發給 EventLoopGroup(worker) 來處理，boss和worker線程組我們稱之為IO線程。

如果服務提供方的邏輯能迅速完成，並且不會發起新的IO請求，那么直接在IO線程上處理會更快，因為這減少了線程池調度。

但如果處理邏輯很慢，或者需要發起新的IO請求，比如需要查詢數據庫，則IO線程必須派發請求到新的線程池進行處理，否則IO線程會阻塞，將導致不能接收其它請求。

Dubbo提供的線程模型

根據請求的消息類被IO線程處理還是被業務線程池處理，Dubbo提供了下面幾種線程模型：

•  all : (AllDispatcher類)所有消息都派發到業務線程池，這些消息包括請求/響應/連接事件/斷開事件/心跳等，這些線程模型如下圖：

• direct : (DirectDispacher類)所有消息都不派發到業務線程池，全部在IO線程上直接執行，模型如下圖：

• message : (MessageOnlyDispatcher類)只有請求響應消息派發到業務線程池，其他連接斷開事件/心跳等消息，直接在IO線程上執行，模型圖如下：

• execution:(ExecutionDispatcher類)只把請求類消息派發到業務線程池處理，但是響應和其它連接斷開事件，心跳等消息直接在IO線程上執行，模型如下圖：

• connection:(ConnectionOrderedDispatcher類)在IO線程上，將連接斷開事件放入隊列，有序逐個執行，其它消息派發到業務線程池處理，模型如下圖：

其中AllDispatcher對應的handler代碼如下：

public class AllChannelHandler extends WrappedChannelHandler{
    public AllChannelHandler(ChannelHandler handler , URL url){
        super(handler,url);
    }

    // 鏈接事件，交給業務線程池處理
    public void connected(Channel channel) throws RemotingExcecption{
        ExecutorService cexecutor = getExecutorService();
        try{
            cexecutor.execute(new ChannelEventRunnable(channel,handler,ChannelState.CONNECTED));
        }catch(Throwable t){
            throw new ExecutionException("connect event" , channel , getClass() + "error when process connected event.",t);
        }
    }

    // 鏈接斷開事件，交給業務線程池處理
    public void disconnected(Channel channel) throws RemotingException{
        ExecutorService cexecutor = getExecutorService();
        try{
            cexecutor.execute(new ChannelEventRunnable(channel,handler,ChannelState.DISCONNECTED));
        }catch(Throwable t){
            throw new ExecutionException("disconnect event",channel,getClass()+" error when process disconnected event.",t);
        }
    }

    // 請求響應事件，交給業務線程池處理
    public void received(Channel channel , Object message) throws RemotingException{        
        ExecutorService cexecutor = getExecutorService();
        try{
            cexecutor.execute(new ChannelEventRunnable(channel,handler,ChannelState.RECEIVED,message));
        }catch(Throwable t){
            // TODO 臨時解決線程池滿后異常信息無法發送到對端的問題。待重構
            // fix 線程池滿了拒絕調用不返回，導致消費者一直等待超時
            if(message instanceof Request && t instanceof RejectedExecutionException ){
                ...
            }
            throw new ExecutionException(message , channel ,getClass() + " error when process received event.",t);
        }
    }

    // 異常處理事件，交給業務線程池處理
    public void caught(Channel channel , Throwable exception) throws RemotingException {
        ExecutorService cexecutor = getExecutorService();
        try{
            cexecutor.execute(new ChannelEventRunnable(channel,handler,ChannelState.CAUGHT,exception));
        }catch(Throwable t){
            throw new ExecutionException("caught event",channel,getClass() + " error when process caught event .");
        }
    }
    ...
}

可知所有事件都直接交給業務線程池進行處理了。

Dubbo提供了常用的線程池模型，這些模型可以滿足我們絕大多數的需求，但是您可以根據自己的需要進行擴展定制。在服務提供者啟動線程時，我們會看到什么時候加載的線程模型的實現。

Dubbo提供的線程池策略

擴展接口 ThreadPool 的SPI實現有如下幾種：

• fixed：固定大小線程池，啟動時建立線程，不關閉，一直持有（缺省）。
• cached：緩存線程池，空閑一分鍾自動刪除，需要時重建。
• limited：可伸縮線程池，但池中的線程數只會增長不會收縮。只增長不收縮的目的是為了避免收縮時突然帶來大流量引起性能問題。

其中fixed策略對應擴展實現類是FixedThreadPool，代碼如下：

public class FixedThreadPool implements ThreadPool{
    public Executor getExecutor(URL url){
        String name = url.getParameter(Constants.THREAD_NAME_KEY,Constants.DEFAULT_THREAD_NAME);
        int threads = url.getParameter(Constants.THREADS_KEY,Constants.DEFAULT_THREADS);
        int queues = url.getParameter(Constants.QUEUES_KEY,Constants.DEFAULT_QUEUES);
        return new ThreadPoolExecutor(threads , threads , 0 , TimeUnit.MILLISECONDS , queues==0 ? new SynchronousQueue<Runnable>() : (queue < 0 ? new LinkedBlockingQueue<Runnable>(queues)) , new NamedThreadFactory(name,true) , new AbortPolicyWithReport(name,url));
    }
}

可知使用ThreadPoolExecutor創建了核心線程數=最大線程池數=threads的線程池。

Dubbo線程池擴展，這些擴展可以滿足我們絕大多數的需求，但是您可以根據自己的需要進行擴展定制。在服務提供者啟動流程時，我們會看到什么時候加載的線程池擴展實現。