Java 线程池 Thread Pool

更多 Java 并发编程方面的文章，请参见文集《Java 并发编程》

线程组 Thread Group

线程的集合，不推荐使用。

线程池 Thread Pool 介绍

所在包：java.util.concurrent

线程池用于多线程处理中，它可以根据系统的情况，可以有效控制线程执行的数量，优化运行效果。
线程池做的工作主要是控制运行的线程的数量，处理过程中将任务放入队列，然后在线程创建后启动这些任务，如果线程数量超过了最大数量，超出数量的线程排队等候，等其它线程执行完毕，再从队列中取出任务来执行。

线程池的3个优点：

线程复用，不需要频繁的创建和销毁线程
控制最大并发数，提高系统资源利用率，同时避免过多的资源竞争，避免堵塞
管理线程

目的：减少大量创建和销毁线程带来的时间和空间上的消耗。

线程池的组成

一般的线程池主要分为以下4个组成部分：

线程池管理器：用于创建并管理线程池
工作线程：线程池中的线程
任务接口：每个任务必须实现的接口，用于工作线程调度其运行
任务队列：用于存放待处理的任务，提供一种缓冲机制

线程池框架

Java 中的线程池是通过 Executor 框架实现的，该框架中用到了 Executor，Executors，ExecutorService，ThreadPoolExecutor ，Callable 和 Future、FutureTask 这几个类。

Executor：所有线程池的接口，只有一个方法
Executors：Executor 的工厂类，提供了创建各种不同线程池的方法，返回的线程池都实现了ExecutorService 接口
ThreadPoolExecutor：线程池的具体实现类，一般所有的线程池都是基于这个类实现的

ThreadPoolExecutor 的构造方法如下：

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue) {
    this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
         Executors.defaultThreadFactory(), defaultHandler);
}

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue<Runnable> workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler) {
    if (corePoolSize < 0 ||
        maximumPoolSize <= 0 ||
        maximumPoolSize < corePoolSize ||
        keepAliveTime < 0)
        throw new IllegalArgumentException();
    if (workQueue == null || threadFactory == null || handler == null)
        throw new NullPointerException();
    this.corePoolSize = corePoolSize;
    this.maximumPoolSize = maximumPoolSize;
    this.workQueue = workQueue;
    this.keepAliveTime = unit.toNanos(keepAliveTime);
    this.threadFactory = threadFactory;
    this.handler = handler;
}

其中：

corePoolSize：
- 线程池的核心线程数，线程池中运行的线程数也永远不会超过 corePoolSize 个
- 在没有设置 allowCoreThreadTimeOut 为 true 的情况下，核心线程会在线程池中一直存活，即使处于闲置状态
maximumPoolSize：
- 线程池中允许的最大线程数
- 当活动线程(核心线程+非核心线程)达到这个数值后，后续任务将会根据 RejectedExecutionHandler 来进行拒绝策略处理
keepAliveTime：
- 非核心线程 闲置时的超时时长。超过该时长，非核心线程就会被回收。
- 若线程池通过 allowCoreThreadTimeOut() 方法设置 allowCoreThreadTimeOut 属性为 true，则该时长同样会作用于核心线程，AsyncTask 配置的线程池就是这样设置的
unit：
- 是一个枚举，它表示的是 keepAliveTime 的单位
workQueue：
- 工作队列，用于存放不能被及时处理掉的任务的一个队列，通过线程池的 execute() 方法提交的 Runnable 对象会存储在该队列中
- 它是一个 BlockingQueue 类型。关于 BlockingQueue，虽然它是 Queue 的子接口，但是它的主要作用并不是容器，而是作为线程同步的工具，他有一个特征，当生产者试图向 BlockingQueue 放入(put)元素，如果队列已满，则该线程被阻塞；当消费者试图从 BlockingQueue 取出(take)元素，如果队列已空，则该线程被阻塞。
ThreadFactory：
- 线程工厂，功能很简单，就是为线程池提供创建新线程的功能
- 这是一个接口，可以通过自定义，做一些自定义线程名的操作
- 默认使用 DefaultThreadFactory：

