• 前言
  最近在看release it，第二章中的案例提到底層服務被資料庫阻塞後把服務執行緒池全部佔滿並導致上層應用一直阻塞，結合自己部門的線上服務考慮，比如一個服務下會http請求外部應用，根據不同引數會處理時間長短會不同，這樣考慮從dubbo中間層做一個保護，比如把處理時間會很長的這種引數的請求歸類到一個執行緒池下，其他的歸類到另外一個執行緒池下，這樣即使處理時間長的請求一直阻塞起碼不影響到其他的功能。
  當然還有其他解決辦法：

  • 增加執行緒池大小或者加機器，但是無法預知什麼時間點會增加到什麼量，即使增加執行緒池大小也無法根治
  • 根據處理時長分服務處理，目前dubbo對於同一應用服務都是使用一個執行緒池處理接收請求，分兩個應用代價太大
  • 減小超時時長，一定程度上可以解決，很有可能大部分正常請求也無法正常響應，比如99%執行時長在7s而設定超時時長為2s顯然不太合適
  • 變同步為非同步，這當然是最能根治的辦法，但是從使用者互動及後臺開發流程都得改變，代價略大

• dubbo網路層結構

  dubbo執行緒模型
  dubbo接收請求基本流程
  netty處理接收請求->NioServerSocketPipelineSink$Boss->註冊NioWorker執行緒用於處理長連線和IO操作->DefaultChannelPipeline（處理請求的反序列化和分發handler，初始化過程見NettyServer，會往pipeline加入decoder、encoder和NettyHandler）->decoder做反序列化->NettyHandler處理請求->NettyServer處理請求（NettyServer初始化封裝了MultiMessageHandler、HeartbeatHandler以及根據dispatcher得到的channelHandler，預設是AllDispatcher）->MultiMessageHandler for each處理批量請求->HeartbeatHandler處理心跳請求->AllDispatcher得到的AllChannelHandler處理將通道所有狀態變更用新執行緒處理（包括建立連線、斷開連線、接收請求、異常）->從指定的執行緒池拿到執行緒執行rpc請求->HeaderExchangeHandler判斷雙向和單向處理請求，雙向則將結果使用當前channel回寫->到DubboProtocol$

  ExchangeHandlerAdapter處理請求呼叫->到此服務端網路傳輸層結束

• 如何自定義一個Dispatcher
  參見dubbo原始碼包/META-INF/dubbo/internal/com.alibaba.dubbo.remoting.Dispather，並在protocol配置時指定為自定義的dispatcher。
  順帶一提，現已有的幾個Dispatcher所實現的功能：

  • AllDispather(注：Dispather不是我打錯了。。類名就是這個)
    轉發給AllChannelHandler，預設200大小的fixed執行緒池，可通過threadpool配置為其他型別執行緒池，由IOThread轉交，處理建立連線、關閉連線、處理接收請求、處理異常四種情況都會使用該執行緒池中的執行緒。
  • ConnectionOrderedDispather
    轉發給ConnectionOrderedChannelHandler，建立連線和關閉連線動作採用同一個單執行緒的執行緒池處理，線上程池佇列超過一定數量提供warn日誌，接收請求和異常處理使用配置的執行緒池。
  • DirectDispather
    直接使用IOThread處理請求。
  • ExecutionDispather
    功能與AllDispatcher類似，更像是AllDispatcher的一個老版本？少了異常處理和獲取executor不存在的處理。
  • MessageOnlyDispather
    只有接收請求才使用配置的執行緒池處理，其他使用IOThread處理。

  還沒理順的一個點是caught、send、receive、connection、disconnection分別是什麼點進入Dispatcher。

