Java多线程探索（一）：为什么要使用ThreadPoolExecutor？

人生苦短，不如养狗

一、前言

前段时间闲鱼在重新翻看《阿里巴巴Java开发手册》时看到

【强制】线程池不允许使用Executors创建，而是使用ThreadPoolExecutor的方式创建。
那么为什么这么规定呢？

二、一些关于多线程的灵魂发问

在开始回答上面的问题之前，我们先来回答一下多线程相关的问题热热身。

1.为什么要使用多线程？

其实可以将这个问题替换成，使用单线程处理问题有什么不足？单线程意味着所有的线程都是串行工作，也就是每个线程都必须等待上一个线程全部处理完成之后才能开始工作。当某个线程需要处理一个极大的文件时，此时用户就只能呆呆地等在电脑前直到这个线程处理完成之后才能进行下一项任务的处理。
而当引入多线程的概念之后，所有的线程可以并发的进行工作。注意，这里的并发执行指的是同一段时间内同时进行，但是从微观来看仍然是串行进行(CPU是单核的情况)。那么有同学会疑惑，既然微观上仍然是串行，为什么说多线程在用户体验上会由于单线程。这里就要归功于线程调度策略，在引入多线程的概念之后，每个线程在使用CPU时都有固定的时间片，如果执行时间超过规定的时间片，那么就需要将CPU让给其他的线程进行使用。从宏观上来看，多个线程同时在进行工作，也就是上面所说的同一段时间同时进行。除此以外，当线程出现由于IO操作等发生阻塞时，也会将资源让给其他线程进行使用。因此，从用户角度来说，多线程的引入提升了用户的体验感。

这里只是介绍了最基本的一些线程调度的知识，诸如线程状态流转等知识，大家可以自行Google，这里就不再赘述。

2.在什么场景下使用多线程？

回答了上面的问题，那么新的问题来了：我们什么时候应该使用多线程呢？从上一个问题的回答中我们可以发现，当对用户响应要求比较高，同时又允许用户并发访问的时候，此时就非常适合使用多线程。还有一种情况就是程序存在需要耗时计算或者处理大文件时，出现阻塞情况时，为了提高程序的响应，会通过创建子线程的方式来进行任务的处理。

3.为什么要使用线程池？

既然多线程这么优秀，那是否能够肆无忌惮的去使用呢？
当然不行。任何看似优秀的东西，使用都是需要付出代价的。就好比女生的化妆品、包包，好看吗？好看。想拥有吗？想。但是一个问题，这些东西都很贵啊。
线程的使用也是这样。每一个线程的创建都需要占用内存，线程的调度也需要CPU进行协调和管理，同时线程之间对共享资源的访问还会导致线程安全问题（这里暂时不讨论线程安全问题），等等一系列需要考虑的问题。这些都是使用多线程的“成本”。
所以无限制、不加管理的使用线程是不可能的，那么如何合理使用多线程呢？—— 线程池。固定的创建一些线程放在线程池中，当有任务来时从中获取线程进行任务的执行，如果线程池中所有的线程都正在使用时，那么将任务放置在阻塞队列中等待空闲的线程池。这样的设计理念既能保证享受到多线程的优势，又能防止无限制、无管理的使用线程的危害。

三、为什么要使用ThreadPoolExecutor而不是Executors？

经过了这么多的灵魂拷问，终于来到了今天最重要的问题：为什么要使用ThreadPoolExecutor？
其实答案很简单，来欣赏一下Executors提供的基础的四种线程池，欣赏完之后，大家就应该明白了：

public static ExecutorService newFixedThreadPool(int nThreads) {
    return new ThreadPoolExecutor(nThreads, nThreads,
                                  0L, TimeUnit.MILLISECONDS,
                                  new LinkedBlockingQueue<Runnable>());
}

