Dubbo-线程模型与线程池策略

1.线程池策略       

resources目录下的com.alibaba.dubbo.common.threadpool.ThreadPool的文件

fixed=com.alibaba.dubbo.common.threadpool.support.fixed.FixedThreadPool
cached=com.alibaba.dubbo.common.threadpool.support.cached.CachedThreadPool
limited=com.alibaba.dubbo.common.threadpool.support.limited.LimitedThreadPool
           

说明：

• dubbo的线程池策略通过SPI配置文件对外提供，在com.alibaba.dubbo.common.threadpool.ThreadPool文件当中定义。
• dubbo的线程池策略对外提供了三种策略，分别是fixed、cached、limited三类。
• 每类策略的定义见下述源码。

配置如：

<dubbo:protocol name="dubbo" dispatcher="all" threadpool="fixed" threads="100" />
           

1.1 CachedThreadPool

public class CachedThreadPool implements ThreadPool {

    public Executor getExecutor(URL url) {
        String name = url.getParameter(Constants.THREAD_NAME_KEY, Constants.DEFAULT_THREAD_NAME);
        int cores = url.getParameter(Constants.CORE_THREADS_KEY, Constants.DEFAULT_CORE_THREADS);
        int threads = url.getParameter(Constants.THREADS_KEY, Integer.MAX_VALUE);
        int queues = url.getParameter(Constants.QUEUES_KEY, Constants.DEFAULT_QUEUES);
        int alive = url.getParameter(Constants.ALIVE_KEY, Constants.DEFAULT_ALIVE);
        return new ThreadPoolExecutor(cores, threads, alive, TimeUnit.MILLISECONDS,
                queues == 0 ? new SynchronousQueue<Runnable>() :
                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }
}
           

说明：

• CachedThreadPool的实现可以理解为JDK当中Executors.newCachedThreadPool()方法。

1.2 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>() :
                        (queues < 0 ? new LinkedBlockingQueue<Runnable>()
                                : new LinkedBlockingQueue<Runnable>(queues)),
                new NamedThreadFactory(name, true), new AbortPolicyWithReport(name, url));
    }
}
           

说明：

• FixedThreadPool的实现可以理解为JDK当中Executors.newFixedThreadPool()方法。

1.3 LimitedThreadPool

public class LimitedThreadPool implements ThreadPool {

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

}
           

说明：

• LimitedThreadPool的实现可以理解为JDK当中Executors.newCachedThreadPool()方法。
• LimitedThreadPool的区别在于线程池中的线程永远不会过期，因为alive时间为最大值。

1.4 NamedThreadFactory

public class NamedThreadFactory implements ThreadFactory {
    private static final AtomicInteger POOL_SEQ = new AtomicInteger(1);

    private final AtomicInteger mThreadNum = new AtomicInteger(1);

    private final String mPrefix;

    private final boolean mDaemon;

    private final ThreadGroup mGroup;

    public NamedThreadFactory() {
        this("pool-" + POOL_SEQ.getAndIncrement(), false);
    }

    public NamedThreadFactory(String prefix) {
        this(prefix, false);
    }

    public NamedThreadFactory(String prefix, boolean daemon) {
        mPrefix = prefix + "-thread-";
        mDaemon = daemon;
        SecurityManager s = System.getSecurityManager();
        mGroup = (s == null) ? Thread.currentThread().getThreadGroup() : s.getThreadGroup();
    }

    public Thread newThread(Runnable runnable) {
        String name = mPrefix + mThreadNum.getAndIncrement();
        Thread ret = new Thread(mGroup, runnable, name, 0);
        ret.setDaemon(mDaemon);
        return ret;
    }

    public ThreadGroup getThreadGroup() {
        return mGroup;
    }
}
           

说明：

• 提供了线程池中线程创建的工厂，推荐使用线程池时候一定要自定义线程池工厂，便于定位线程的用途。

2.线程模型

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

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

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

        具体配置方式如下两种：

<dubbo:provider dispatcher="xxx" />
           

        或者

<dubbo:protocol name="dubbo" dispatcher="xxx" />
           

目前dubbo提供的Dispatcher扩展实现有如下5种实现，默认派发方式是all

resources目录下的com.alibaba.dubbo.remoting.Dispatcher文件

all=com.alibaba.dubbo.remoting.transport.dispatcher.all.AllDispatcher
direct=com.alibaba.dubbo.remoting.transport.dispatcher.direct.DirectDispatcher
message=com.alibaba.dubbo.remoting.transport.dispatcher.message.MessageOnlyDispatcher
execution=com.alibaba.dubbo.remoting.transport.dispatcher.execution.ExecutionDispatcher
connection=com.alibaba.dubbo.remoting.transport.dispatcher.connection.ConnectionOrderedDispatcher
           

说明：

• Dubbo线程模型在resources目录下的com.alibaba.dubbo.remoting.Dispatcher文件当中。
• 提供5类线程模型，分别是all、direct、message、execution、connection。
• 每类的用途见下面的介绍。

2.1 线程模型对比

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

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

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

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

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

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