上一篇用Semaphore控制同时进入处理区的任务数量。接着往前想一步任务来得快、处理得慢暂时做不完的工作该放在哪里例如一个文件处理程序前台负责接收文件后台负责转换格式。两部分可以通过队列交接但还要回答三个问题队列满了怎么办、暂时没任务怎么办、最后怎样结束等待。BlockingQueue把前两个问题变成明确的方法选择第三个问题则需要业务自己设计。本文以Java 21为适用版本使用容量固定的ArrayBlockingQueue说明机制。API语义按2026年9月25日的官方文档核对。一、先画清楚任务怎样流动生产者创建任务放入队列消费者取出任务再执行处理。队列让双方不必在同一个调用栈里同步完成工作。有界队列为“尚未取走的任务”设定容量。假设容量是2两个格子都占满时新的put调用需要等待空位消费者take取走一项后空位才有机会被等待的生产者使用。队列空时take会等待元素到来。这里的“有机会”很重要多线程竞争下刚空出的格子可能被其他生产者拿走不应把唤醒理解为已经拿到容量。图中暂存格只表示排队区。任务取出后仍可能正在处理所以队列容量不等于系统内任务总数。容量2配一个消费者可能出现“两项等待、一项处理中”的状态。ArrayBlockingQueue使用固定容量的数组按FIFO顺序取出元素。FIFO描述元素的出队顺序多个消费者各自耗时不同完成顺序仍可能变化。官方说明二、满了或空了选择哪种方法先决定业务愿意怎样等待再挑API比背方法名更实用。处理策略入队取出典型用途无法立即完成就抛异常add(e)remove()把异常状态交给异常处理立即返回结果offer(e)poll()调用方自己决定是否重试等待条件满足put(e)take()后台工作线程持续交接最多等待指定时间offer(e, time, unit)poll(time, unit)有明确等待预算的操作offer失败返回falsepoll取不到元素返回null。队列本身不接受null因此不能用null作为“工作结束”的标记。put、take及带超时的方法可以在等待时响应中断。BlockingQueue方法约定对HTTP请求入口可以考虑在有限时间内尝试交接失败后返回明确的繁忙结果。后台批处理则可以允许生产者等待。选择取决于调用方的时间预算不宜默认所有入口都无限等待。还有一个常见误区if(queue.remainingCapacity()0){// 此刻有空位不代表下一行执行时仍有空位}观察值适合监控是否入队成功应检查实际入队操作的结果。三、完整示例处理三项任务再正常收尾下面把“容量测试”和“生产消费”分开演示。前半段没有并发便于看清offer的返回值后半段使用一个生产者和一个消费者通过专用结束对象收尾。保存为BlockingQueueDemo.javaimportjava.util.concurrent.ArrayBlockingQueue;importjava.util.concurrent.BlockingQueue;publicclassBlockingQueueDemo{recordJob(intid,booleanend){}privatestaticfinalJobENDnewJob(-1,true);publicstaticvoidmain(String[]args)throwsInterruptedException{BlockingQueueIntegerprobenewArrayBlockingQueue(2);System.out.println(offer 1: probe.offer(1));System.out.println(offer 2: probe.offer(2));System.out.println(offer 3: probe.offer(3));System.out.println(poll: probe.poll());BlockingQueueJobqueuenewArrayBlockingQueue(2);ThreadconsumernewThread(()-{try{while(true){Jobjobqueue.take();if(job.end()){System.out.println(consumer stopped);return;}System.out.println(processed: job.id());}}catch(InterruptedExceptione){Thread.currentThread().interrupt();}},consumer);ThreadproducernewThread(()-{try{for(inti1;i3;i){queue.put(newJob(i,false));}queue.put(END);}catch(InterruptedExceptione){Thread.currentThread().interrupt();}},producer);consumer.start();producer.start();try{producer.join();consumer.join();}finally{// 主线程等待被中断时也向两个工作线程发出取消信号。producer.interrupt();consumer.interrupt();}}}编译与执行javac--release21BlockingQueueDemo.javajavaBlockingQueueDemo正常运行、没有外部中断时按代码逻辑可得以下输出offer 1: true offer 2: true offer 3: false poll: 1 processed: 1 processed: 2 processed: 3 consumer stopped这段示例没有通过sleep猜测线程调度顺序。探测队列只在主线程中使用工作队列只有一个消费者打印处理结果因此更容易推导输出。END排在普通任务之后消费者顺序取出三项工作后才结束。示例刻意限定为单生产者、单消费者便于聚焦交接过程它不包含业务重试、持久化或进程崩溃恢复。四、结束信号要覆盖真实的参与者BlockingQueue没有统一的close或shutdown方法。结束对象是一种应用层协议不能只照抄一个END就认为所有线程都会退出。如果有三个消费者一枚结束对象通常只会被其中一个取走。可以在所有生产者确认结束后由协调者投递三枚结束对象或设计统一的取消机制。关键是先确定不会再有普通任务入队再发送终止信号。如果处理代码抛出未捕获异常消费者可能提前退出生产者却仍阻塞在满队列上。真实程序应让监督者观察工作线程失败取消同组任务并决定未处理数据的去向。只在消费者里打印异常通常不足以完成恢复。同样线程池shutdown管理的是线程池生命周期并不会自动替你给一个独立BlockingQueue发送业务结束信号。五、有界队列把压力传到哪里把容量从2调到2000会容纳更多排队任务也会增加等待时间和内存占用。它不会提高单个消费者的处理速度。背压在这里体现为生产速度超过消费速度后缓冲区用完生产者开始等待或获得失败结果。上游如果忽略失败、无限重试或者把任务转存进另一个无限增长的集合压力仍然没有被控制住。需要控制的量可以关注的机制容易遗漏的部分等待处理的任务数有界BlockingQueue任务对象本身占用多少内存同时处理的任务数消费者数量、Semaphore一次处理是否又发起更多异步工作单位时间接收量速率限制策略突发流量与拒绝后的重试可接受的等待时长超时入队、请求截止时间超时之后任务是否已经被接收ArrayBlockingQueue还可选择公平访问模式。它影响等待线程取得访问机会的顺序并不改变元素FIFO这一基本特征是否开启应结合负载测量不能直接等同于更快。六、 思维导图BlockingQueue交接模型生产者入队消费者取出再处理等待策略put与take等待offer与poll返回结果有界容量限制排队数量满队列向上游传递压力收尾协议生产完成后发送结束信号失败时协调取消边界出队有序不保证完成有序容量不等于每秒速率七、总结总结要点BlockingQueue把生产者与消费者的交接规则放到了清晰的接口里。先决定立即返回、持续等待还是限时等待再选择对应的方法。有界容量让积压变得可控但还需要上游认真处理失败和超时并观察队列外正在执行的工作。容量、并发数和速率各有作用。正常结束与异常取消同样属于设计的一部分。示例中的结束对象适合说明协议参与者增加后要把所有生产者、消费者和失败路径一起考虑。下一篇继续Java进阶系列看看ConcurrentHashMap并发读写、原子更新以及复合操作容易踩到的坑。如果你觉得这篇文章对你有所帮助欢迎点赞、收藏、分享