生产者-消费者模式这四个词在并发编程里几乎等同于“入门必修课”但很多人学完就扔在一边到了真实项目里遇到订单积压、日志丢失、线程池打满才发现自己根本没吃透它。我入行这些年在支付系统、设备数据采集、定时任务调度里都栽过跟头最后发现绝大多数并发问题都能收敛到一个朴素的结论中间一定要有一个队列生产的速度和消费的速度不能互相堵死。这次我围绕“生产者-消费者模式和并行任务调度”重写了一版完整实现没有追求花哨的并发原语只用最稳妥的JDK内置组件但把每一处改进都落了解释和注释。这篇文章适合两类人看一类是刚把synchronized、BlockingQueue背得滚瓜烂熟、但没做过完整并发模块的开发者另一类是在线上被线程池炸过、想系统梳理任务调度要点的工程师。我会把从手写wait/notify到使用阻塞队列、再到线程池并行消费的完整演进路线讲清楚同时把“为什么这么改”“为什么注释要这么写”这些平常容易被忽略的部分补上。1. 内容整体设计与思路拆解1.1 生产者-消费者模式的核心是解耦和削峰先明确一个认知生产者-消费者模式并不是什么高深的数学结构它就是生产数据和消费数据之间隔着一个队列。我在代码里经常用一句话概括它的价值让上游和下游之间不再互相等待。真实业务里接口请求的到达是突发的而下游的数据库写入、远程调用、文件落盘都是有吞吐上限的如果直接让请求线程阻塞在业务处理上整个系统的响应时间会被最慢的那个环节拖垮。中间加一个队列之后生产者只需要把任务丢进队列就返回消费者按照自己的节奏处理这就是解耦。削峰是第二个核心价值。我做过一个设备上报数据的接入服务凌晨两点有一波设备集中上报瞬时QPS能飙到正常情况的十倍如果系统硬扛这波流量数据库连接池会直接被冲垮。当时就是在接入层和应用层之间加了一个有界队列把峰值流量先存在队列里让应用层按照自己的最大处理能力慢慢消费。效果是数据库负载曲线从尖峰变成了平缓的长尾整晚没报警。理解了这个场景再看并行任务调度就顺理成章了单一消费者处理不过来就在队列的另一侧部署多个消费者线程它们并行地从同一个队列里拿任务处理。由此引出三个关键设计问题队列用什么结构、并发边界怎么控制、消费者线程数设为多少。整篇文章的代码演进就是围绕这三个问题一步步展开的。1.2 这次工程改进的另一个重点是注释质量代码健壮性之外这次我特意要求自己把每处注释都重新写一遍并且规定了两条原则注释只解释“为什么”不解释“是什么”能够从变量命名和代码结构中直接读出来的信息不写进注释。原因很直接我吃过“看起来注释很多、实际等于没写”的亏。翻维护过一段老代码满屏都是// 添加数据、// 返回结果这种注释等于把代码读了一遍但遇到真正需要知道的关键信息——比如“这个方法为什么必须加锁”“这个超时时间为什么是10秒”——注释里一个字都没有。后来我总结出一套适合并发代码的注释习惯这次也会在代码里示范字段注释用一个名词短语说清楚它的约束条件比如“队列容量200超过后上游自行降级”方法注释写明前置条件和失败路径比如“offer返回false表示队列已满调用方必须处理该分支”锁边界和线程模型在关键方法上一两句话标注清楚。注释不是写得越多越好关键位置的几句精炼注释价值远超一百行废话。1.3 演进路线设计每一步都对应一个明确的痛点这次的代码分三个版本递进每个版本都解决前一个版本暴露出的具体问题。第一版是手写synchronized配合wait/notifyAll实现的经典模型。这个版本的意义在于展示并发控制的基本原理所有用到锁、等待、唤醒的细节都在眼前适合理解底层机制但不适合直接上生产。第二版用BlockingQueue替代手写同步容量参数、阻塞语义、线程安全都由容器帮我们管理代码量减少一半出错概率也大幅降低。这一版已经可以用于绝大多数中小型业务。第三版引入线程池做并行任务调度把单一消费者升级为多个消费者并发运行同时补上线程池核心参数的选择、线程命名、优雅关闭等生产必须的细节。这一版的目标是能扛住真实流量。由于版本之间是递进关系每一处改动都有明确原因所以对应的注释也能更清晰地表达“为什么旧做法有问题、新做法好在哪”。这正是标题里“每项改进的详细解释”的核心呈现方式。2. 核心细节解析与实操要点2.1 手写wait/notify版本的三个关键约束第一版虽然不推荐直接用于生产但理解它的约束对后面吃透阻塞队列有很大帮助。完整骨架如下public class ProducerConsumerV1 { // 共享缓冲区LinkedList非线程安全必须靠synchronized约束访问 private final LinkedListInteger buffer new LinkedList(); private final int capacity 10; // 生产队列满时等待不满时放入并唤醒消费者 public synchronized void produce(int item) throws InterruptedException { while (buffer.size() capacity) { wait(); } buffer.add(item); notifyAll(); } // 消费队列空时等待不空时取出并唤醒生产者 public synchronized int consume() throws InterruptedException { while (buffer.isEmpty()) { wait(); } int item buffer.removeFirst(); notifyAll(); return item; } }这个版本有三个必须遵守的细节。第一个判断条件必须用while而不是if。因为wait被唤醒之后队列的状态可能已经被其他线程改变了如果用if唤醒后会直接往下执行可能把数据放进一个本就已满的队列。多线程环境下“虚假唤醒”是真实存在的while循环让线程醒来后重新检查条件这是保险丝一样的存在。第二个wait和notifyAll必须放在synchronized代码块内这是Java内置锁的硬性要求。主线程外还有一个隐藏规则值得注意消费者被唤醒后需要重新竞争内置锁所以锁的获取顺序和释放时机直接影响整体吞吐。第三个notifyAll几乎总是比notify安全。用notify只唤醒一个线程如果唤醒的是一个消费者而队列状态实际上需要生产者生产就可能出现所有线程都在等待的死锁场景。notifyAll会唤醒所有等待线程代价是唤醒的线程里会有一些重新检查条件后再次等待但安全性远远优先于这一点性能损耗。这个版本的注释写法正好呼应前面的注释原则不写“从缓冲区取出元素”这种废话而是把“正在等待的条件”和“锁的约束边界”标清楚。2.2 用BlockingQueue替代手写同步是编写者的减负第二版的核心改动是把缓冲区替换为LinkedBlockingQueue同步控制交给JDK容器。这里对队列结构的选择会直接影响系统行为我用一张表把常用阻塞队列的差异列出来方便你在工程选型时对照。队列类型是否支持有界数据结构典型使用场景ArrayBlockingQueue支持必须指定容量数组需要严格控制内存占用容量可预估LinkedBlockingQueue支持不传容量则无边链表默认无边有隐患建议显式传容量SynchronousQueue不存储元素无内部缓冲直接交接适合一对一传递PriorityBlockingQueue无界堆任务按优先级出队注意不会阻塞生产者我在生产环境里最常用的是LinkedBlockingQueue显式指定容量因为链表结构在并发读写场景下head和tail分离冲突比数组小。不过ArrayBlockingQueue也有自己的优势容量固定后有更好的内存可预测性。选择哪个的关键不是性能而是你对业务积压容量的判断。这一版的代码简化到了几乎不需要同步注释的程度public class TaskQueueV2 { // 队列容量100既是积压上限也是背压阈值 private final BlockingQueueTask queue new LinkedBlockingQueue(100); // 生产非阻塞入队队列满时返回false由调用方决定重试或降级 public boolean offer(Task task) { return queue.offer(task); } // 消费轮询取任务超时返回null用于支持优雅退出 public Task poll(long timeout, TimeUnit unit) throws InterruptedException { return queue.poll(timeout, unit); } }注意这里我特意没有使用put和take而是选择了offer和poll。背后是可控性的考量put在队列满时会无限期阻塞调用方完全无法感知队列状态offer则立刻返回结果让上游决定如何应对积压。在生产系统中“尽快感知压力”比“死等一个位置”重要得多。这也是一个非常重要但又很容易被忽略的设计选择。2.3 并行任务调度要解决的是把任务分发到多个消费者第三版引入了线程池做并行消费这自然带来了ExecutorService的使用问题。核心是五个参数核心线程数、最大线程数、空闲存活时间、任务队列、饱和策略。这里不展开基础概念只讲生产环境下几个容易踩坑的点。线程池必须显式指定线程工厂并命名线程。如果不设置线程名会是pool-1-thread-1这样的流水号线上排查问题时日志里全是数字分不清哪个线程在干什么。我通常会这样创建线程池ExecutorService consumerPool Executors.newFixedThreadPool( consumerNum, r - { Thread t new Thread(r, order-consumer); t.setDaemon(true); return t; } );这里把线程设置成daemon是另一个值得解释的细节如果业务主线程退出后不希望消费者线程继续阻塞整个进程守护线程是更安全的选择。但如果你的消费者承担着必须落盘的任务这个设置反而危险主进程退出时会直接丢掉未完成的任务。所以这个参数要看你想要的行为不要在多个项目里直接粘贴同一套配置。饱和策略的选择也是一个容易被忽视的点。默认AbortPolicy在队列满时会直接抛异常这相当于用异常告诉调用方“处理不过来了”。在部分场景里这个策略是合理的但如果策略设置不当异常会变成吞入的故障。我的建议是无论选择哪种策略都要保证失败行为有日志有监控不能让任务无声消失。2.4 线程池线程数的估算是起点不是终点线程数开多少这个问题几乎每次上线前都会被问到。工程上有一个很实用的估算公式N_threads N_cpu * U_cpu * (1 W/C)其中N_cpu是CPU核心数U_cpu是目标CPU利用率0到1之间W/C是等待时间和计算时间的比值。这个公式的核心逻辑是IO等待时间越长为了占满CPU就需要越多线程。举一个我实际测算过的例子。一个订单处理服务部署在4核机器上每个任务内部有一次耗时约500ms的远程调用本地纯计算约10ms则W/C50。目标是让CPU利用率维持在40%左右那么N_threads 4 * 0.4 * (1 50)约等于82。很多人看到82会吓一跳但IO密集型任务的线程数确实不能依靠“核心数加一”来估算因为那个公式只适用于纯CPU计算场景。还有一个粗略的参考口径IO密集型按2 * N_cpu起步CPU密集型按N_cpu 1起步。但请记住这些数字只是让你有一个安全的起点最终必须用压测验证。我在项目里通常的做法是先按公式设定然后逐步调整压测时观察队列水位和处理延迟找到最优区间。3. 实操过程与核心环节实现3.1 场景设定模拟一个订单接收与处理管道为了让代码不悬浮在抽象概念上我用一个具体的业务场景贯穿全文假设有一个订单接收服务外部系统调用接口提交订单服务需要把订单数据写入本地日志并进行下游同步处理。生产者的任务是把订单放进队列消费者的任务是从队列取出订单并执行处理逻辑。这个场景足够简单却同时包含了生产者-消费者模式和并行任务调度两个核心问题上游接口的到达速率不可控下游处理包含IO操作需要较长耗时。3.2 完整实现核心代码与逐段解释下面给出第三版完整代码注释按前面的原则做了精简每个关键决策都标注了原因。import java.util.concurrent.*; public class OrderPipeline { // 队列容量200对应系统允许的积压需求超过后submit直接失败 private final BlockingQueueOrder queue new LinkedBlockingQueue(200); // 消费者线程池固定线程数避免动态扩容引起线程竞争 private final ExecutorService consumerPool; // 消费者数量由业务压测确定构造函数注入便于调整 private final int consumerNum; public OrderPipeline(int consumerNum) { this.consumerNum consumerNum; this.consumerPool Executors.newFixedThreadPool(consumerNum, r - { Thread t new Thread(r, order-consumer); t.setDaemon(true); return t; }); } // 生产者入口非阻塞入队队列满返回false由调用方决定是否丢弃或重试 // 这里不用put因为put无限阻塞会让上游感知不到背压 public boolean submit(Order order) { return queue.offer(order); } // 启动消费为每个消费者提交一个循环任务 public void startConsumers() { for (int i 0; i consumerNum; i) { consumerPool.execute(() - { while (!Thread.currentThread().isInterrupted()) { try { Order order queue.poll(1, TimeUnit.SECONDS); if (order ! null) { handleOrder(order); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } }); } } // 业务处理模拟IO耗时操作实际场景可替换为DB写入或远程调用 private void handleOrder(Order order) throws InterruptedException { Thread.sleep(50); } // 优雅关闭先停止接收新任务再等待剩余任务完成 public void shutdown() { consumerPool.shutdown(); try { if (!consumerPool.awaitTermination(10, TimeUnit.SECONDS)) { consumerPool.shutdownNow(); } } catch (InterruptedException e) { consumerPool.shutdownNow(); Thread.currentThread().interrupt(); } } }这段代码里有两个值得单独提的注释点。第一个是submit方法里的注释它说明了一个行为动机而不是行为本体读者看到这一句就能理解为什么这里不调用阻塞的put。第二个是handleOrder方法的注释它指出这个位置可以替换成真实业务明确了代码骨架的职责边界。消费者循环里poll(1, TimeUnit.SECONDS)的设计也有讲究。如果使用take()无限阻塞线程在等待时被中断不会响应而每次最多阻塞1秒的轮询方式让线程有机会定期检查中断标志从而实现优雅退出。这个细节是很多线上系统关闭后线程无法退出的元凶之一。3.3 运行观察怎么确认并行生效了跑起来之后想确认多个消费者真的在并不同时处理任务最直观的办法是看日志里的线程名。在startConsumers启动前给handleOrder方法临时加一行System.out.println(Thread.currentThread().getName() handle order order.getId())日志里会出现多个order-consumer-1、order-consumer-2这样的线程名交替输出。如果日志中始终只有一个线程名说明消费者线程没有真正并行常见原因有两个线程池被错误地配置成了newSingleThreadExecutor或者业务代码里有锁把并行度压成了串行。前者是配置问题后者更隐蔽比如所有消费逻辑都经过同一个synchronized方法那无论开多少线程都白搭。观察队列水位也是重要的监控手段。可以定时输出queue.size()如果这个数值持续上涨说明消费能力跟不上生产速度除了加消费者线程之外还要检查处理逻辑本身是否有可优化的IO等待。4. 常见问题与排查技巧实录4.1 消费者处理不过来的第一反应别是加线程队列积压时很多人的第一反应是调大线程池数量。但线程不是越多越好。我遇到过线上消费者线程从8调到32结果吞吐反而下降的情况原因在于所有业务共用一个数据库连接池消费者线程增加后连接池成为瓶颈大量线程阻塞在获取连接的等待上上下文切换开销反而拖垮了性能。正确的排查顺序是先看队列积压曲线再看消费者线程的等待状态然后用jstack抓线程栈确认阻塞点最后才决定是否扩容线程。4.2 线程池队列满了之后任务去哪了这是我要提醒的最重要的一类问题。我在排查一个偶发性丢单问题时发现的根本原因就是默认参数下ThreadPoolExecutor的AbortPolicy在队列满时直接抛出了RejectedExecutionException而上游代码catch住了这个异常但没有记录完整日志导致看起来任务“凭空消失”。下面这张表概括了四种饱和策略的差异方便你对照业务选择合适的策略。策略行为适用场景AbortPolicy抛出RejectedExecutionException默认适合能接受异常处理的任务CallerRunsPolicy提交任务的线程自己执行适合需要降速保护的上游DiscardPolicy静默丢弃适合允许丢数据的场景DiscardOldestPolicy丢弃队列最旧的任务适合追求新任务优先的场景我的建议是在核心链路中不要使用任何静默丢弃策略至少要有一行日志记录被拒绝的任务数量。如果你需要背压感知CallerRunsPolicy是一个不错的折中。4.3 用了并发安全队列为什么业务数据还会错乱一个容易踩的坑是队列本身线程安全但消费者从队列取出对象后对对象内部状态的修改不是线程安全的。比如多个消费者线程拿到了同一个订单对象同时修改这个对象的字段就会出现数据错乱。这类问题在排查看起来很诡异但本质还是对象共享缺少保护。解决办法有三个方向使用不可变对象取出后不允许修改对可变字段使用AtomicReference等原子类或者给临界区加锁。从注释角度说这类共享可变对象的边界应该被明确标注出来否则接手的同事很可能在无意中破坏这个约定。4.4 注释不是写文档别让烂注释淹没真信息最后回到注释这件事。我看过很多“注释模板泛滥”的工程IntelliJ IDEA新建类时自动生成的Created by xxx on 2024/xx/xx每个人都保留着然而这类注释对阅读者毫无信息量。我在代码里示范的简洁注释只关注三类信息约束条件、行为动机、并发边界。约束条件比如“队列容量200”行为动机比如“不用put因为需要感知背压”并发边界比如“必须在synchronized内调用wait”。如果你想让团队注释风格统一可以在IDEA中配置一组注释模板但模板里的变量只保留作者名和功能描述不要默认生成日期和“Created by”这种机器痕迹。字段注释应当说出这个字段的约束而不是把变量名翻译成中文。举个例子// 任务缓冲区容量为200超过后submit返回false是有效注释// 任务列表就是废话。5. 总结这些经验都是踩坑踩出来的我自己的习惯是把这套代码放在项目里的concurrent包下从最基础的版本开始维护每次遇到新的并发问题就往里面加一个新的变体从而形成属于团队的并发模式库。在实际项目中真正帮到我的不是哪一行具体的语法而是对“生产消费平衡”的敏感度系统但凡出现偶发的数据错乱、任务消失、线程假死我都会先检查队列状态、线程池状态和消费者循环的退出条件这几个点排查完后绝大部分问题都能定位到根因。特别是那个简洁注释的约定让每次回看代码都省了不少时间——因为我写下的不是代码在做什么而是“为什么当时要这么做”。这也是“生产者-消费者模式 并行任务调度 一个完整并发骨架”这句话的最初由来希望你也能写出比这更好用的版本。