static class DefaultThreadFactory implements ThreadFactory {
    private static final AtomicInteger poolNumber = new AtomicInteger(1);
    private final ThreadGroup group;
    private final AtomicInteger threadNumber = new AtomicInteger(1);
    private final String namePrefix;

    DefaultThreadFactory() {
        SecurityManager s = System.getSecurityManager();
        group = (s != null) ? s.getThreadGroup() :
                              Thread.currentThread().getThreadGroup();
        namePrefix = "pool-" +
                      poolNumber.getAndIncrement() +
                     "-thread-";
    }

    public Thread newThread(Runnable r) {
        Thread t = new Thread(group, r,
                              namePrefix + threadNumber.getAndIncrement(),
                              0);
        if (t.isDaemon())
            t.setDaemon(false);
        if (t.getPriority() != Thread.NORM_PRIORITY)
            t.setPriority(Thread.NORM_PRIORITY);
        return t;
    }
}

RejectedExecutionHandler：当任务无法被执行时(超过线程最大容量 maximumPoolSize 并且 workQueue 已经被排满了)的处理策略，这里有四种任务拒绝类型
- CallerRunsPolicy：当线程池中的数量等于最大线程数时，重试添加当前的任务；它会自动重复调用execute() 方法。
- AbortPolicy: 默认策略，当线程池中的数量等于最大线程数时抛 java.util.concurrent.RejectedExecutionException 异常
- DiscardPolicy：当线程池中的数量等于最大线程数时，默默丢弃不能执行的新加任务，不报任何异常。
- DiscardOldestPolicy：当线程池中的数量等于最大线程数时，抛弃线程池中工作队列头部的任务(即等待时间最久的任务)，并执行新传入的任务。

线程池的工作过程

线程池刚创建时，里面没有一个线程。
当调用 execute() 方法添加一个任务时，线程池会做如下判断：
- 如果正在运行的线程数量小于 corePoolSize，那么马上创建 核心线程 运行这个任务；
- 如果正在运行的线程数量大于或等于 corePoolSize，那么将这个任务放入工作队列 workQueue；
  - 如果这时候队列满了，而且正在运行的线程数量小于 maximumPoolSize，那么还是要创建 非核心线程 立刻运行这个任务；
  - 如果队列满了，而且正在运行的线程数量大于或等于 maximumPoolSize，那么线程池会抛出异常RejectExecutionException。
当一个线程完成任务时，它会从工作队列 workQueue 中取下一个任务来执行。
当一个 非核心线程 无事可做，超过一定的时间（keepAliveTime）时，线程池会判断，如果当前运行的线程数大于 corePoolSize，那么这个 非核心线程 就被停掉。所以线程池的所有任务完成后，它最终会收缩到 corePoolSize 的大小。

常见的 Java 线程池

Executors：提供一系列工厂方法来创建线程池，返回的线程池都实现了 ExecutorService 接口：

SingleThreadExecutor
- 单个线程的线程池，即线程池中每次只有一个线程在运行，单线程串行执行任务。如果这个唯一的线程因为异常结束，那么会有一个新的线程来替代它。此线程池保证所有任务的执行顺序按照任务的提交顺序执行。
- Executors 类中方法定义：public static ExecutorService newSingleThreadExecutor()
- 实现：

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

FixedThreadPool
- 固定数量的线程池，只有核心线程，每提交一个任务就是一个线程，直到达到线程池的最大数量，然后后面进入等待队列，直到前面的任务完成才继续执行。线程池的大小一旦达到最大值就会保持不变，如果某个线程因为执行异常而结束，那么线程池会补充一个新线程。
- Executors 类中方法定义：public static ExecutorService newFixedThreadPool(int nThreads)
- 实现：

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

CachedThreadPool
- 可缓存线程池。如果线程池的大小超过了处理任务所需要的线程，那么就会回收部分空闲（60秒不执行任务）的线程，当任务数增加时，此线程池又可以智能的添加新线程来处理任务。此线程池不会对线程池大小做限制，线程池大小完全依赖于操作系统（或者说JVM）能够创建的最大线程大小。其中，SynchronousQueue 是一个是缓冲区为1的阻塞队列。
- Executors 类中方法定义：public static ExecutorService newCachedThreadPool()
- 实现：

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

