【Java核心-进阶】线程池（Executor）

线程池主要类概览

 

Executor

Executor的初衷是将 任务提交 和 任务执行的细节 解耦。
只有一个提交任务的方法：

void execute(Runnable command);

 

ExecutorService

虽然我们习惯将ExecutorService称为线程池，但它并不是简单的线程“池”。
它提供了比较全面的 线程管理 与 任务提交 等方法，如 shutDown、submit。
submit 可以解决 Runnable 无法返回结果的困扰；
其返回的Future提升了任务的可操控性，弥补了 execute 方法的不足：

 Future submit(Callable task);

 

ThreadPoolExecutor

这是最常用到的线程池。Executors中的多个工厂方法内部都用到了此类。

 

ScheduledThreadPoolExecutor

这是对ThreadPoolExecutor扩展，增加了一些调度逻辑，适用于定时或周期性的任务。

 

ForkJoinPool

这是为ForkJoinTask定制的线程池。内部使用 Work-Stealing 算法。
主要是将大问题拆解成小问题，分而治之的方式。也有任务间先后顺序的特性 —— 即解决大问题需先解决小问题。

 

线程池工作原理

关键组件

 

工作队列

工作队列负责暂存用户提交的任务。
它的容量可以是0。如，Executors.newCachedThreadPool 使用容量为0的SynchronousQueue。
也可以设定一个固定的容量值（使用ArrayBlockingQueue 或 LinkedBlockingQueue）。但一般不建议给一个“无界队列”（合理应对OOM风险）

 

内部“线程池”

private final HashSet workers = new HashSet<>();

这个内部的线程“池”是工作线程的集合。
这些内部线程被抽象为内部类 Worker（继承自 AbstractQueuedSynchronizer）。
 

线程池会在运行过程中管理线程的 创建、销毁。
如，带缓冲的线程池（corePoolSize < maximumPoolSize）会，
在任务压力较大时，创建新的工作线程；
当任务压力退去，工作线程空闲一段时间后，又会结束这些空闲线程（默认空闲60秒后回收）。

 

ThreadFactory

ThreadFactory用于创建工作线程实例。
通常会通过它来设置线程的名称，并指定线程是否为“守护（daemon）线程”。
规范有意义的线程名称对于排查异常非常有用。如果JVM只有守护线程在运行，它将会退出。

public interface ThreadFactory {
  Thread newThread(Runnable r);
}

 

RejectedExecutionHandler

任务提交被拒绝时，RejectedExecutionHandler负责处理相关事宜。
线程池已处于关闭状态，或新任务超过线程池的额定负载时，将拒绝新任务。

public interface RejectedExecutionHandler {
  void rejectedExecution(Runnable r, ThreadPoolExecutor executor);
}

ThreadPoolExecutor 中定义了几个RejectedExecutionHandler实现：

• AbortPolicy：这是默认的handler。会抛出一个RejectedExecutionException
• DiscardPolicy：直接悄无声息地抛弃新任务。
• CallerRunsPolicy：由提交任务的线程执行新任务。如果线程池已关闭，将直接悄无声息地抛弃新任务。
• DiscardOldestPolicy：抛弃任务队列中最老的任务，并再次尝试提交新任务。如果线程池已关闭，将直接悄无声息地抛弃新任务。

 

线程池状态转换

 

ThreadPoolExecutor中定义了上图中的5个状态。

 

关键参数与字段

构造方法参数

public ThreadPoolExecutor(int corePoolSize,
                          int maximumPoolSize,
                          long keepAliveTime,
                          TimeUnit unit,
                          BlockingQueue workQueue,
                          ThreadFactory threadFactory,
                          RejectedExecutionHandler handler)

• corePoolSize：所谓的“核心线程数”。这些线程会长期驻留，除非将 allowCoreThreadTimeOut 设置为 true。
• maximumPoolSize：能创建的最大线程数。
• keepAliveTime、TimeUnit：指定额外线程能闲置多久。非核心线程超过闲置时间就会被终结。
• workQueue：工作队列

关键字段：ctl

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));
private static final int COUNT_BITS = Integer.SIZE - 3;
private static final int CAPACITY   = (1 << COUNT_BITS) - 1;

private static final int RUNNING    = -1 << COUNT_BITS;
private static final int SHUTDOWN   =  0 << COUNT_BITS;
private static final int STOP       =  1 << COUNT_BITS;
private static final int TIDYING    =  2 << COUNT_BITS;
private static final int TERMINATED =  3 << COUNT_BITS;

private static int runStateOf(int c)     { return c & ~CAPACITY; }
private static int workerCountOf(int c)  { return c & CAPACITY; }
private static int ctlOf(int rs, int wc) { return rs | wc; }

private static boolean runStateLessThan(int c, int s) { return c < s; }
private static boolean runStateAtLeast(int c, int s) { return c >= s; }
private static boolean isRunning(int c) { return c < SHUTDOWN; }