• 如何通過自定義Dispatcher完成分執行緒池處理同一服務

  新建自定義Dispatcher和對應的handler，可以根據需要繼承AllDispatcher等，重寫handleReceive，如有必要可再自定義ThreadPool。
  先貼一段Dispatcher和Handler可實現方案，邏輯優化可以再考慮，比如框架不應該使用業務DTO這種,自己的執行緒池（保證全域性唯一，這裡TimeDecideChannelHandler是初始化構造的，所以成員屬性也是全域性唯一的）處理方案等。
  增加配置檔案：
  timeDecide=com.web.foo.dispatcher.TimeDecideDispatcher
  配置dispather：<dubbo:protocol name="dubbo" port="-1" accesslog="d:/access.log" dispather="timeDecide"></dubbo:protocol>

package com.web.foo.dispatcher;

import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.remoting.ChannelHandler;
import com.alibaba.dubbo.remoting.Dispather;
import com.web.foo.handler.TimeDecideChannelHandler;


public class TimeDecideDispatcher implements Dispather
{

     public static final String NAME = "timeDecide";

    @Override
    public ChannelHandler dispath(ChannelHandler handler, URL url)
    {
        return new TimeDecideChannelHandler(handler, url);
    }

}

package com.web.foo.handler;

import java.util.concurrent.ExecutorService;

import com.alibaba.dubbo.common.Constants;
import com.alibaba.dubbo.common.URL;
import com.alibaba.dubbo.common.threadpool.support.fixed.FixedThreadPool;
import com.alibaba.dubbo.remoting.Channel;
import com.alibaba.dubbo.remoting.ChannelHandler;
import com.alibaba.dubbo.remoting.ExecutionException;
import com.alibaba.dubbo.remoting.RemotingException;
import com.alibaba.dubbo.remoting.exchange.Request;
import com.alibaba.dubbo.remoting.transport.dispather.ChannelEventRunnable;
import com.alibaba.dubbo.remoting.transport.dispather.ChannelEventRunnable.ChannelState;
import com.alibaba.dubbo.remoting.transport.dispather.WrappedChannelHandler;
import com.alibaba.dubbo.rpc.Invocation;
import com.web.foo.dto.TestDto;

/**
 * @Project
 * @Description:
 * @Create 2015年7月19日下午6:25:17
 * @author zhoushaoyu
 * 
 */

public class TimeDecideChannelHandler extends WrappedChannelHandler
{
    FixedThreadPool tp = new FixedThreadPool();
    ExecutorService exe;
    /**
     * @param handler
     * @param url
     */
    public TimeDecideChannelHandler(ChannelHandler handler, URL url)
    {
        super(handler, url);
        url = url.addParameter(Constants.THREAD_NAME_KEY, "my-fixed-thread-pool");
        exe = (ExecutorService) tp.getExecutor(url);
    }

    public void connected(Channel channel) throws RemotingException
    {
        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(message);
        try
        {
            cexecutor.execute(new ChannelEventRunnable(channel, handler, ChannelState.RECEIVED, message));
        } catch (Throwable t)
        {
            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 .", t);
        }
    }

    private ExecutorService getExecutorService(Object message)
    {
        Object inv = message;
        if (message instanceof Request)
        {
            inv = ((Request) message).getData();
        }
        if (!(inv instanceof Invocation))
        {
            return getExecutorService();
        }
        for (Object p : ((Invocation) inv).getArguments())
        {
            if (p instanceof String && "longCall".equalsIgnoreCase((String) p))
            {

                return (ExecutorService) exe;
            }
            if (p instanceof TestDto && "longCall".equalsIgnoreCase(((TestDto) p).getTest()))
            {
                return (ExecutorService) exe;
            }
        }
        return getExecutorService();
    }

    private ExecutorService getExecutorService()
    {
        ExecutorService cexecutor = executor;
        if (cexecutor == null || cexecutor.isShutdown())
        {
            cexecutor = SHARED_EXECUTOR;
        }
        return cexecutor;
    }
}

可以看到最終的效果，如果是longCall進來就是在my-fixed-thread-pool，如果是其他的就在DubboServerHandler