生产者-消费者模式与并行任务调度：从BlockingQueue到虚拟线程的工程实践

我想先把这次做的东西说清楚——这个项目围绕的是生产者-消费者模式、并行任务调度，以及一个经常被忽略的细节：更简洁的注释和每项改进的详细解释。我自己维护过一套高吞吐的通知推送组件，早期代码就是“能跑就行”的水平，队列选型靠拍脑袋，线程池参数全是默认值，注释写了等于没写，后来踩了数据丢失、线程阻塞、代码完全没人敢改的坑，才回过头来用这套思路彻底重构了一遍。

适合读这篇文章的人有两类：一类是刚接触并发编程，想知道 BlockingQueue、线程池、虚拟线程到底各自解决什么问题；另一类是已经写了几个月生产者-消费者代码，但总觉得哪里别扭，想看看“规范版本”长什么样。这篇内容不会只丢给你一段能跑的代码，而是把我踩过的坑、每一步改进背后的原因、包括注释为什么要那样写，全部拆开讲清楚。

1. 先搞清楚：生产者-消费者模式在工程里到底解决什么问题

1.1 生产慢、消费慢，怎么让他们别互相拖累

很多人在面试里背过生产者-消费者模式的定义，但真到了业务里，未必知道它到底在解什么题。我用一个日常场景说明：你去咖啡店点单，收银员是生产者，咖啡师是消费者，中间那个放订单的台子就是队列。

如果台子太小，收银员每一次都要等咖啡师做完一杯才敢接下一单，点单速度被拖死；如果台子太大，订单堆得到处都是，消费者又忙不过来。生产者-消费者模式要做的事就是：在“产生任务”和“执行任务”中间加一层缓冲，让两边速度不匹配时不至于互相阻塞。

工程里最常见的应用场景是这几类：

• 消息推送：业务系统产生推送请求，推送服务异步消费，高峰期削峰
• 日志收集：应用线程写日志到队列，后台线程批量刷盘
• 批量写库：上游产生大量变更，下游攒一批再写，减少数据库连接压力
• 爬虫任务解析：抓取任务一条条进队列，解析线程并行处理
• 订单状态通知：订单系统产生事件，通知服务消费并发送短信/邮件

你会发现它们的共同点：生产端和消费端的速率天然不匹配。比如订单瞬间暴涨时，数据库只能每秒写200条，但上游每秒能产生2000个事件，这时候队列就是缓冲水池，让消费端稳定的200条/秒慢慢消化，而不是被瞬时流量冲垮。

实际业务中，它带来的核心价值是四个：

• 解耦：生产者完全不需要知道下游有几个消费者、每个消费者怎么处理
• 缓冲削峰：瞬时流量压进来时，消息先落在队列里，消费端按自己的节奏处理
• 可用性提升：消费端挂掉或者处理失败时，生产端可以继续投递，配合重试机制能兜底
• 天然支持并行：同一个队列可以挂多个消费者，这是后续并行任务调度的基础

1.2 三种经典实现方式对比

实现生产者-消费者模式，Java 里大体有三条路：synchronized + wait/notify、Lock + Condition、BlockingQueue。初学者最容易被前两种绕晕，我直接给对比表。

先看第一版最原始的写法，网上教材里最常见：

