Dubbo学习笔记8：Dubbo的线程模型与线程池策略

拼搏现实的明天。 2022-05-09 11:46 326阅读 0赞

Dubbo默认的底层网络通讯使用的是Netty，服务提供方NettyServer使用两级线程池，其中 EventLoopGroup(boss) 主要用来接受客户端的链接请求，并把接受的请求分发给 EventLoopGroup(worker) 来处理，boss和worker线程组我们称之为IO线程。

如果服务提供方的逻辑能迅速完成，并且不会发起新的IO请求，那么直接在IO线程上处理会更快，因为这减少了线程池调度。

但如果处理逻辑很慢，或者需要发起新的IO请求，比如需要查询数据库，则IO线程必须派发请求到新的线程池进行处理，否则IO线程会阻塞，将导致不能接收其它请求。

Dubbo提供的线程模型

根据请求的消息类被IO线程处理还是被业务线程池处理，Dubbo提供了下面几种线程模型：

*   all : (AllDispatcher类)所有消息都派发到业务线程池，这些消息包括请求/响应/连接事件/断开事件/心跳等，这些线程模型如下图：

![1057539-20180527101500666-905926318.png][]

*  direct : (DirectDispacher类)所有消息都不派发到业务线程池，全部在IO线程上直接执行，模型如下图：

![1057539-20180527101645432-1444766905.png][]

*  message : (MessageOnlyDispatcher类)只有请求响应消息派发到业务线程池，其他连接断开事件/心跳等消息，直接在IO线程上执行，模型图如下：

![1057539-20180527101947012-783716641.png][]

*  execution:(ExecutionDispatcher类)只把请求类消息派发到业务线程池处理，但是响应和其它连接断开事件，心跳等消息直接在IO线程上执行，模型如下图：

![1057539-20180527102219225-1001726464.png][]

*  connection:(ConnectionOrderedDispatcher类)在IO线程上，将连接断开事件放入队列，有序逐个执行，其它消息派发到业务线程池处理，模型如下图：

![1057539-20180527102603109-106667064.png][]

其中AllDispatcher对应的handler代码如下：

![复制代码][copycode.gif]

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 .");
            }
        }
        ...
    }

![复制代码][copycode.gif]

可知所有事件都直接交给业务线程池进行处理了。

Dubbo提供了常用的线程池模型，这些模型可以满足我们绝大多数的需求，但是您可以根据自己的需要进行扩展定制。在服务提供者启动线程时，我们会看到什么时候加载的线程模型的实现。

Dubbo提供的线程池策略

扩展接口 ThreadPool 的SPI实现有如下几种：

*  fixed：固定大小线程池，启动时建立线程，不关闭，一直持有（缺省）。
 *  cached：缓存线程池，空闲一分钟自动删除，需要时重建。
 *  limited：可伸缩线程池，但池中的线程数只会增长不会收缩。只增长不收缩的目的是为了避免收缩时突然带来大流量引起性能问题。

其中fixed策略对应扩展实现类是FixedThreadPool，代码如下：

![复制代码][copycode.gif]

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));
        }
    }

![复制代码][copycode.gif]

可知使用ThreadPoolExecutor创建了核心线程数=最大线程池数=threads的线程池。

Dubbo线程池扩展，这些扩展可以满足我们绝大多数的需求，但是您可以根据自己的需要进行扩展定制。在服务提供者启动流程时，我们会看到什么时候加载的线程池扩展实现。

[1057539-20180527101500666-905926318.png]: /images/20220509/b480f6c748a44952ad20682cfad65bae.png
[1057539-20180527101645432-1444766905.png]: /images/20220509/604f85b3911e42389d4ab0ab76d336c6.png
[1057539-20180527101947012-783716641.png]: /images/20220509/5dd14c475cce4b46ac3af92a8cf114ea.png
[1057539-20180527102219225-1001726464.png]: /images/20220509/80b1affb4efd40b48ab296b98d16e667.png
[1057539-20180527102603109-106667064.png]: /images/20220509/4b451c928cd64af6abdc70f8cde9cecb.png
[copycode.gif]: /images/20220509/e3057bf9c6bc44b78a9932674993aa31.png