前言
在多线程编程中,线程池是一种非常重要的工具。它可以帮助我们高效地管理线程资源,避免频繁创建和销毁线程带来的性能开销。Java中提供了强大的线程池实现,如 ThreadPoolExecutor ,但有时我们可能希望了解其内部原理,并实现一个简单的线程池来加深对其工作机制的理解，手写线程池也是很多大厂常考的笔试题。
开个新坑-手搓简易线程池。本系列文章将从零开始,一步步手工编码实现一个简单但功能完备的线程池。我们将逐步介绍线程池的各个核心组件,并分析它们的作用和实现思路。通过这个过程,我们可以更好地掌握多线程编程的技巧,并培养编码能力和系统性思维。
本篇文章的任务是带领大家定义任务等待队列，任务等待队列是线程池中一个十分重要的组成部分，在各种生产者-消费者模型场景下十分常见，如下我们先来介绍一下等待队列。
等待队列是什么
等待队列(Task Queue)是一种用于临时存储任务的队列数据结构。在多线程环境下,它常被用作生产者-消费者模型中的"缓冲区",用于平衡任务的产生速度和任务的处理速度之间的差异。
等待队列本质上是一个先进先出(FIFO)的队列,新加入的任务会被存储在队列尾部,而消费者则从队列头部获取任务执行。根据队列的实现方式不同,它可以是无界的或者有界的。无界队列理论上可以存储无限多的任务,而有界队列则有最大容量限制。
我们这里的等待队列底层可以借助JDK提供的双端队列ArrayDeque实现。
为什么需要等待队列
在现实场景中,任务的产生速度往往是不均匀的,而执行任务的线程资源又是有限的。如果没有等待队列,当任务瞬间扩增时,可能会出现以下问题:
任务无法被及时消费,导致任务丢失或被阻塞。
需要临时创建大量线程来处理任务,线程的创建和销毁开销巨大,影响系统性能。
任务的执行顺序无法得到保证,可能导致某些重要任务长期得不到执行。
引入等待队列后,生产者可以先将任务存入队列,而消费者则持续从队列中获取并执行任务。这种"缓冲"机制可以有效应对任务瞬间扩增的情况,并保证任务按先后顺序被逐个消费。
此外,通过设置有界队列,我们还可以限制队列的最大容量,防止任务无限堆积导致内存溢出。当队列已满时,我们可以采取拒绝策略(如直接丢弃、暂存等)来应对新加入的任务。如下图，任务等待队列其实就是联系任务生成者与任务消费者的一个桥梁，生产者生产消息放入等待队列中，再由消费者拿取消费。
实现思路
为了实现一个功能完备的任务等待队列,我们需要设计一个阻塞队列BlockQueue,它具有以下几个关键特性:
有界队列 BlockQueue将设置一个固定的容量size,队列中最多只能存储size个任务。这样可以防止任务无限制地堆积,导致内存溢出。当队列满时,新添加的任务将被阻塞,直到队列中有空位。
线程安全 BlockQueue的操作,包括添加任务put和获取任务take,都需要保证线程安全。我们将使用Java的重入锁ReentrantLock和条件变量Condition来实现线程的正确同步。
支持超时 在添加或获取任务时,BlockQueue将提供带超时时间的方法offer和poll。这样可以防止任务被无限期阻塞,提高系统的响应性和健壮性。
基于以上思路,BlockQueue的实现将涉及以下几个核心部分:
使用双端队列Deque作为底层数据结构存储任务
使用ReentrantLock和两个Condition(emptyCondition和fullCondition)来实现阻塞和唤醒机制
实现put、offer、take、poll等核心方法,正确控制任务的添加和获取
代码实现
1.新建BlockQueue类
我们这里新建一个名为BlockQueue的类，并声明一些属性与字段，其中ArrayDeque为一个双端队列，可以在队列的两端操作元素，size为定义的队列大小
public class BlockQueue<T> { // 双端队列 private Deque<T> deque = new ArrayDeque<>(); // 队列的容量 private int size; public BlockQueue(int size) { this.size = size; } }
2.任务的添加和获取方法
接下来我们编写向队列中添加任务与获取任务的方法，但是这里要注意的是，读写操作都是在线程池多线程的环境下进行的，存在线程安全问题，所以我们需要配合锁机制来保证操作的原子性。我们这里用JDK提供的ReentrantLock可重入锁来解决这个问题。
继续新增ReentrantLock字段：
private ReentrantLock lock = new ReentrantLock();
此外，我们还需要解决任务添加与获取时的等待操作，也就是当队列为空时，消费者需要等待任务产生，当队列满时，生产者需要等待队列中有空位才能存入任务。如下图，当消费者获取到锁时，会尝试获取任务，但发现队列为空，就会阻塞等待。
当生产者生产任务后，肯定不能让消费者干等着，而是去通知消费者有活做啦~如下图：
我们可以看到这里涉及到了生产者与消费者不同线程的通讯，这里我们可以借助Condition来完成消费者线程与生产者线程之间的通讯。
分别定义两个Condition代表队列满情况的等待室与队列空情况的等待室
// 队列空情况的休息室 Condition emptyCondition = lock.newCondition(); // 队列满情况的休息室 Condition fullCondition = lock.newCondition();
定义阻塞添加任务方法put
// 添加任务 阻塞添加 public void put(T task) { lock.lock(); try { while (size == deque.size()) { try { fullCondition.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } } log.debug("task 添加成功 ，{}", task); deque.addLast(task); emptyCondition.signal(); } finally { lock.unlock(); } }
这里添加任务会存在两张情况：
- 队列满了：调用fullCondition.await()方法挂起当前生产者线程，也就是让当前生产者线程等待。
- 队列没满：将任务加入队列中，并调用emptyCondition.signal()通知挂起的消费者。
编写堵塞拿取任务方法take
// 获取任务 public T take() { lock.lock(); try { while (deque.isEmpty()) { try { emptyCondition.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } } T t = deque.removeFirst(); fullCondition.signal(); log.debug("获取了任务 {}", t); return t; } finally { lock.unlock(); } }
同理，这里拿取任务也会存在两张情况：
- 队列空的：调用emptyCondition.await()方法挂起当前消费者线程，也就是让当前消费者线程等待。
- 队列存在元素：任务出栈，并返回出栈的任务元素，然后调用fullCondition.signal()通知挂起的生产者。
以上我们其实我们就完成了一个简单的任务堵塞队列，但是我们会发现，如果这两个方法都是会一直堵塞，显然是不合理的，所以我们这里新增添加和获取任务的超时方法。
带超时时间的阻塞添加方法offer
// 带超时时间阻塞添加 public boolean offer(T task, long timeout, TimeUnit timeUnit) { lock.lock(); try { long nanos = timeUnit.toNanos(timeout); while (deque.size() == size) { try { if (nanos <= 0) { return false; } log.debug("等待加入任务队列 {} ...", task); nanos = fullCondition.awaitNanos(nanos); } catch (InterruptedException e) { e.printStackTrace(); } } log.debug("加入任务队列 {}", task); deque.addLast(task); emptyCondition.signal(); return true; } finally { lock.unlock(); } }
我们可以利用fullCondition.awaitNanos来实现超时等待，当超过给定参数时间时，就会被自动唤醒，并且将传入时间-等待时间作为返回值，下一次循环nanos <= 0时，就会判定为本次操作超时失败了
带超时时间的阻塞获取方法poll
// 带超时时间阻塞获取 public T poll(long timeout, TimeUnit unit) { // 1.上锁 lock.lock(); try { long nanos = unit.toNanos(timeout); // 转为毫秒 // 2.首先检查队列是否存在元素 while(deque.isEmpty()){ try { // 2.1超时判断，返回值是剩余时间 if(nanos <= 0){ return null; } // 2.2超时等待 log.debug("等待获取任务"); nanos = emptyCondition.awaitNanos(nanos); } catch (InterruptedException e) { e.printStackTrace(); } } // 3.拿取元素 T task = deque.removeFirst(); log.info("任务拿取成功:{}", task); // 4.唤醒挂起的生产者 fullCondition.signal(); return task; } finally { // 释放锁 lock.unlock(); } }
总结
我们在本节内容中实现线程池中一个重要的组件 - 任务等待队列BlockQueue，编写的对应的代码也已同步到了github仓库中(本章节可参考liushijie-240328-factory分支)，估计还要两次文章的内容才能完成手写线程池部分的内容，后面主要要完成的就是线程池内部运行的基本逻辑，自定义拒绝策略等内容了，博主也是边学边实践边输出教学文章，如果有什么问题都可以在评论区留言，觉得写得不错的话就多多支持吧~