public class BasicProducerConsumer { private final Queue<String> queue = new LinkedList<>(); private static final int CAPACITY = 1000; public synchronized void produce(String item) throws InterruptedException { while (queue.size() == CAPACITY) { wait(); // 队列满了，暂时停下等待消费端取走 } queue.offer(item); notifyAll(); // 唤醒所有等待的消费者 } public synchronized String consume() throws InterruptedException { while (queue.isEmpty()) { wait(); // 队列空了，等待生产端放入数据 } String item = queue.poll(); notifyAll(); // 唤醒生产者 return item; } }

这段代码看起来没问题，但这里有几个隐藏的细节。

第一个：wait 为什么必须放在 while 而不是 if 里。因为存在“虚假唤醒”，线程可能在没有被 notify 的情况下醒过来，或者被 notifyAll 唤醒后又发现条件不满足。用 if 只会判断一次，醒来直接往下走，队列还是满的或空的，就会出问题。while 是二次确认条件，这是长期实践留下的铁律。

第二个：为什么用 notifyAll 而不是 notify。notify 只随机唤醒一个线程，如果唤醒的是一个同类线程（比如生产者唤醒的是另一个生产者），可能永远等不到条件满足。notifyAll 把等待线程全部唤醒，让它们自己再判断一轮，虽然开销大一点，但不会丢信号。

第三个：锁对象必须一致。produce 和 consume 都用 synchronized 锁了 this，如果有一处不小心锁了另一个对象，生产者和消费者各自的锁就完全隔离开了，整体就是废的。

1.3 为什么很多项目写着写着变成了“伪队列”

这个标题里的“伪队列”是我自己起的名字，指的是看起来用了 BlockingQueue，实际上既没解决生产消费平衡，也没解决并行调度，只是把一个 List 换了个线程安全的壳。

我自己见过、也亲手写过几种典型问题。

队列容量不设上限。很多人直接new LinkedBlockingQueue<>()，默认容量是 Integer.MAX_VALUE，等于是无限队列。生产端偶尔抽风或者流量突增，队列就能吞掉几百万条消息，内存直接顶着走。这种代码在测试环境永远测不出问题，因为测试流量太小了，一上生产就 OOM。

消费失败就 continue。取了一条消息，process 方法抛了个异常，外层循环直接 continue，消息就永远没了。这在通知推送场景下意味着用户收不到短信，在订单场景下意味着直接丢单。

用线程池但只提交了一个任务。启动消费者线程池时写了个 for 循环，结果循环条件写错了只 submit 了一次，后面的人看代码完全没察觉，只能通过日志里消费者的编号永远是 1 来发现。

这节说这么多不是为了吓人，而是想强调：生产者-消费者模式本身不难，难的是把它写成一个可以长期迭代、能排查问题、性能不拖后腿的工程组件。后面的并行任务调度、注释优化，都是围绕这个目标展开的。

2. 并行任务调度：让多个消费者真正“并行”起来

2.1 单消费者瓶颈：一个工人干所有活

很多第一次优化这套链路的人，第一反应是“把队列换得更快一点”“把 LinkedBlockingQueue 换成 ConcurrentLinkedQueue”，但实际上瓶颈往往根本不在队列本身，而在你只开了一个消费线程。

我举个例子。假设一个批处理任务，每次从队列取 1000 条数据，写一次数据库耗时大约 200ms。如果你只启动了一个消费者，哪怕队列里已经堆了 100 万条任务，每秒钟能处理的也就是 5000 条，吞吐完全被单线程锁死。而内存里明明有 8 核 CPU，一个线程只能占满一个核，其他核全在空转。

单消费者模型还有一个隐性缺点：如果消费者在处理一条任务时偶发慢调用，比如数据库连接池满了，等待 30 秒，那么整个消费链路都被卡住，队列不断堆积。虽然多消费者也会遇到慢调用，但至少其余消费者还能继续干活，系统不会“单点停滞”。

所以并行任务调度要解决的第一件事就是：把“一个消费者”变成“一组消费者”，让它们并行地消费同一个队列。

2.2 多消费者线程池设计与参数选择

并行消费最直接的实现方式，是用一个线程池来跑多个消费循环。这里我直接给出一个在生产环境稳定运行过的配置思路，而不是甩给你一串神秘参数。

int processors = Runtime.getRuntime().availableProcessors(); int consumers = Math.max(2, processors); // 保守一点，至少 2 个，最多不超过核数太多 ThreadPoolExecutor consumerPool = new ThreadPoolExecutor( consumers, consumers, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(consumers), // 工作队列只存任务，不存业务数据 new NamedThreadFactory("batch-consumer"), new ThreadPoolExecutor.CallerRunsPolicy() );

这里有几个关键点，逐一解释。

核心线程数和最大线程数设成一样，都是 consumers。消费循环是常驻任务，不像普通请求那样有高峰低谷，所以不要搞“核心 2 个、最大 20 个”这种弹性配置。最大线程数只在核心线程不够用时临时扩容，但消费循环本身是无限阻塞的，扩容上来的线程出来后立刻又去队列里阻塞等待，反而可能造成资源浪费。固定成一样，行为更可预测。

工作队列的大小设成 consumers，而不是设成业务队列的大小。这里很多人的误区是把线程池的队列当成业务队列用，实际上消费线程池的工作队列只是存放“消费循环任务”的，任务数量很少，不需要留大空间。

拒绝策略选 CallerRunsPolicy。如果消费者线程全部挂掉或者提交任务失败，CallerRunsPolicy 会直接在提交线程（也就是启动消费者的主线程）上继续执行，保证任务不会无声无息地丢。相比默认的 AbortPolicy 直接抛 RejectedExecutionException，这更安全。

线程工厂必须自定义命名。用过Executors.defaultThreadFactory()的人都知道，线程名一堆 “pool-2-thread-1”，出了问题你连是哪个池的线程都看不出来。用 NamedThreadFactory 可以统一命名成batch-consumer-1、batch-consumer-2，日志排查方便非常多。

线程数的估算，我一般用这个公式作为起点：N = CPU 核数 * (1 + 等待时间 / 计算时间)。如果任务是纯 IO 型，比如批量写库、调第三方接口，等待时间远大于计算时间，线程数可以放宽到核数的几倍甚至几十倍。如果是 CPU 密集型的解析、加密等任务，线程数接近核数就行了，开太多反而因为线程切换拖慢速度。

实际场景里，我曾经把一个单消费者的批量订单处理改成 8 个消费者并发处理，订单表从每秒 5000 条提升到 35000 条左右，数据库本身成了瓶颈，但吞吐的提升是肉眼可见的。这也说明，很多并发优化其实不需要复杂的算法，先把单消费者变成多消费者，效果就立竿见影。

2.3 批量合并提交：减少锁竞争，提升吞吐

当你已经开了多消费者，下一步值得做的优化是批量合并提交。这里的“批量”不是指消费者一次只取一条消息，而是指攒够一批再处理，尤其适合写库、推送等 IO 型操作。

我写一个实际可用的消费循环模板，注意看 drainTo 的用法：

public void consumeLoop() { while (!Thread.currentThread().isInterrupted()) { try { List<Task> batch = new ArrayList<>(BATCH_SIZE); Task first = pendingQueue.poll(POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (first == null) { continue; // 超时没有任务，继续下一次循环 } batch.add(first); pendingQueue.drainTo(batch, BATCH_SIZE - 1); // 尽可能多取一些 processBatch(batch); // 批量处理 } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { // 生产环境这里必须做失败隔离、重试、死信等兜底 log.error("consume batch failed", e); } } }

为什么用 poll 而不是 take？take 会无限阻塞直到有数据，如果消费端需要优雅关闭，线程会因为无法响应中断而卡住。poll 带超时，超时后循环体有机会检查线程中断状态并退出，这是顺手解决了一个很隐晦的停机问题。

为什么用 drainTo 而不是循环里逐个 take？逐个拿一条处理一条，每次都要竞争队列的锁。drainTo 一次性把当前队列里最多 N 条拿出来，只需要竞争一次锁，批量处理时锁开销大幅降低。之前压测过，批量 100 条和逐条处理相比，整体吞吐能提升 30% 到 50%，队列越大效果越明显。

还有一个不能忽略的细节：batch 的最后一批数据要能及时刷出去，不能等积累到 BATCH_SIZE 才处理。上面的代码里，poll 的超时起到了“兜底”的作用，即使一直没凑够一批，超时后也会把已有的几条拿出去处理，避免数据无限积压。

2.4 虚拟线程：JDK21 下的简洁并行方案

JDK21 正式发布虚拟线程之后，并行任务调度的代码可以大幅简化。虚拟线程最大的特点是“一个任务一个线程”，线程本身非常轻量，不再需要手动设计“线程池 + 消费者循环”这种结构。

你只需要这样写：

ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); while (!executor.isShutdown()) { Task task = pendingTaskQueue.poll(500, TimeUnit.MILLISECONDS); if (task == null) { continue; } executor.submit(() -> process(task)); // 每个任务一个虚拟线程 }

每个任务来了，虚拟线程池就创建一个虚拟线程去执行，任务执行完虚拟线程自动销毁。因为没有平台线程的 1:1 映射，百万级并发线程也不会把内存打爆。这个模型天然支持大批量阻塞 IO 任务，代码比手动维护线程池简单很多。

但虚拟线程不是万能药。我踩过的几个边界得提前说清楚。

CPU 密集型任务不适合虚拟线程。如果任务是纯计算，比如 JSON 解析、加解密、图像缩放，CPU 核数有限，虚拟线程照样排队等 CPU，优势体现不出来，反而因为调度开销增加一点性能损失。

synchronized 锁钉住虚拟线程的问题。在虚拟线程在 synchronized 块内阻塞时，会占住底层的平台线程，导致平台线程被“钉住”，大量这种操作会耗尽载体线程。虽然 JDK24 对 synchronized 做了改进，但如果你还在跑 JDK21/22，遇到阻塞 IO 的场景最好用 ReentrantLock 而不是 synchronized。

虚拟线程适合“大量任务 + 偶发阻塞”的场景，并不适合“少量常驻任务 + 高频切换”。对一个已经跑稳定的多消费者线程池，没有必要为了用虚拟线程而重写，量力而行。

3. 代码注释的“做得少”与“说清楚”

项目的标题里特别提到“更简洁的注释和每项改进的详细解释”，这其实是我在这个项目里最有感触的部分。很多人对注释的理解停留在“每行代码都写注释”，结果是代码上贴满了废话，真正需要说明的决策理由却一个字没有。

3.1 差注释长什么样：逐行翻译式、废话式

我见过太多这种注释了：

// 获取队列中的数据 Task task = queue.take(); // 有数据就取，没有就一直等 // 处理任务 process(task); // 判断是否成功 if (task.isSuccess()) { // 记录日志 log.info("task success"); }

这段注释的唯一作用就是占行数。queue.take()这个方法名已经说明了它在做什么，读者要的不是“它做了什么”，而是“为什么在这里用 take 而不是 poll”“为什么没有设置超时”“队列空的时候会怎样”。

还有一种自我陶醉式的注释，每个类都要写作者、创建时间、修改人：

/** * @author zhangsan * @date 2023-05-20 * @version 1.0 */

这类信息在 Git 提交记录里本来就有，写在代码里除了增加维护负担，没有任何价值。如果哪天代码被改了几十轮，作者名字还挂在那里，新人以为出错可以找这个人，非常误导。

3.2 用代码自解释替代注释

删注释容易，但删了之后代码必须自己说话。我总结了一套“注释精简三板斧”。

第一，命名要具体。queue改成pendingTaskQueue，process改成dispatchTask，data改成orderEvent。命名精确之后，一半注释都可以删掉。

第二，消灭魔法数。poll(500)里的 500 是什么？是超时毫秒数。声明成常量后，这个数字就有了语义。

private static final long POLL_TIMEOUT_MS = 500L;

第三，把复杂条件提取成方法。代码里的if (error != null && error.retryCount < 3 && !isShutdown)很难懂，但提取成一个方法之后：

if (canRetry(error)) { ... }

方法的命名本身就解释了这段判断的意图，不需要再写注释。

这三板斧执行完之后，你会惊喜地发现代码里还剩的注释，几乎都是真正值得写的内容。

3.3 关键注释才值得写：原因型注释、并发约定、参数范围

留下来的注释应该长什么样？核心就一句话：注释只回答“为什么不按常规来”和“这里有什么约束”。

拿前面的消费循环举例，我最终会在代码里保留这几条原因型注释：

// 必须用带超时的poll，否则优雅停机时线程无法响应中断，永远卡在take上 Task first = pendingQueue.poll(POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS); // drainTo一次拿走批量任务，减少锁竞争；不能逐个take，否则性能至少下降30% pendingQueue.drainTo(batch, BATCH_SIZE - 1);

这两条注释都是在读者可能产生疑问的地方提前解释，而不是解释代码本身。看到这些注释的人，不用重新踩一遍坑就能知道为什么这么写。

并发场景下，有几类注释其实和代码逻辑同样重要：不可变约束、可见性约定、锁顺序约定。比如：

// 该集合只能由consumer线程修改，其他线程只读，所以不需要同步 private final Map<String, Task> cache = new ConcurrentHashMap<>();

这种注释解释了“为什么不需要同步”“谁来保证线程安全”，对于维护并发代码的人至关重要。我自己见过太多因为没有写这类约束，后来人看到集合就以为不安全，亲手加上了一段性能很差的全局锁。

参数范围注释也挺值钱。比如队列容量为什么是 2048，可以在常量旁边写明计算依据：

// 容量 = 峰值生产速率(500/s) * 单次消费延迟(2s) + 80%缓冲区余量 ≈ 1800，取整到2048 private static final int QUEUE_CAPACITY = 2048;

读者一看就知道改参数时从哪里入手，而不是拍脑袋调一个 99999。

3.4 注释模板的坑

现在 IDE 都很智能，自动生成注释模板特别方便。我见过有人对每个方法都自动生成 Javadoc，里面全是@param xxx 参数xxx，@return 返回值，有的还带@deprecated但没写替代方案。

这类模板注释的危险在于：它制造了一种“这个类很完整很规范”的假象，实际上读者得不到任何有效信息。真正要读代码的人，只能忽略这些模板，逐行看代码。注释越多，噪音越大，真话越容易被淹没。

我的习惯是：方法的注释只在这段逻辑有协作时序、并发约束、事务边界时才写。字段注释只在字段含义容易被误解时才写。单行注释只用于解释“反直觉”的决策。

换句话说，注释的目标不是让代码显得很多，而是让代码显得很少。如果一段代码本身已经够简洁、命名够准确，不做注释是完全正常的。

4. 实例复盘：从单消费者基础版到高性能并行版

空讲理论没意思，我把一个简化版的实例完整贴出来。这是从十几万行生产代码里抽出来的模板，去掉业务细节之后大概长这样，但核心思路和注释方式都是实际用过的。

4.1 第一版：synchronized 基础版

public class BasicPipeline { private final Queue<String> queue = new LinkedList<>(); private static final int CAPACITY = 1000; public synchronized void produce(String item) throws InterruptedException { while (queue.size() == CAPACITY) { wait(); } queue.offer(item); notifyAll(); } public synchronized String consume() throws InterruptedException { while (queue.isEmpty()) { wait(); } String item = queue.poll(); notifyAll(); return item; } }

这一版适合学习原理，但生产环境我不会直接用。原因：整个队列只有一个锁，生产者、消费者完全串行互斥；队列用 LinkedList，没有容量上限保护（这里虽然设了判断，但如果有多个生产者并发判断，size 判断并不是原子的）；没有超时机制，线程可能无限期阻塞。它最大的问题是没有发挥多核并行能力。

4.2 第二版：BlockingQueue + 线程池批量并行版

public class BatchParallelPipeline { private final BlockingQueue<Task> pendingTaskQueue = new LinkedBlockingQueue<>(QUEUE_CAPACITY); private final ExecutorService consumerPool; public BatchParallelPipeline(int consumerCount) { consumerPool = new ThreadPoolExecutor( consumerCount, consumerCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(consumerCount), new NamedThreadFactory("batch-consumer"), new ThreadPoolExecutor.CallerRunsPolicy() ); } public void start() { for (int i = 0; i < consumerCount; i++) { consumerPool.submit(this::consumeLoop); } } private void consumeLoop() { while (!Thread.currentThread().isInterrupted()) { try { List<Task> batch = new ArrayList<>(BATCH_SIZE); Task first = pendingTaskQueue.poll(POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (first == null) { continue; // 空闲时间不做无意义循环 } batch.add(first); pendingTaskQueue.drainTo(batch, BATCH_SIZE - 1); processBatch(batch); // 这里做业务处理 } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { log.error("consume batch failed, batchSize=" + batch.size(), e); // 生产环境必须有死信或重试策略 } } } public void shutdown() { consumerPool.shutdown(); // 先停止接收新消费任务 try { consumerPool.awaitTermination(30, TimeUnit.SECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }

这版就是项目里真正能够上生产的骨架。多消费者并行消费，吞吐量成倍增加。使用 BlockingQueue 之后，不再需要手写 wait/notify，线程安全性由队列内部保证。带超时的 poll 加上 drainTo 批量取出，既解决了停机问题，又降低了锁竞争。

有一点要注意，第二版的 processBatch 里如果抛出异常，不能直接打死消费循环，否则这个消费者线程就退出不干活了。生产环境我的做法是：记录异常，把失败批次写进一个死信队列，给重试线程去处理。虽然会增加一点复杂度，但数据不丢才是最底线的事。

4.3 第三版：虚拟线程简化版

JDK21 及以上环境，可以考虑虚拟线程方案。它省去了手动管理线程池的环节，整个消费模型变得更加直接。

public void startWithVirtualThreads() { ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); int processors = Runtime.getRuntime().availableProcessors(); for (int i = 0; i < processors; i++) { executor.submit(this::consumeLoopV2); } } private void consumeLoopV2() { while (!Thread.currentThread().isInterrupted()) { try { Task task = pendingTaskQueue.poll(POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS); if (task != null) { process(task); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } }

这个方案有一个明显的变化：不再需要 drainTo 批量合并了。因为虚拟线程足够轻量，为每个任务创建一个虚拟线程的开销比平台线程小得多，所以你可以退回到更简单的单条处理模型。实测下来，在 IO 密集型任务上，简化版和批量版的吞吐差距可以接受，但代码读起来通顺很多。

这个方案最值得警惕的还是 CPU 密集型任务。虚拟线程的调度靠 JVM，底层平台线程数量有限，一旦所有虚拟线程都在做 CPU 计算而不是阻塞等待，载入过重后整体性能反而下降。用它之前先确认你的任务是不是真的以阻塞 IO 为主。

4.4 注释优化前后对比

最后用一个具体的注释优化对比收束这节。这是我从一个真实项目里摘出来的重构前后对照。

重构前：

// 取出一个任务 Task task = queue.take(); // 执行任务 execute(task); // 检查结果，如果失败就重试 if (!task.isSuccess()) { retry(task); }

重构后：

Task task = pendingTaskQueue.poll(POLL_TIMEOUT_MS, TimeUnit.MILLISECONDS); // 带超时poll避免队列空时线程永远阻塞，无法优雅退出；task==null说明队列暂时为空，直接进入下一轮 if (task == null) { continue; } dispatch(task); // dispatch内部已做失败分类，只有可重试异常才走重试队列

重构后的代码第一眼看上去好像“注释变少了”，实际上有效信息变多了。读者一下就知道：为什么用 poll 而不用 take、什么情况算暂时为空、重试逻辑被封装在哪里。

这种注释风格的核心逻辑是：把代码里每一个“容易让人困惑的决策”变成明面上的解释，而不是逐行翻译代码。注释的价值密度高了很多。

5. 常见问题与排查技巧实录

最后这部分，全是这些年实际踩坑后才总结出来的内容。建议收藏，出了问题来对照着看。

5.1 死锁：程序卡住，日志还打着，就是不动

典型症状：队列消费速率降为 0，线程池看着还有线程存活，但日志不再输出。

排查方式：第一时间打 jstack 抓线程栈，看线程处于什么状态、卡在哪一行。

常见原因：

• 用了 synchronized 包住 BlockingQueue 的 take，但唤醒条件不对，互相等待
• 多个任务持有多个锁，嵌套获取，两个线程互相等对方释放
• 消费任务里又调用了生产者端的方法，循环等待

我的经验：死锁问题 90% 是锁的嵌套获取导致的，另外 10% 是 wait/notify 信号丢失。设计上尽量避免在持有锁的情况下再获取其他锁；非要嵌套时，必须保证全局锁顺序一致。

5.2 队列满了或者空了，处理策略是什么

队列满的时候，生产者会阻塞，这是 BlockingQueue 的默认行为。阻塞的好处是背压自然传递到上游，坏处是如果一直满，会拖慢整个生产链路。

我通常给生产者的写入方法加上超时：

boolean offered = pendingTaskQueue.offer(task, 1, TimeUnit.SECONDS); if (!offered) { // 队列已满，可以选择走降级策略：丢弃、落本地文件、或者同步处理 }

队列空的时候，消费端循环需要避免空转。poll 超时设为 50ms 到 500ms 比较合适，太短会空转浪费 CPU，太长会降低延迟敏感度。如果线程需要优雅退出，poll 超时给你一个定期检查中断状态的机会。

5.3 数据重复消费或丢失

这是并行消费最怕的两件事。

数据丢失常见原因：消费线程取出了任务，process 抛异常，外层直接 catch 吞掉；或者 process 先删了源数据，然后自己执行失败，数据没了。

数据重复常见原因：消费成功后，ack 确认失败，消息重新投递；或者 process 执行到一半，消费者宕机，任务被重新消费。

应对方案只有两层：第一层，消费端必须做幂等，用唯一业务键去重；第二层，处理失败的消息必须进入死信队列，不能直接丢弃。对于“先处理还是先确认”，我的习惯是先持久化处理结果，再确认消费，这样即使确认失败也只是重复处理，不会丢。

5.4 线程池线程耗尽：任务全排队，没线程干活

这个问题常见于把线程池用于“异步执行”业务任务时。业务任务本身会去调慢接口、等待锁，一个任务卡住，线程池里的线程全被占住，新的任务排在工作队列里，但永远轮不到执行。

排查方式：

• 用 ThreadPoolExecutor 提供的 getActiveCount/getQueue 大小做监控
• 出现活跃线程数长时间等于最大线程数，就要警惕了

解决方式：

• 把任务按类型拆分成多个独立线程池，互相不拖累
• 线程池中的任务尽量设置超时，避免永久阻塞
• 情况允许时，考虑用虚拟线程池替代，让阻塞任务不再占用线程

5.5 性能压测与调优的一点实操经验

最后聊一下怎么验证你的优化真的有效。

我习惯的做法是写一个小的压测入口，模拟生产者以不同的速率注入任务，然后统计消费延迟和吞吐：

long start = System.nanoTime(); // 注入100万条任务 pipeline.produceBatch(1_000_000); long end = System.nanoTime(); System.out.println("throughput: " + 1_000_000L * 1_000_000_000 / (end - start) + " tasks/s");

注意一定要测 P99 延迟，而不是只测平均延迟。并发场景下，平均延迟往往被大多数快的任务拉低，真正影响用户体验的是那些卡在尾部的最慢任务。以前我优化完只看平均延迟，然后上线后还是被用户投诉，后来加了 P99 监控才定位到是某类大任务偶尔耗时特别长，把线程池占满了。

调优的顺序应该是：先确认瓶颈在 CPU、IO 还是锁竞争上，再动手改参数。实践里很多问题不是队列太小，而是消费逻辑里有慢查询；也不是线程数不够，而是线程被无意义的自旋浪费了。性能调优最忌讳一上来就盲目改并发数，连监控数据都没看，改了一天，方向全错了。

我个人的体会是，并发编程里 70% 的价值来自于把模型划分正确，剩下的 30% 才来自参数调优。生产者-消费者模式是模型，线程池和虚拟线程是模型落地的手段，而注释就是你留给下一个维护者的使用说明书。每次重构完一套并发代码，如果能顺手记录这次改动的决策理由，后面排查问题的成本会低很多。这个习惯坚持一年，你大概率会回来感谢自己。