前面已经讨论了Condition如何让线程在业务条件不满足时释放锁并等待。BlockingQueue正是这类机制在队列场景中的标准化封装队列为空时消费者不需要自己写await()队列已满时生产者也不需要自己管理notFull。它把“存取元素”和“空满等待”合并成了一套统一的队列接口。普通Queue只回答一个问题元素如何进入队列、如何从队列中取出。BlockingQueue额外回答了另一个问题当队列暂时不能继续操作时线程应该怎么办。这个问题在生产者消费者模型中非常常见生产者负责放入任务消费者负责取走任务如果两边速度不一致队列就会出现空或满。因此BlockingQueue不是简单的“线程安全队列”而是一个带等待语义的线程安全队列。线程安全保证多个线程同时访问队列时内部状态不会损坏等待语义保证队列空或满时线程可以按约定阻塞、超时、失败或抛异常。一、BlockingQueue 的方法差异本质是失败策略差异队列操作主要分成插入、获取并移除、只查看三类。BlockingQueue在普通Queue的基础上对“操作无法立即完成”提供了几组不同策略。失败策略插入元素获取并移除元素查看队首元素队列满或空时的行为抛异常add(e)remove()element()无法操作时抛异常返回特殊值offer(e)poll()peek()插入失败返回false获取失败返回null一直等待put(e)take()无满了不能放就等空了不能取就等超时等待offer(e, time, unit)poll(time, unit)无等一段时间超时后返回失败结果这张表的重点不在于记 API 名称而在于区分调用者想要的语义。put(e)表示“这个元素必须放进去满了我可以等”take()表示“我必须取到元素空了我可以等”。而offer(e)和poll()更像是“试一下”不能立即完成就返回结果不会长期阻塞当前线程。在生产者消费者模型中最典型的是put()和take()queue.put(task);Tasktaskqueue.take();如果队列有容量上限put()可能因为队列满而等待如果队列没有元素take()可能因为队列空而等待。至于等待时如何释放锁、如何进入条件队列、被通知后为什么要重新竞争锁这些属于上一章Condition已经完成的机制闭环本文后面只在队列实现中简短回指。二、ArrayBlockingQueue固定数组上的有界阻塞队列ArrayBlockingQueue是最接近手写有界阻塞队列的实现。它的内部结构可以概括为固定数组、一把锁、两个条件队列。字段作用items真正保存元素的数组putIndex下一次插入元素的位置takeIndex下一次取出元素的位置count当前队列中的元素数量lock保护数组、下标和数量的一把锁notEmpty队列为空时消费者等待的条件notFull队列满时生产者等待的条件假设容量为 5依次放入A、B、C后数组状态大致如下此时执行一次take()取出的不是移动整个数组而是读取takeIndex指向的位置然后让takeIndex后移。数组不会把B、C搬到前面因为每次搬移都会带来额外成本。队列只需要维护“下次从哪里取”和“下次往哪里放”。数组下标走到末尾后会回到 0所以这个数组是循环使用的。继续放入D、E再放入F时putIndex会回到开头把已经被取走的位置重新利用。这里可以看到一个细节putIndex takeIndex并不一定表示队列为空也可能表示队列已满。因此ArrayBlockingQueue不能只靠两个下标判断空满而要用count区分。状态判断条件队列为空count 0队列已满count items.length在这个结构上put()的流程就是先获得锁如果count items.length说明队列已满进入notFull等待否则把元素写入items[putIndex]移动putIndex增加count最后通知notEmpty上的消费者。对应地take()会在count 0时等待notEmpty否则读取items[takeIndex]清空该位置引用移动takeIndex减少count最后通知notFull上的生产者。简化代码可以这样理解// put 侧写入数组尾部items[putIndex]e;if(putIndexitems.length){putIndex0;}count;notEmpty.signal();// take 侧读取数组头部Eeitems[takeIndex];items[takeIndex]null;if(takeIndexitems.length){takeIndex0;}count--;notFull.signal();returne;这里的items[takeIndex] null不是为了让本次返回正确因为返回值已经保存到局部变量e中它的作用是断开数组对已出队对象的引用避免对象逻辑上已经离开队列却仍然因为数组持有引用而不能被 GC 回收。三、ArrayBlockingQueue 为什么只用一把锁ArrayBlockingQueue的所有核心状态都由同一把ReentrantLock保护。生产者修改items、putIndex和count消费者修改items、takeIndex和count其中count是双方共同修改的状态。使用一把锁可以让所有修改串行化状态一致性更容易保证。这也带来明显代价即使队列既不空也不满一个生产者正在put()一个消费者也不能同时take()因为二者竞争的是同一把锁。两个Condition只是把等待线程按业务条件分组notEmpty给消费者等待数据notFull给生产者等待空间并不代表有两把锁。所以ArrayBlockingQueue的特点可以概括为固定数组、容量明确、内存稳定、实现简单但put()和take()不能真正并行。在容量边界清晰、希望内存可控的场景中它是一个很稳妥的选择。四、LinkedBlockingQueue用两把锁提高生产消费并发度LinkedBlockingQueue的底层结构不是数组而是链表。链表的一个天然特点是入队主要修改尾部出队主要修改头部。利用这个特点LinkedBlockingQueue把入队和出队拆成了两把锁。锁保护的操作对应条件putLock入队修改链表尾部notFulltakeLock出队修改链表头部notEmpty这和前面的单锁模型形成对比ArrayBlockingQueue中生产者和消费者必须竞争同一把锁LinkedBlockingQueue中只要队列既不空也不满生产者可以持有putLock追加节点消费者可以持有takeLock摘取节点两类操作可以并发推进。两把锁提高了并发度但也引出一个新问题入队和出队都会修改元素数量而它们并不持有同一把锁。为了解决这个共享计数问题LinkedBlockingQueue使用AtomicInteger count。链表尾部由putLock保护链表头部由takeLock保护数量变化由原子变量保护。如果count是普通int生产者执行count、消费者执行count--时可能并发发生导致丢失更新。原子变量使数量变化在两把锁之间仍然保持一致。五、LinkedBlockingQueue 的链表为什么需要哨兵节点LinkedBlockingQueue的链表节点大致包含两个字段item保存元素next指向后继节点。队列内部维护head和last两个指针其中head通常是哨兵节点不保存有效元素真正的第一个元素在head.next。初始状态下队列只有一个哨兵节点head和last都指向它。第一次放入A时入队逻辑可以简化为last.nextnode;lastnode;第一句是把新节点接到旧尾节点后面第二句是把尾指针移动到新节点。放入A后last指向A而不是让A.next指向自己。继续放入B就是把B接到旧尾节点A后面再把last移到B。出队时取的不是head而是head.next。假设当前第一个真实节点是Atake()会读取A.item然后把head移动到A并把A.item置为null。这样原来的A节点变成新的哨兵节点B成为新的第一个真实元素。这里的item null不是原子并发控制手段而是引用清理。由于头部摘取在takeLock下完成同一时间不会有两个消费者同时移动head由于尾部追加在putLock下完成同一时间不会有两个生产者同时修改last。链表指针本身不需要 CAS锁已经提供了互斥保护。哨兵节点让空队列和非空队列的处理更加统一head永远不保存有效元素head.next才是第一个真实元素空队列时head last非空时head.next ! null。六、两把锁拆开后通知也要分成两类在单锁队列中生产者放入元素后直接通知notEmpty消费者取出元素后直接通知notFull。但在LinkedBlockingQueue中notEmpty绑定的是takeLocknotFull绑定的是putLock。这意味着要操作哪个Condition就必须先持有创建它的那把锁。生产者完成入队后如果队列从空变成非空就需要唤醒消费者但消费者等待在notEmpty上所以生产者要临时获取takeLock再执行notEmpty.signal()。反过来消费者完成出队后如果队列从满变成未满就需要临时获取putLock再执行notFull.signal()。状态变化通知对象需要持有的锁队列从空变成非空等待数据的消费者takeLock队列从满变成未满等待空间的生产者putLock这就是两锁模型下的跨边通知。除此之外LinkedBlockingQueue还会做同侧通知生产者放完以后如果队列仍然没满可以继续唤醒其他等待空间的生产者消费者取完以后如果队列仍然有数据可以继续唤醒其他等待数据的消费者。这种接力式通知可以让同侧线程在条件仍然满足时继续推进而跨边通知负责处理空、满边界被打破后的另一类线程唤醒。可以把LinkedBlockingQueue的同步结构压缩成一条链链表头尾分离所以入队和出队可以用两把锁两把锁无法共同保护count所以数量用原子变量两边状态变化会影响对方等待条件所以需要跨边通知同一侧如果条件仍然满足还可以接力唤醒本侧等待线程。下面以生产者执行put(e)为例把同侧通知和跨边通知串起来看。简化后的put()逻辑如下publicvoidput(Ee)throwsInterruptedException{if(enull){thrownewNullPointerException();}intc-1;NodeEnodenewNode(e);finalReentrantLockputLockthis.putLock;finalAtomicIntegercountthis.count;putLock.lockInterruptibly();try{while(count.get()capacity){notFull.await();}enqueue(node);ccount.getAndIncrement();if(c1capacity){notFull.signal();}}finally{putLock.unlock();}if(c0){signalNotEmpty();}}这段代码可以分成三段看。第一段是生产者侧的等待和入队putLock.lockInterruptibly();try{while(count.get()capacity){notFull.await();}enqueue(node);ccount.getAndIncrement();...}finally{putLock.unlock();}生产者只持有putLock。如果队列已满它等待的是notFull如果队列还有空间它就把新节点追加到链表尾部。这里的enqueue(node)只修改尾部指针privatevoidenqueue(NodeEnode){lastlast.nextnode;}由于这段代码在putLock内执行所以多个生产者不会同时修改last尾部追加不需要 CAS。第二段是同侧通知if(c1capacity){notFull.signal();}这里的c是入队之前的元素数量。执行count.getAndIncrement()后队列数量变成c 1。如果c 1 capacity说明当前生产者放完元素后队列仍然没有满还可以继续放入元素于是唤醒另一个等待在notFull上的生产者。这一步仍然发生在putLock保护范围内因为notFull绑定的就是putLock。第三段是跨边通知if(c0){signalNotEmpty();}如果c 0说明入队之前队列是空的当前生产者放入一个元素后队列从空变成非空。消费者原来可能正在notEmpty上等待所以需要唤醒消费者。但notEmpty绑定的是takeLock不能直接在只持有putLock的情况下调用notEmpty.signal()。因此signalNotEmpty()会临时获取takeLockprivatevoidsignalNotEmpty(){finalReentrantLocktakeLockthis.takeLock;takeLock.lock();try{notEmpty.signal();}finally{takeLock.unlock();}}这样生产者的一次put()就完整串起了三件事这个例子体现了LinkedBlockingQueue的通知规则生产者侧的尾部追加由putLock保护数量变化由AtomicInteger count保证放完后如果本侧条件仍然满足就继续唤醒其他生产者如果这次入队打破了“空队列”边界就跨到takeLock一侧唤醒消费者。七、LinkedBlockingQueue 默认近似无界ArrayBlockingQueue创建时必须指定容量而LinkedBlockingQueue可以不传容量newLinkedBlockingQueue();newLinkedBlockingQueue(capacity);无参构造下它的容量是Integer.MAX_VALUE。这不是数学意义上的无限但在大多数业务场景中可以视为近似无界。生产者通常很难因为队列满而阻塞任务会不断堆积最终风险从“线程等待”转移成“内存压力”。创建方式容量特点主要风险new LinkedBlockingQueue()近似无界任务堆积可能 OOMnew LinkedBlockingQueue(capacity)有界满了以后可以形成反压new ArrayBlockingQueue(capacity)有界容量固定内存边界清晰因此LinkedBlockingQueue的优势不是“可以无限放”而是用链表和两把锁提高生产消费并发度。实际使用时通常更应该显式指定容量让系统在消费能力不足时尽早暴露压力而不是把压力隐藏到内存中。八、SynchronousQueue没有容量的直接交接SynchronousQueue虽然也实现了BlockingQueue但它不是用来缓存元素的。它的容量为 0生产者和消费者必须直接配对元素不会进入一个中间缓冲区等待以后再取。普通阻塞队列允许生产者和消费者在时间上错开SynchronousQueue则要求双方同时在场如果生产者先调用put(A)但没有消费者正在take()生产者就会阻塞如果消费者先调用take()但没有生产者正在put()消费者也会阻塞。非阻塞的offer(A)在没有等待消费者时会直接返回falsepoll()在没有等待生产者时会直接返回null。这个特性在线程池中非常重要。使用普通队列时核心线程忙了以后任务可以先排队使用SynchronousQueue时任务没有地方缓存只能直接交给空闲工作线程。如果没有空闲线程接收线程池就更倾向于创建新线程直到达到maximumPoolSize或执行拒绝策略。SynchronousQueue还支持公平模式和非公平模式newSynchronousQueue();// 默认非公平newSynchronousQueue(true);// 公平模式两种模式不改变“容量为 0、不能缓存元素”的本质只影响等待配对线程的匹配顺序。模式内部结构倾向匹配顺序优点代价非公平模式栈LIFO后等待的线程可能先匹配吞吐通常更高早等待的线程可能等待更久公平模式队列FIFO先等待的线程优先匹配顺序更稳定吞吐通常低一些默认非公平模式更偏性能适合大多数线程池直接交接任务的场景公平模式更偏顺序稳定适合更关注等待顺序而不是极限吞吐的场景。九、PriorityBlockingQueue按优先级取而不是按插入顺序取PriorityBlockingQueue的出队顺序由元素优先级决定而不是由入队时间决定。默认情况下元素越“小”越先出队因为底层是小顶堆。最简单的例子是整数队列PriorityBlockingQueueIntegerqueuenewPriorityBlockingQueue();queue.put(30);queue.put(10);queue.put(20);System.out.println(queue.take());// 10System.out.println(queue.take());// 20System.out.println(queue.take());// 30如果队列中放的是任务对象就需要给任务定义比较规则。下面这个例子中priority越小优先级越高如果两个任务优先级相同再用sequence保证先入队的任务先出队。importjava.util.concurrent.PriorityBlockingQueue;importjava.util.concurrent.atomic.AtomicLong;publicclassPriorityBlockingQueueDemo{staticclassPriorityTaskimplementsComparablePriorityTask{privatestaticfinalAtomicLongsequencernewAtomicLong();privatefinalStringname;privatefinalintpriority;privatefinallongsequence;PriorityTask(Stringname,intpriority){this.namename;this.prioritypriority;this.sequencesequencer.getAndIncrement();}OverridepublicintcompareTo(PriorityTaskother){intresultInteger.compare(this.priority,other.priority);if(result!0){returnresult;}returnLong.compare(this.sequence,other.sequence);}OverridepublicStringtoString(){returnname(prioritypriority, sequencesequence);}}publicstaticvoidmain(String[]args)throwsInterruptedException{PriorityBlockingQueuePriorityTaskqueuenewPriorityBlockingQueue();queue.put(newPriorityTask(normal-1,10));queue.put(newPriorityTask(urgent-1,1));queue.put(newPriorityTask(important-1,5));queue.put(newPriorityTask(urgent-2,1));System.out.println(queue.take());System.out.println(queue.take());System.out.println(queue.take());System.out.println(queue.take());}}输出顺序大致是urgent-1(priority1, sequence1) urgent-2(priority1, sequence3) important-1(priority5, sequence2) normal-1(priority10, sequence0)这里体现了两层规则先按priority排序优先级更高的任务先出队如果priority相同再按sequence排序先入队的任务先出队。如果compareTo()只比较priorityOverridepublicintcompareTo(PriorityTaskother){returnInteger.compare(this.priority,other.priority);}那么同优先级任务在比较器看来是相等的PriorityBlockingQueue不保证它们按插入顺序出队。原因是它底层维护的是堆结构堆只保证优先级关系不维护同优先级元素的 FIFO 顺序。PriorityBlockingQueue也是无界队列。它的put()通常不会因为容量满而阻塞真正会阻塞的是队列为空时的take()。所以它适合任务确实存在优先级的场景但不适合用来做容量反压。生产速度长期超过消费速度时优先级队列同样会不断膨胀带来内存风险。十、DelayQueue队列非空也不一定能取DelayQueue可以看成按到期时间排序的阻塞队列。它和PriorityBlockingQueue类似底层都依赖优先级排序区别在于DelayQueue的优先级不是普通大小而是元素的到期时间。队列中的元素必须实现Delayed接口publicinterfaceDelayedextendsComparableDelayed{longgetDelay(TimeUnitunit);}其中compareTo()用来决定元素的排序通常是谁更早到期谁排在前面getDelay()用来判断当前元素还剩多久到期。只有堆顶元素的延迟时间小于等于 0 时take()才能真正取出它。例如有三个延迟任务A10 秒后到期 B3 秒后到期 C5 秒后到期即使插入顺序是A → B → C取出顺序也会是B → C → A因为DelayQueue关注的是到期时间而不是插入顺序。这一点会带来一个和普通阻塞队列不同的行为队列非空take()也可能继续阻塞。普通队列只要非空消费者就可以取元素但DelayQueue会先看堆顶元素是否到期。如果堆顶没到期说明后面的元素只会更晚到期所以消费者没有必要扫描整个队列只需要等待堆顶元素到期。可以简单理解为多个消费者同时等待时DelayQueue内部使用leader优化。它不会让所有消费者都等待同一个到期时间而是只让一个线程作为 leader负责等待最近的到期时间其他线程普通等待。这样可以避免到期时间一到多个消费者同时醒来抢锁最后却只有一个线程真正取到元素。Consumer-1: leader, waits until top expires Consumer-2: awaits normally Consumer-3: awaits normally如果生产者新放入的任务比原堆顶更早到期原来的等待时间就不准确了。这时队列会清空 leader 并唤醒一个等待线程让它重新检查新的堆顶任务。所以DelayQueue的核心可以压缩成一句话DelayQueue是按到期时间释放元素的阻塞队列队列非空不代表可以取只有最早到期的元素真正到期后take()才会返回。它也属于无界队列put()通常不会因为容量满而阻塞。真正阻塞的是没有到期元素时的take()。因此它适合延迟任务、订单超时关闭、缓存过期清理、定时重试等场景但不适合用来做容量反压。十一、BlockingQueue 在线程池中决定任务是排队还是扩容ThreadPoolExecutor的构造参数中workQueue就是BlockingQueueRunnable。它不仅是任务容器还会影响线程池的扩容路径。简化后的execute()流程是这里有一个容易忽略的点线程池提交任务时入队通常调用的是offer()不是put()。也就是说队列满了以后提交线程不会在put()上一直阻塞而是入队失败然后线程池尝试创建非核心线程如果线程数也达到上限才执行拒绝策略。不同队列会让线程池表现出不同倾向队列线程池行为倾向主要风险ArrayBlockingQueue核心线程满后先排队队列满后再扩容队列太小容易拒绝太大延迟升高有界LinkedBlockingQueue类似先排队满后扩容但生产消费并发度更高节点对象更多GC 压力较大无界LinkedBlockingQueue几乎一直排队maximumPoolSize不易发挥作用任务堆积可能 OOMSynchronousQueue不缓存任务倾向直接交接或创建新线程线程数可能快速增长PriorityBlockingQueue按优先级排队无界堆积同优先级不保证 FIFODelayQueue按到期时间取任务无界堆积只适合延迟任务模型因此线程池参数不能只看corePoolSize和maximumPoolSize。如果队列是无界的maximumPoolSize往往很难发挥作用如果队列是SynchronousQueue线程池又会更积极地创建新线程。队列选择本身就是线程池调度策略的一部分。十二、几个常见使用误区第一个误区是把BlockingQueue等同于“安全无风险的队列”。它确实解决了多线程并发访问和空满等待问题但不自动解决容量设计问题。无界队列不会因为队列满而阻塞生产者压力会转移到内存有界队列虽然能形成反压但容量过大也会让任务等待时间变长故障暴露更晚。第二个误区是以为BlockingQueue一定会让生产者在队列满时阻塞。是否阻塞取决于调用的方法和具体队列。put()满了会等待offer()满了会返回失败无界队列的put()通常不会因为容量满而等待线程池内部也更常使用offer()决定是否入队。第三个误区是忽略队列的出队顺序。ArrayBlockingQueue和LinkedBlockingQueue是 FIFOPriorityBlockingQueue按优先级DelayQueue按到期时间SynchronousQueue根本不缓存元素。它们都实现了BlockingQueue但调度语义完全不同。本章总结本章的因果链可以从生产者消费者速度不一致开始理解普通队列只能保存元素不能决定空或满时线程如何等待Condition提供了释放锁并等待业务条件的能力BlockingQueue则把这种能力封装到队列接口中并通过不同方法表达抛异常、返回特殊值、一直等待和超时等待这几种失败策略。在具体实现上数组结构引出了循环下标和单锁保护得到的是容量明确、内存稳定但并发度较低的ArrayBlockingQueue链表结构让头尾操作可以拆开进一步引出两把锁、原子计数和跨边通知得到的是生产消费并发度更高但对象开销更大的LinkedBlockingQueue。再往后队列不一定只是 FIFO 缓冲区SynchronousQueue把缓存取消变成生产者和消费者的直接交接PriorityBlockingQueue把出队顺序交给优先级DelayQueue又把优先级具体化为到期时间。所以BlockingQueue最终解决的不是“用哪个容器存任务”这么简单而是把线程等待、容量边界、出队顺序和任务调度策略绑定在一起。尤其在线程池中队列类型会直接决定任务是先排队、先扩容、按优先级执行还是必须直接交给工作线程。理解阻塞队列就是理解生产速度、消费速度和系统承压方式之间的关系。