ScheduledThreadPool
- 核心线程池固定，大小无限制的线程池，支持定时和周期性的执行线程。创建一个周期性执行任务的线程池。如果闲置，非核心线程池会在 DEFAULT_KEEPALIVEMILLIS 时间内回收。
Executors 类中方法定义：public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)
- 实现：

public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize) {
    return new ScheduledThreadPoolExecutor(corePoolSize);
}

public ScheduledThreadPoolExecutor(int corePoolSize) {
    super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
          new DelayedWorkQueue());
}

注意事项：

线程池 ExecutorService 中可以使用 execute(t) 或者 submit(t) 方法加入多个任务，任务的执行顺序与加入顺序无关
- 关于 execute(t) 和 submit(t) 的区别，请参见 Java ExecutorService
如果线程池中某个线程因为执行异常而结束，线程池会自动补充一个新线程

示例：

public class ThreadPool_Test {
    public static void main(String[] args) {
        ExecutorService es = Executors.newSingleThreadExecutor();

        for (int i = 0; i < 5; i++) {
            es.execute(new MyThread());
        }

        es.shutdown();
    }

    static class MyThread extends Thread {
        public void run() {
            System.out.println(Thread.currentThread().getName());
        }
    }
}

以上代码的输出如下：
因为创建的线程池中只有一个线程，因此每次只会有一个任务执行，getName() 都会得到相同的结果。

pool-1-thread-1
pool-1-thread-1
pool-1-thread-1
pool-1-thread-1

pool-1-thread-1

如果将 Executors.newSingleThreadExecutor() 更改为 Executors.newFixedThreadPool(2)，创建一个有两个线程的线程池，则输出如下：

pool-1-thread-2
pool-1-thread-1
pool-1-thread-2
pool-1-thread-1
pool-1-thread-2

Thread Factory

在我们通过 Executors 来创建线程池的时候，还可以传入一个参数 ThreadFactory，例如：
public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)

An object that creates new threads on demand. Using thread factories removes hardwiring of calls to new Thread.

ThreadFactory 遵循了工厂模式，提供 newThread() 方法用来创建线程。
示例：我们希望创建一个线程，它能够在产生 不能捕获的 Runtime Exception 时能自动 Logging：

public class ThreadFactory_Test {
    public static void main(String[] args) {
        ExecutorService es = Executors.newSingleThreadExecutor(new MyThreadFactory());

        es.execute(new MyThread());

        es.shutdown();
    }

    static class MyThread extends Thread {
        public void run() {
            // Runtime exception
            int i = 1 / 0;
        }
    }

    static class MyThreadFactory implements ThreadFactory {
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r);

            t.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() {
                @Override
                public void uncaughtException(Thread t, Throwable e) {
                    System.err.println(t.getName() + e.getMessage());
                }
            });

            return t;
        }
    }
}

线程池的关闭

可以通过调用线程池的 shutdown 或 shutdownNow 方法来关闭线程池，但是它们的实现原理不同：

shutdown的原理是只是将线程池的状态设置成 SHUTDOWN 状态，然后中断所有没有正在执行任务的线程
shutdownNow 的原理是遍历线程池中的工作线程，然后逐个调用线程的 interrupt 方法来中断线程，所以无法响应中断的任务可能永远无法终止。shutdownNow 会首先将线程池的状态设置成 STOP，然后尝试停止所有的正在执行或暂停任务的线程，并返回等待执行任务的列表

只要调用了这两个关闭方法的其中一个，isShutdown 方法就会返回 true。当所有的任务都已关闭后,才表示线程池关闭成功，这时调用 isTerminaed 方法会返回 true。

线程池的监控

通过线程池提供的参数进行监控。
线程池里有一些属性在监控线程池的时候可以使用：

public long getTaskCount()：线程池需要执行的任务数量。
public long getCompletedTaskCount()：线程池在运行过程中已完成的任务数量。小于或等于taskCount。
public int getPoolSize()：线程池的线程数量。如果线程池不销毁的话，池里的线程不会自动销毁，所以这个大小只增不减。
public int getLargestPoolSize()：线程池曾经创建过的最大线程数量。通过这个数据可以知道线程池是否满过。如等于线程池的最大大小，则表示线程池曾经满了。
public int getActiveCount()：获取活动的线程数。

