再谈ThreadPoolExecutor

上次提到ThreadPoolExecutor浅析分析了ThreadPoolExecutor的核心参数，本文则主要分析下常用队列、拒绝策略和线程池初始化方法。

常用队列

• SynchronousQueue

直接提交
适用：一组请求有依赖关系，所以串行
需要：maximumPoolSizes 无界或无限大，防止丢任务
隐患：任务增长速度超过处理速度，OOM

• LinkedBlockingQueue

无界队列，默认 capacity=Integer.MAX_VALUE
适用：任务相互独立，无任何依赖和影响
问题1：核心线程忙碌时，任务需要等待
问题2：队列无界，只有corePoolSize个线程，maximumPoolSize无意义
问题3：任务增长速度超过处理速度，OOM

• ArrayBlockingQueue

有界队列
适用：任务相互独立，无任何依赖和影响
优势：防止资源耗尽
参数难调整：
a. large queues and small pool sizes
节省系统资源和CPU开销、减少上下文切换，但人为造成吞吐量下降
b. small queues requires larger pool sizes
CPU 使用率较高，可能遇到不可接受的调度开销，这样也会降低吞吐量。

• 建议
  不同领域的人员给出了不同的公式，暂无统一的定论和公式
  实际研发过程应综合评估CPU性能、任务复杂度、流量大小、业务重要性和实时性要求等多个因素。

以下仅为建议：

> CPU密集型任务
尽量使用较小的线程池，一般为CPU核心数+1。
> IO密集型任务
可以使用稍大的线程池，一般为2*CPU核心数。
> 混合型任务
可以将任务分成IO密集型和CPU密集型任务，然后分别用不同的线程池去处理。
1
2
3
4
5
6

常用拒绝策略

RejectedExecutionHandler#rejectedExecution 触发场景：
1. 线程池 shut down
2. 队列满且maximum threads都在忙

• ThreadPoolExecutor.AbortPolicy

默认此策略
抛出拒绝任务的异常 RejectedExecutionException

• ThreadPoolExecutor.CallerRunsPolicy

线程池没有 shut down, 调用execute的线程执行当前任务
线程池 shut down,当前任务被忽略

• ThreadPoolExecutor.DiscardPolicy

忽略当前任务

• ThreadPoolExecutor.DiscardOldestPolicy

线程池没有 shut down, 抛弃队列头部的任务
线程池 shut down,当前任务被忽略

常见的线程池

• newFixedThreadPool (定长线程池)
  该线程池的最大线程数等于核心线程数，所以在默认情况下，该线程池的线程不会因为闲置状态超时而被销毁。
  如果当前线程数小于核心线程数，并且也有闲置线程的时候提交了任务，这时也不会去复用之前的闲置线程，会创建新的线程去执行任务。如果当前执行任务数大于了核心线程数，大于的部分就会进入队列等待。等着有闲置的线程来执行这个任务。

• newSingleThreadExecutor（单线程线程池）
  有且仅有一个工作线程执行任务
  所有任务按照指定顺序执行，即遵循队列的FIFO规则

• newCachedThreadPool （可缓存线程池）
  这种线程池内部没有核心线程，线程的数量是有没限制的。
  在创建任务时，若有空闲的线程时则复用空闲的线程，若没有则新建线程。
  没有工作的线程（闲置状态）在超过了60s还不做事，就会销毁。

• newScheduledThreadPool（周期调度线程池）
  多线程进行定时或周期性的工作调度

• newSingleThreadScheduledExecutor
  newSingleThreadExecutor和newScheduledThreadPool的结合
  单线程进行定时或周期性调度

• newWorkStealingPool
  构建ForkJoinPool，利用Work-Stealing算法，并行地处理任务，不保证处理顺序。

• Executors 返回的线程池对象的弊端：

newFixedThreadPool 和 newSingleThreadExecutor

使用无界队列 LinkedBlockingQueue
允许请求队列的长度为Integer.MAX_VALUE，可能会堆积大量请求，导致OOM

newCachedThreadPool 和 newScheduledThreadPool

maximumPoolSize=Integer.MAX_VALUE
允许创建线程数量为 Integer.MAX_VALUE，可能会创建大量线程，导致OOM

• 推荐做法

	private Executor myExecutor =
		new ThreadPoolExecutor(5, 10,
			60L, TimeUnit.SECONDS,
			new LinkedBlockingQueue<>(200),
			new ThreadFactoryBuilder().setNameFormat("xxx-pool-%d").build(),
			new ThreadPoolExecutor.CallerRunsPolicy());

# 可以对 ThreadPoolExecutor 进行继承和封装，以提供更简便的线程池初始化方法
1
2
3
4
5
6
7
8