Java 线程池框架核心代码分析

前言

多线程编程中,为每个任务分配一个线程是不现实的,线程创建的开销和资源消耗都是很高的。线程池应运而生,成为我们管理线程的利器。Java 通过 Executor 接口,提供了一种标准的方法将任务的提交过程和执行过程解耦开来,并用 Runnable 表示任务。

下面,我们来分析一下 Java 线程池框架的实现 ThreadPoolExecutor

下面的分析基于JDK1.7

生命周期

ThreadPoolExecutor 中,使用 CAPACITY 的高3位来表示运行状态,分别是:

terminated()
terminated()

Java 线程池框架核心代码分析

ThreadPoolExecutor 中用原子类来表示状态位

private final AtomicInteger ctl = new AtomicInteger(ctlOf(RUNNING, 0));

线程池模型

核心参数

  • corePoolSize :最小存活的工作线程数量(如果设置 allowCoreThreadTimeOut ,那么该值为 0)
  • maximumPoolSize :最大的线程数量,受限于 CAPACITY
  • keepAliveTime :对应线程的存活时间,时间单位由TimeUnit指定
  • workQueue :工作队列,存储待执行的任务
  • RejectExecutionHandler拒绝策略,线程池满后会触发

线程池的最大容量: CAPACITY 中的前三位用作标志位,也就是说工作线程的最大容量为 (2^29)-1

四种模型

CachedThreadPool
FixedThreadPool
SingleThreadPool
ScheduledThreadPool

执行任务 execute

核心逻辑:

  1. 当前线程数量 < corePoolSize ,直接开启新的核心线程执行任务 addWorker(command, true)
  2. 当前线程数量 >= corePoolSize ,且任务加入工作队列成功

    RUNNING
    
  3. 开启普通线程执行任务 addWorker(command, false) ,开启失败就拒绝该任务

从上面的分析可以总结出线程池运行的四个阶段:

  1. poolSize < corePoolSize 且队列为空,此时会新建线程来处理提交的任务
  2. poolSize == corePoolSize ,此时提交的任务进入工作队列,工作线程从队列中获取任务执行,此时队列不为空且未满。
  3. poolSize == corePoolSize ,并且队列已满,此时也会新建线程来处理提交的任务,但是 poolSize < maxPoolSize
  4. poolSize == maxPoolSize ,并且队列已满,此时会触发拒绝策略

拒绝策略

前面我们提到任务无法执行会被拒绝, RejectedExecutionHandler 是处理被拒绝任务的接口。下面是四种拒绝策略。

AbortPolicy
CallerRunsPolicy
DiscardPolicy
DiscardOldersPolicy

线程池中的 Worker

Worker 继承了 AbstractQueuedSynchronizerRunnable ,前者给 Worker 提供的功能,后者执行工作线程的主要方法 runWorker(Worker w) (从任务队列捞任务执行)。Worker 引用存在 workers 集合里面,用 mainLock 守护。

private final ReentrantLock mainLock = new ReentrantLock();
private final HashSet<Worker> workers = new HashSet<Worker>();

核心函数 runWorker

下面是简化的逻辑,注意:每个工作线程的 run 都执行下面的函数

final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    while (task != null || (task = getTask()) != null) {
        w.lock();
        beforeExecute(wt, task);
        task.run();
        afterExecute(task, thrown);
        w.unlock();
    }
    processWorkerExit(w, completedAbruptly);
}
  1. getTask() 中获取任务
  2. 锁住 worker
  3. 执行 beforeExecute(wt, task) ,这是 ThreadPoolExecutor 提供给子类的扩展方法
  4. 运行任务,如果该worker有配置了首次任务,则先执行首次任务且只执行一次。
  5. 执行 afterExecute(task, thrown);
  6. 解锁 worker
  7. 如果获取到的任务为 null,关闭 worker

获取任务 getTask

线程池内部的任务队列是一个阻塞队列,具体实现在构造时传入。

private final BlockingQueue<Runnable> workQueue;

getTask() 从任务队列中获取任务,支持阻塞和超时等待任务,四种情况会导致返回 null ,让 worker 关闭。

STOP
SHUTDOWN

核心逻辑:根据 timed 在阻塞队列上超时等待或者阻塞等待任务,等待任务超时会导致工作线程被关闭。

timed = allowCoreThreadTimeOut || wc > corePoolSize;
Runnable r = timed ?
    workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
    workQueue.take();

在以下两种情况下等待任务会超时:

allowCoreThreadTimeOut(true)
wc > corePoolSize

工作队列使用的是 BlockingQueue ,这里就不展开了,后面再写一篇详细的分析。

总结

  • ThreadPoolExecutor 基于生产者-消费者模式,提交任务的操作相当于生产者,执行任务的线程相当于消费者。
  • Executors 提供了四种基于 ThreadPoolExecutor 构造线程池模型的方法,除此之外,我们还可以直接继承 ThreadPoolExecutor ,重写 beforeExecuteafterExecute 方法来定制线程池任务执行过程。
  • 使用有界队列还是无界队列需要根据具体情况考虑,工作队列的大小和线程的数量也是需要好好考虑的。
  • 拒绝策略推荐使用 CallerRunsPolicy ,该策略不会抛弃任务,也不会抛出异常,而是将任务回退到调用者线程中执行。

原文 

https://blog.xiaohansong.com/java-executor.html

本站部分文章源于互联网,本着传播知识、有益学习和研究的目的进行的转载,为网友免费提供。如有著作权人或出版方提出异议,本站将立即删除。如果您对文章转载有任何疑问请告之我们,以便我们及时纠正。

PS:推荐一个微信公众号: askHarries 或者qq群:474807195,里面会分享一些资深架构师录制的视频录像:有Spring,MyBatis,Netty源码分析,高并发、高性能、分布式、微服务架构的原理,JVM性能优化这些成为架构师必备的知识体系。还能领取免费的学习资源,目前受益良多

转载请注明原文出处:Harries Blog™ » Java 线程池框架核心代码分析

赞 (0)
分享到:更多 ()

评论 0

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址