通过扩展线程池进行监控。通过继承线程池并重写线程池的 beforeExecute，afterExecute 和 terminated方法，我们可以在任务执行前，执行后和线程池关闭前干一些事情。如监控任务的平均执行时间，最大执行时间和最小执行时间等。这几个方法在线程池里是空方法。如：

protected void beforeExecute(Thread t, Runnable r) { }

合理的配置线程池

任务的性质：CPU密集型任务，IO密集型任务和混合型任务。
任务的优先级：高，中和低。
任务的执行时间：长，中和短。
任务的依赖性：是否依赖其他系统资源，如数据库连接。

任务性质不同的任务可以用不同规模的线程池分开处理：

CPU密集型任务配置尽可能少的线程数量，如配置 Ncpu+1个线程的线程池。
IO密集型任务则由于需要等待IO操作，线程并不是一直在执行任务，则配置尽可能多的线程，如2*Ncpu。
混合型的任务，如果可以拆分，则将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐率要高于串行执行的吞吐率，如果这两个任务执行时间相差太大，则没必要进行分解。
我们可以通过 Runtime.getRuntime().availableProcessors() 方法获得当前设备的CPU个数。

优先级不同的任务可以使用优先级队列 PriorityBlockingQueue 来处理。它可以让优先级高的任务先得到执行，需要注意的是如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行。

执行时间不同的任务可以交给不同规模的线程池来处理，或者也可以使用优先级队列，让执行时间短的任务先执行。

依赖数据库连接池的任务，因为线程提交 SQL 后需要等待数据库返回结果，如果等待的时间越长 CPU 空闲时间就越长，那么线程数应该设置越大，这样才能更好的利用CPU。

引用：
ThreadFactory usage in Java
Java线程池的简介以及使用
 关于线程池的执行原则及配置参数详解
 聊聊并发（3）：Java线程池的分析和使用

最后编辑于：2019.03.03 04:55:58

人面猴
序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
沈念sama阅读 206,602评论 6赞 481
死咒
序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
沈念sama阅读 88,442评论 2赞 382
救了他两次的神仙让他今天三更去死
文/潘晓璐我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
开封第一讲书人阅读 152,878评论 0赞 344
道士缉凶录：失踪的卖姜人
文/不坏的土叔我叫张陵，是天一观的道长。经常有香客问我，道长，这世上最难降的妖魔是什么？我笑而不...
开封第一讲书人阅读 55,306评论 1赞 279
港岛之恋（遗憾婚礼）
正文为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
茶点故事阅读 64,330评论 5赞 373
恶毒庶女顶嫁案：这布局不是一般人想出来的
文/花漫我一把揭开白布。她就那样静静地躺着，像睡着了一般。火红的嫁衣衬着肌肤如雪。梳的纹丝不乱的头发上，一...
开封第一讲书人阅读 49,071评论 1赞 285
城市分裂传说
那天，我揣着相机与录音，去河边找鬼。笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。我是一名探鬼主播，决...
沈念sama阅读 38,382评论 3赞 400
双鸳鸯连环套：你想象不到人心有多黑
文/苍兰香墨我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
开封第一讲书人阅读 37,006评论 0赞 259
万荣杀人案实录
序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
沈念sama阅读 43,512评论 1赞 300
护林员之死
正文独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋以下内容为张勋视角年9月15日...
茶点故事阅读 35,965评论 2赞 325
白月光启示录
正文我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
茶点故事阅读 38,094评论 1赞 333
活死人
序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
沈念sama阅读 33,732评论 4赞 323
日本核电站爆炸内幕
正文年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
茶点故事阅读 39,283评论 3赞 307
男人毒药：我在死后第九天来索命
文/蒙蒙一、第九天我趴在偏房一处隐蔽的房顶上张望。院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
开封第一讲书人阅读 30,286评论 0赞 19
一桩弑父案，背后竟有这般阴谋
文/苍兰香墨我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。一阵脚步声响...
开封第一讲书人阅读 31,512评论 1赞 262
情欲美人皮
我被黑心中介骗来泰国打工，没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。一个月前我还...
沈念sama阅读 45,536评论 2赞 354
代替公主和亲
正文我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
茶点故事阅读 42,828评论 2赞 345