public static ExecutorService newSingleThreadExecutor() {
    return new FinalizableDelegatedExecutorService
        (new ThreadPoolExecutor(1, 1,
                                0L, TimeUnit.MILLISECONDS,
                                new LinkedBlockingQueue<Runnable>()));
}

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(0, Integer.MAX_VALUE,
                                  60L, TimeUnit.SECONDS,
                                  new SynchronousQueue<Runnable>());
}

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
	  // 这里底层使用的是ThreadPoolExecutor
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

可以看到这四种线程池底层使用的都是ThreadPoolExecutor，只不过对于相应的参数Executors已经贴心的帮开发们设置好了。但正是Executors将底层的具体细节进行封装，使得开发无法进行线程池执行过程的掌控和根据实际情况进行线程池的修改。对于多线程的使用来说，这是一个很危险的事情。所以，为了能够让开发能够详细了解到线程池的运作机制，在《阿里巴巴Java开发手册》中推荐使用ThreadPoolExecutor而不是Executors来创建线程池。

四、一个简单的小例子

话不多说，下面我们就用一个例子赏析一下ThreadPoolExecutor是如何工作的。

具体使用到的代码如下：
TestThreadFactory

public class TestThreadFactory implements ThreadFactory {

    /**
     * 姓名前缀
     */
    private final String namePrefix;

    private final AtomicInteger nextId = new AtomicInteger(0);

    public TestThreadFactory(String whatFeatureOfGroup){
        this.namePrefix = "From TestThreadFactory's " + whatFeatureOfGroup + "-Worker-";
    }

    @Override
    public Thread newThread(Runnable task) {
        String name = namePrefix
                + nextId.getAndIncrement();
        Thread thread = new Thread(null, task, name, 0);
        System.out.println(thread.getName());
        return thread;
    }
}

ThreadPoolManager

public class ThreadPoolManager {

    /**
     * 核心线程大小
     */
    private int corePoolSize;
    /**
     * 最大线程池大小
     */
    private int maximumPoolSize;
    /**
     * 保持时间
     */
    private int keepAliveTime;

    /**
     * 时间单位
     */
    private TimeUnit timeUnit;

    /**
     * 阻塞队列
     */
    private BlockingQueue blockingQueue;

    /**
     * 线程工厂
     */
    private ThreadFactory threadFactory;

    /**
     * 线程饱和策略
     */
    private RejectedExecutionHandler handler;

    /**
     * 构造函数
     *
     * @param corePoolSize
     * @param maximumPoolSize
     * @param keepAliveTime
     * @param timeUnit
     * @param blockingQueue
     * @param threadFactory
     * @param handler
     */
    public ThreadPoolManager(int corePoolSize, int maximumPoolSize, int keepAliveTime, TimeUnit timeUnit, BlockingQueue blockingQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler){
        this.corePoolSize = corePoolSize;
        this.maximumPoolSize = maximumPoolSize;
        this.keepAliveTime = keepAliveTime;
        this.timeUnit = timeUnit;
        this.blockingQueue = blockingQueue;
        this.threadFactory = threadFactory;
        this.handler = handler;
    }


    /**
     * 创建线程池
     *
     * @return
     */
    public ThreadPoolExecutor createThreadPool(){
        return new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, blockingQueue, threadFactory, handler);
    }
}

TestMain

public class TestMain {

    public static void main(String[] args) {
        ThreadPoolManager threadPoolManager = new ThreadPoolManager(1,2,3,
                TimeUnit.MILLISECONDS, new ArrayBlockingQueue<Runnable>(3), new TestThreadFactory("测试"), new ThreadPoolExecutor.AbortPolicy());
        ThreadPoolExecutor threadPoolExecutor = threadPoolManager.createThreadPool();
        for(int i = 0; i<5; i++){
            try {
                threadPoolExecutor.execute(() -> {
                    System.out.println(Thread.currentThread().getName() + " test is running");
                    try {
						   // 用于更方便的看清执行情况
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });
            } catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}