ctl 字段有双重职责：高3位维护线程池状态，低29位工作线程数。
我们可以指定线程数上限为 Integer.MAX_VALUE，但因为实际系统中的资源限制，不会达到这个值，所以可以将空闲位存储其它信息。
所以根据上述代码，线程数量理论上限就成了 536870911

 

线程池实践策略

避免任务堆积

如果工作线程数太少，处理任务的速度跟不上任务入队的速度，积压的任务就会占用大量内存，甚至OOM。
可通过 jmap 等工具查看是否有大量任务对象入队。

 

避免过度扩展线程

虽然增加工作线程通常可以加快总体任务处理效率，但过多的活动线程会导致大量的上下文切换开销，浪费CPU资源。
线程数的设置需要结合具体业务场景确定。这是一个比较繁琐的实践活动。

 

避免线程泄漏

有问题的任务逻辑（如，死锁）可能导致工作线程迟迟不能被释放，也就是线程泄漏。
可通过 jstack 查看线程栈来排查。（《排查死锁、避免死锁》）

 

谨慎使用 ThreadLocal

因为工作线程存在复用的情况，所以其生命周期通常都会超过任务的生命周期。
在工作线程中使用 ThreadLocal 需非常谨慎，否则可能导致OOM等问题。

 

如何选择合适的线程池大小

线程池大小，即工作线程数，太少会导致任务多度堆积，太多会导致线程切换开销过大。
现实中，通常无法在编码时就精确预计任务压力、任务数据特征等关键信息。
所以需要根据采样或概要分析等方法确定线程池大小，并在后续实践中验证调整相关设置。
 

对于计算密集型的任务，通常使用 CPU核数 N 或 N+1 作为线程数。
对于等待较多任务（如，IO密集型），则通常使用如下公式确定线程数：

线程数 = CPU核数 × 目标CPU利用率 × (1 + 平均等待时间 / 平均工作时间)

 

实际项目中，文件句柄、内存等其它资源也会成为线程数上限的关键制约因素。
（微服务+容器化部署 的模式更容易性能调优。因为任务的特征更稳定，更容易预测，相关资源都是应用独占的，没有其它进程来抢占。）
另，优化架构（包括技术架构和业务架构）很多时候也是非常值得尝试的，不应完全指望调整线程池。

 

Executors提供的几种典型的线程池创建方法

注意

Exectors 这个辅助类提供了多个创建线程池的工厂方法。我们可以借助这些方法快速构建合适的线程池实例。
需要注意的是，在正规产品研发中其实更推荐“手动”创建线程池。
因为这些工厂方法内部所使用的一些默认值可能会导致线程过多降低性能，或OutOfMemoryError。
如：

• newFixedThreadPool 和 newSingleThreadExecutor 使用 LinkedBlockingQueue 作为工作队列，且容量上限为 Integer.MAX_VALUE。
  这可能会耗费很多内存，甚至OOM。
• newCachedThreadPool 和 newScheduledThreadPool 设定的最大线程数为 Integer.MAX_VALUE。
  这可能会创建过多线程，降低性能，甚至OOM。

其实强调“手动”创建线程池是为了提醒代码编写人员有意识地评估处理这些潜在的风险。
此外，线程名称、空闲线程存活时间、因负荷达到上限而拒绝请求的策略 都是一个合格的产品所需要考虑的细节。
如，dubbo中的 CachedThreadPool 就会对线程名称（利用ThreadFactory）和 请求中止策略进行定制。

 

newCachedThreadPool

适用于处理大量短时间工作任务。

它会试图缓存线程并重用；
当无缓存线程时，会新建线程；
如果线程闲置时间超过60秒，会被终止并移除；
长时间闲置时，不会消耗什么资源。
其内部使用 SynchronousQueue 作为工作队列。

 

newFixedThreadPool

重用固定数量的线程（需在创建时指定线程数 nThreads）；
内部使用 LinkedBlockingQueue 作为工作队列（无界队列）。

 

newSingleThreadExecutor

只有 1个 工作线程；
内部使用 LinkedBlockingQueue 作为工作队列（无界队列）；
可保证所有已提交的任务按提交顺序先后执行。

 

newSingleThreadScheduledExecutor

用于执行 定时 或 周期性 任务；
只有 1个 工作线程。
内部使用 DelayedWorkQueue 作为工作队列；这是非常定制化的队列。

 

newScheduledThreadPool

用于执行 定时 或 周期性 任务；
与 newSingleThreadScheduledExecutor 的区别是，它可以有多个工作线程（通过参数指定最小工作线程数）。

 

newWorkStealingPool

使用频度不高，经常被忽略。
内部使用 ForkJoinPool 和 Work-Stealing 算法并发处理任务。