★ FEATURED ARTICLE 先说结论Java 里一个线程从出生到销毁拢共就六个状态而阻塞队列的实现核心就是 wait/notify 这两个最朴素的等待唤醒机制在撑场面。很多人用 ThreadPoolExecutor 用得飞起但一问到线程什么时候处于 WAITING、什么时候处于 BLOCKED立马含糊。我带过好几个新人十个里八个卡在这一关。这篇就把线程生命周期和基于 Object 的 wait/notify 阻塞队列一次讲透最后再聊聊线程池里的队列到底该怎么选、为什么这么选。这里的 Object就是 Java 的 java.lang.Object别想歪了。1. 线程生命周期六个状态背后的真实含义六个状态分别是NEW、RUNNABLE、BLOCKED、WAITING、TIMED_WAITING、TERMINATED。死记硬背没用你得把它理解成线程当前在干嘛的描述。这六个状态按行为可以归成四类还没开始跑、能抢 CPU、被挡住、已经结束。1.1 NEW、RUNNABLE、TERMINATED相对简单的三个先说最简单的两个。NEW是你new Thread()之后、还没调用start()之前的状态线程对象已经存在但底层操作系统线程还没创建啥也不干。TERMINATED则是run()方法正常返回或者抛异常结束之后的状态线程已经凉透了不能重启重启会直接抛IllegalThreadStateException。RUNNABLE这个最容易误解。很多人以为 RUNNABLE 就是正在运行错。Java 把就绪和运行中合并成了 RUNNABLE 一个状态线程拿到 CPU 时间片在跑是 RUNNABLE线程在等待队列里排队等 CPU 同样是 RUNNABLE。你去看 Linux 下 top 命令一个进程里可能有一堆 RUNNABLE 状态的线程但 CPU 只有那么几个核不可能所有线程同时在跑。RUNNABLE 只是说这线程没在睡觉也没被锁挡着随时可以被调度器选中执行。所以Thread.yield()这个操作其实就是从运行中退回就绪状态还是 RUNNABLE只是主动让出 CPU。想直观验证写几行代码就能看到状态Thread t new Thread(() - { try { Thread.sleep(2000); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }); System.out.println(t.getState()); // NEW t.start(); Thread.sleep(100); System.out.println(t.getState()); // RUNNABLE就绪或运行中 Thread.sleep(100); System.out.println(t.getState()); // 大概率 TIMED_WAITING t.join(); System.out.println(t.getState()); // TERMINATED这里有个细节值得留意t.getState()是让你观察别人家的线程不是查自己的状态。你没法在一个线程内部准确拿到我正在跑这种信息因为此刻你反手一查自己当然是 RUNNABLE。1.2 BLOCKED 和 WAITING重点中的重点BLOCKED和WAITING日常最容易混。区别一句话就能说清BLOCKED 是进不去 synchronized 的门口WAITING 是已经进到门里之后主动挂起、等一个信号。BLOCKED 只出现在抢synchronized内置锁的场景。比如锁被另外的线程持有你这边执行到synchronized(lock)进不去线程就停在 BLOCKED。注意ReentrantLock的lock()抢不到锁时线程处于 WAITING而不是 BLOCKED。这是一个特别容易踩的面试坑——JUC 的锁走的是LockSupport.park()走的是 WAITING跟 synchronized 的 BLOCKED 是两套机制。WAITING 则是有明确触发条件的Object.wait()无超时版本、Thread.join()无超时版本、LockSupport.park()。这三个动作的共同点是线程主动表示我不跑了等有人叫我。进入 WAITING 后不消耗 CPU也不参与锁竞争就死等一个唤醒信号。还有个细节很多人不知道wait()被唤醒、重新去抢锁的时候如果锁还被别人占着线程会从 WAITING 转入 BLOCKED而不是直接回到 RUNNABLE。wait 唤醒只是拿到了竞争锁的资格还没拿到锁本身。这个细节和你后面自己写阻塞队列的体验强相关因为 put/take 操作里 wait 和 synchronized 是嵌套在一起的。TIMED_WAITING可以理解为带闹钟的 WAITINGThread.sleep()、wait(timeout)、join(timeout)、parkNanos()都是这个状态。到时间自动醒不需要外部通知。2. 为什么值得把阻塞队列手写一遍JDK 里现成的阻塞队列都有现成的LinkedBlockingQueue、ArrayBlockingQueue、SynchronousQueue随便用一个不就完了吗自己手写不是脱裤子放屁我年轻的时候也这么想直到面试被问傻、线上排查死锁抓瞎才明白手写一遍的价值在哪。2.1 手写的价值面试、排查、读源码三合一面试必考题就是请手写一个生产者消费者模型或者实现一个有界阻塞队列。你要是只会new ArrayBlockingQueue(10)这道题基本白给。但手写一遍之后你会真的理解两件事第一wait()被调用后会释放锁这句话很多人背得滚瓜烂熟但只有自己写了队列、压测了多线程程序才知道释放锁和等待之间那个原子性有多重要第二为什么队列空和队列满必须用两个不同的等待条件以及 JUC 源码里那些 lock 和 condition 到底在替代什么。另一个用得上手写知识的场景是线上排查。你拿 top -H 看到一堆线程堆在 WAITING得能分辨它们卡在哪个 condition 上。如果线程是被 JUC 的Condition.await()挂起的线程栈里能看到park字样如果是Object.wait()栈里会显示对应的 wait 方法。没有手写过 while 循环等待你根本不会去想为什么 wait 要放在循环里这个问题也就不会警觉线上可能出现虚假唤醒导致的脏读。2.2 Object 的 wait/notify 是怎么工作的synchronized加在对象上本质是拿这个对象当锁锁的状态其实存在对象头里。HotSpot 的 mark word 里就记录了锁相关信息锁竞争升级后对象头会指向一个 ObjectMonitor 对象这个 monitor 内部维护两个关键集合一个用来记录当前持有锁的线程一个用来挂起调用了wait()的线程对应 WAITING 状态。wait()的语义是调用线程必须已经持有该对象的锁调用后原子地做三件事——把自己加入该对象的等待集合、释放持有的锁、挂起。为什么强调原子因为如果释放锁和挂起不是一步完成的中间被其他线程插进来就可能出现通知丢失你还没挂起来别人已经把数据放进去并调了 notify等你去等的时候已经没有信号了。这一块在后面第 4 章会专门展开。notify()随机唤醒一个正在该对象上 wait 的线程notifyAll()唤醒全部。被唤醒的线程不会立刻执行它要先重新竞争对象锁抢到锁才能从wait()返回。这也是为什么wait()之后的代码不能假设条件已满足必须重新检查。3. 基于 Object 的阻塞队列实现保姆级拆解现在动真格的。我用java.lang.Object的 wait/notify 实现一个容量固定的有界阻塞队列支持并发 put 和 take满时 put 阻塞空时 take 阻塞。所有代码可以直接复制跑。3.1 数据结构设计环形数组怎么选队列底层我用Object[]数组存储元素配合head、tail、count三个指针做成环形数组。为什么用环形因为普通数组队列在重复入队出队后队头指针往前移动数组头部空间就废弃了需要频繁搬移数据。环形数组让 head 和 tail 都在数组范围内循环移动模拟一个逻辑上无限长的队列空间复用率最高。这是 JDKArrayBlockingQueue的做法。容量用capacity固定构造时传入。索引推进统一用(index 1) % capacity取模回绕。count 记录当前元素数量它同时是空和满的判断依据count 0空count capacity满。用一个共享的锁对象lock保护所有变量和判断逻辑。public class ObjectBlockingQueueE { private final Object[] items; private final int capacity; private int head; private int tail; private int count; private final Object lock new Object(); public ObjectBlockingQueue(int capacity) { if (capacity 0) { throw new IllegalArgumentException(容量必须为正数); } this.capacity capacity; this.items new Object[capacity]; } public void put(E e) throws InterruptedException { if (e null) { throw new NullPointerException(); } synchronized (lock) { while (count capacity) { lock.wait(); } items[tail] e; tail (tail 1) % capacity; count; lock.notifyAll(); } } SuppressWarnings(unchecked) public E take() throws InterruptedException { synchronized (lock) { while (count 0) { lock.wait(); } E e (E) items[head]; items[head] null; head (head 1) % capacity; count--; lock.notifyAll(); return e; } } public int size() { synchronized (lock) { return count; } } }为什么用 lock 对象而不是直接用 synchronized 加在 this 上主要是可读性和封装性考虑。队列对外暴露的 API 是 put、take、size但对象锁 this 是公开可访问的外部代码如果也对这个实例加锁会和你内部 put/take 的锁形成同一个竞争入口增加死锁风险。用一个私有的 lock 对象相当于把锁的访问范围限定在类内部这点和 JDK 源码的 ReentrantLock 是同一个思路。3.2 put 的核心逻辑入队、通知、释放put 的逻辑分四步。第一步校验非空阻塞队列不允许存 null这是 JDK 的约定因为 null 在 take 端还被用来做元素清空标记混入 null 会制造混乱。第二步进入 synchronized 临界区检查容量如果满了lock.wait()让当前线程挂起并释放锁。这里必须用 while 而不是 if原因放在第 4 章细说先记住wait 永远活在 while 里面。第三步入队往 tail 位置放元素tail 前移count 自增。第四步notifyAll()唤醒所有正在这个锁上等待的线程然后在方法返回时释放锁。这里有一个关键点notifyAll()必须放在锁内、在修改完共享状态之后调用。因为唤醒的消费者线程从 wait 返回后会重新抢锁抢到后要检查 count 是不是真的大于 0 了。如果通知发生在状态修改之前消费者醒来读到的还是旧值就会造成信号比数据先到的脏读。3.3 take 的核心逻辑出队、置空、避免内存泄漏take 流程和 put 正好对称。第一步检查空count 0就 wait。第二步从 head 位置取元素。第三步特别容易漏items[head] null。这一步必须做否则数组里那个位置的引用还指向旧对象这个对象本来已经被消费者拿走业务上用完了但队列数组还攥着它的引用垃圾收集器永远回收不掉等于人为制造内存泄漏。数组和 ArrayList 扩容缩容时底层数组也常留着这种过期引用这是老手和新手的一个典型区分点。第四步让 head 前移、count 自减、notifyAll()。注意这里唤醒的是谁——唤醒的是正在等待队列有空位、准备 put 的生产者线程。因为只有 take 会让队列变空出一格。同理put 里唤醒的是等数据、准备 take 的消费者线程。虽然notifyAll()是全部唤醒但被唤醒的线程醒来后都要重新检查条件不该干活的比如生产者醒来发现队列还是满的会再次wait()所以功能上是安全的只是多几次无谓的锁竞争这就是前面说的为了正确性牺牲少许性能。3.4 写一个多生产者多消费者测试验证单靠看代码心里没底直接上多线程压一下import java.util.concurrent.TimeUnit; public class QueueDemo { public static void main(String[] args) throws Exception { ObjectBlockingQueueInteger queue new ObjectBlockingQueue(3); Thread producer new Thread(() - { try { for (int i 1; i 10; i) { queue.put(i); System.out.println(Thread.currentThread().getName() 放入 i); TimeUnit.MILLISECONDS.sleep(300); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, Producer); Thread consumer new Thread(() - { try { for (int i 0; i 10; i) { Integer value queue.take(); System.out.println(Thread.currentThread().getName() 取出 value); TimeUnit.MILLISECONDS.sleep(500); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } }, Consumer); producer.start(); consumer.start(); producer.join(); consumer.join(); System.out.println(全部完成); } }你跑起来会看到生产者产得快300ms 一个消费者消得慢500ms 一个很快队列就被塞满生产者在第 4 次 put 之后开始等待消费者每取走一个生产者才会被唤醒放入下一个。整个过程队列容量始终不超过 3这就是阻塞的作用——生产速度被消费速度反向压制而不是无脑往内存里灌。想测试多生产者多消费者再开两个线程塞 100 个任务最后断言 count 归零即可。我在实测中发现多线程场景下notifyAll换成notify会出现偶发的线程永远挂起也就是第 4 章要讲的丢失唤醒。3.5 和 JDK ArrayBlockingQueue 对比我们缺了什么自己写的版本和ArrayBlockingQueue对比功能上基本一致但有几个缺失点值得知道。第一JDK 用的是ReentrantLock加两个ConditionnotEmpty 和 notFull把队列满和队列空分成两个独立的等待队列这样 put 时只需要唤醒 notEmpty 上的消费者take 时只唤醒 notFull 上的生产者不会像我们这样无差别唤醒全部。第二JDK 的lockInterruptibly()可以在等待锁期间响应中断我们的 synchronized 无法响应中断所以在 put/take 之外加了InterruptedException抛出等锁期间想中断是做不到的。第三JDK 用itrs维护迭代器一致性我们完全不涉及迭代。这些差距不说明手写白费。恰恰相反正因为你亲手写出了这几个条件组合再回去看 JDK 源码看到lock.lockInterruptibly()、while (count items.length) notFull.await()这些行才会觉得每一句都眼熟而不是天书。4. 手写阻塞队列最容易踩的四个坑我把这些年线上线下见过的坑整理成四类每一个都是我或者身边同事真真切切踩过的属于不做不知道做了才后怕的类型。4.1 条件判断用 if 还是 while这是个生死问题网上能找到大量教程把条件判断写成synchronized (lock) { if (count 0) { lock.wait(); } // 取元素 }单生产者单消费者这样偶尔也能跑通于是很多人就一直这么交作业。一旦换成多生产者多消费者必出 bug。原因有两个。第一个是 wait 的语义要求线程从wait()返回的那一刻它只是重新获得了锁并不代表条件已经满足。比如两个消费者同时等空队列生产者放入一个元素并notifyAll()两个消费者都醒了一个抢到锁把唯一的元素取走另一个抢到锁后发现 count 已经是 0——如果它用的是 if就会直接越过取元素逻辑读到 head 位置的 null轻则空指针重则把脏数据发出去。必须用 while醒来后再检查一次不满足继续等。第二个是虚假唤醒。Java 官方文档明确说过wait 可能在没有被 notify、没有中断、没有超时的情况下自己醒来这是底层实现的允许行为在 Linux 的 futex 等机制下并不常见但规范层面就是允许的。如果你用 if一次虚假唤醒就能让你的线程在条件不成立的情况下继续往下执行。用 while 天然免疫这个问题因为每次醒来都会重新验证。4.2 notify 和 notifyAll 选错信号可能永远丢失前面测试代码里我用的是notifyAll。为什么不用notify单消费者场景下 notify 看起来没什么问题——队列里只有一个线程在等唤醒它恰好。但多线程场景notify 是随机唤醒一个正在等待的线程如果被挑中的那个线程和自己的业务方向不一致就会出事。举一个具体例子队列已满一个生产者正在 wait一个消费者正在 wait没满但空了——等等实际不可能两个 wait 同时发生得理顺场景。更经典的是空队列加一个满队列假设队列空两个消费者在等同时队列还有一个生产者刚被阻塞不对队列空的时候生产者不会阻塞。我们换一个简单直接的场景队列容量为 1已满一个生产者 P1 在 wait同时队列空两个消费者 C1、C2 在 wait——但一个队列不可能既满又空。好实际能同时 wait 的只有两类一类等非空消费者在空队列上等一类等非满生产者在满队列上等。如果队列非满非空谁都不用等。关键场景如下队列已满生产者 P1 在等空位。此时消费者 C1 调用 take取走一个元素然后notify()。如果 wait 集合里只有 P1P1 被唤醒一切正常。但如果 wait 集合里同时还有另一个消费者 C2它之前因为某种原因在等而 notify 偏偏选中了 C2 而不是 P1C2 醒来发现队列没满它是消费者它等的是非空条件其实不满足于是继续 wait。P1 没人通知永远挂起。这就是「唤醒错对象导致的信号丢失」。notifyAll把所有线程都拉起来让它们自己判断谁该干活谁继续睡虽然笨但正确。4.3 中断异常不能一 catch 了之put 和 take 抛InterruptedException测试代码里我 catch 之后调用了Thread.currentThread().interrupt()恢复中断标志。很多人不理解为什么要多此一举。中断标志是线程的一个属性interrupt()设置它Thread.sleep()、wait()这类方法在检测到中断标志后会先清除标志再抛异常。也就是说异常抛出来的时候中断标志已经被清掉了。如果你 catch 住什么都不做外层代码继续用isInterrupted()去判断会发现这个线程好像从来没被中断过中断信号就这样被吞掉了。正确的习惯是如果你不打算立刻响应中断比如当前正在自己实现的框架里做清理工作catch 后必须Thread.currentThread().interrupt()把标志补回去让上层调用方看到。如果当前方法语义就是被中断就退出那可以不上抛也不补直接处理完 return但必须在注释里说明清楚。这个习惯属于并发编程的基本素养规则和小费文化差不多——可以不给但默认该给。4.4 共享变量必须进锁否则可见性没保证我在第 3 章的实现里head、tail、count、items 的所有读写全部在 synchronized(lock) 内部。这不仅仅是互斥还解决了内存可见性synchronized建立 happens-before 关系线程退出同步块时对共享变量的写对于后续进入同一个锁的线程是可见的。假设有人偷懒take 方法里为了减少锁竞争把 count 的读取放到 synchronized 外面做预检查if (count 0) { // 直接返回或做别的不进锁 }这在单线程下没问题多线程下 count 可能是过期的脏值。你可能读到 count0 就返回队列为空但实际上另一个生产者刚 put 了元素count 已经是 1。这就是为什么 JUC 里ConcurrentLinkedQueue这类无锁队列要用 volatile 加 CAS你手写 wait/notify 队列时根本没有 CAS所有共享状态的读写只能统一交给一把锁来管。任何逃出锁的读写都在破坏这个模型。5. 从手写队列到线程池选型队列决定线程池的脾气把队列搞明白之后再回头看线程池会发现ThreadPoolExecutor的很多行为不过是队列满了/空了这件事的延伸。这一节把线程池的阻塞队列选型讲透这也是面试里排队出现的追问点。5.1 ThreadPoolExecutor 里队列扮演什么角色ThreadPoolExecutor的构造参数里workQueue就是阻塞队列。线程池的工作流程是这样的提交一个任务如果核心线程数没满直接开新线程执行核心线程满了任务进队列排队队列也满了再尝试扩到最大线程数最大线程数也满了就触发拒绝策略。队列在这里是核心线程和最大线程之间的缓冲层它的大小直接决定了线程池从排队到扩线程的切换点。假如队列是无界的比如默认构造用的LinkedBlockingQueue不设容量那么任务永远进得去队列队列满这个条件永远不会发生线程数也就永远不会超出核心线程数。这带来的一个副作用是突然爆发的十万个任务全部堆在内存里核心线程按自己的速度慢慢消化看起来线程池稳如老狗实际上内存已经涨到吓人。FixedThreadPool 就是这么设计的适合任务量平稳的场景但你在代码里如果敢用默认构造的无界队列接业务流量迟早要吃一次 OOM 的教训。5.2 四种队列对应四种脾气队列是否有界特点典型搭档LinkedBlockingQueue默认无界可指定容量链表结构吞吐量高创建时可传容量FixedThreadPoolArrayBlockingQueue必须有界数组结构可指定公平性锁竞争时支持公平模式自定义业务线程池SynchronousQueue无容量相当于 0不存任务put 必须等 take直接交接CachedThreadPoolPriorityBlockingQueue无界按优先级而不是 FIFO 出队任务带优先级排序的场景SynchronousQueue是最有意思的一个它内部不存任何元素生产者 put 必须等待消费者 take反之亦然。这个直接交接的语义恰恰就是我们第 3 章手写队列在容量为 0 时的极限情况。CachedThreadPool用它配合maximumPoolSize为整型的最大值实现的效果是来一个任务就开一个线程线程空闲 60 秒后回收。因为队列永远装不下任务所以永远不会排队所有任务都在寻求直接执行。代价是如果高并发任务量很大线程数会无限膨胀。我之前在服务里压测过CachedThreadPool 在持续峰值下能开出几百上千个线程光是线程栈内存就能吃掉几百兆。PriorityBlockingQueue优先级队列出队顺序按任务的compareTo决定适合有优先级诉求的场景。但它有个问题要留意无界意味着内存风险一旦生产速度超过消费速度任务堆积同样没有上限。5.3 我给的真实选型建议性能敏感且任务量可控的接口优先考虑自定义线程池搭配有界ArrayBlockingQueue。队列容量怎么定我一般按「期望的排队任务上限」来设比如接口平均耗时 100ms核心线程 8 个希望一个任务最多等 500ms那么排队容量大概就是8 * 5 40这个量级再加一点余量定 64。这不是精确公式但比拍脑袋强。LinkedBlockingQueue带容量构造相比ArrayBlockingQueue在有界场景下往往有更好的吞吐因为链表入队出队操作更分散锁竞争概率更低。JDK 里Executors.newFixedThreadPool默认用的是无界LinkedBlockingQueueExecutors.newCachedThreadPool用的是SynchronousQueue这两个都是拿出来即用的典型但Executors工具类整体上不建议在正式项目里直接用因为它给不了你拒绝策略和队列容量的控制权。如果任务允许丢弃优先用SynchronousQueue配合CallerRunsPolicy或者AbortPolicy效果好且不会积压。我踩过最惨的一个坑就是给一个消息推送服务用了无界队列后端消费变慢后任务肉眼可见地堆到了上千万最后 OOM 直接把进程打挂。自那以后我给自己立了一个规矩凡是自定义线程池队列容量必须显式声明最大线程数和拒绝策略必须显式声明省得后人包括三个月后的我自己看着默认参数猜半天。最后再分享一个小技巧线程生命周期那个状态机平时写代码可能用不上但排查问题的时候就是武器。我处理过一次诡异的线上故障看线程 dump 发现几十个线程都卡在同一个ObjectBlockingQueue.take的 wait 上顺着状态一查就是队列空了、生产者挂了没人唤醒。那一刻我终于明白当初老老实实手写一遍阻塞队列真没白费。
阅读完成 · 觉得有帮助？
