手写Master-Worker并发框架：线程安全设计与实战避坑

1. 从一次线上事故说起：为什么要自己造一个Master-Worker轮子

去年帮一个做数据清洗的朋友排查问题，他们的服务在高峰期会偶发数据错乱——同一批任务被重复执行了两次，日志里却找不到任何异常。最后定位到根因：他们用线程池提交任务时，共享了一个非线程安全的累加器来统计进度，多个Worker线程同时读写导致计数漂移，进而触发了任务调度的重复分发逻辑。

这个坑其实非常典型。很多团队在做并发任务处理时，第一反应是直接上ThreadPoolExecutor或者消息队列，觉得“线程池本身就是线程安全的，我只要把任务丢进去就行了”。但真正的问题往往不在线程池本身，而在于任务分发、状态汇总、结果收集这三个环节中，你自定义的那些共享数据结构是否扛得住并发。

这就是我想聊的Master-Worker并发组件设计模拟。它不是让你去重新实现一个JDK自带的线程池，而是让你理解：当你需要把一个大任务拆成若干子任务、分发给多个Worker并行执行、再汇总结果时，Master节点和Worker节点之间的协作机制该怎么设计，哪些地方会出线程安全问题，以及怎么用最朴素的方式把这些坑填上。

这篇文章适合两类人看：一类是正在准备面试、被“手写一个Master-Worker模型”这类题目卡住的开发者；另一类是工作中确实需要做批量并行处理、但对并发细节心里没底的后端或数据工程师。我会从整体设计思路讲到具体代码实现，再到实际踩过的坑，尽量把每个决策背后的“为什么”说清楚。

2. 整体设计思路：Master-Worker到底在解决什么问题

2.1 核心模型拆解：谁负责分发，谁负责执行

Master-Worker模式的本质是一种任务分解与并行归约的架构。你可以把它想象成一个包工头带一群工人干活：包工头（Master）负责把一个大工程拆成若干小活，分配给工人（Worker），工人干完活把结果交回来，包工头最后汇总。

在这个模型里，有几个核心角色和数据结构：

• Master节点：负责接收原始任务、拆分子任务、将子任务放入任务队列、启动并管理Worker线程、收集Worker的处理结果、最终汇总输出。
• Worker节点：每个Worker是一个独立线程，循环从任务队列中领取子任务，执行具体的业务逻辑，然后把结果写入结果集。
• 任务队列：Master往里放，Worker从里取。这个队列是并发访问的核心热点，必须线程安全。
• 结果集：多个Worker同时往里写，同样必须线程安全。
• 任务对象：每个子任务需要携带自己的标识和参数，Worker执行完后要能追溯到是哪个任务产生了什么结果。

为什么不用简单的parallelStream或者直接提交给线程池？因为Master-Worker模式给了你更细粒度的控制权：你可以控制任务拆分的粒度、可以自定义任务队列的容量和阻塞策略、可以在Master层做任务优先级调度、可以在Worker层做失败重试。这些灵活性在复杂的批处理场景里非常关键。

2.2 为什么线程安全是绕不开的坎

在这个模型里，线程安全问题集中在三个地方：

第一，任务队列的并发读写。Master线程往队列里put任务，多个Worker线程同时从队列里take任务。如果用的是普通的LinkedList或ArrayList，必然出现索引越界、数据丢失、死循环等问题。必须用BlockingQueue的实现类，比如LinkedBlockingQueue或ArrayBlockingQueue。

第二，结果集的并发写入。多个Worker执行完任务后，需要把结果放到一个共享的容器里。如果用HashMap，在并发put的时候可能触发扩容导致链表成环，CPU飙到100%。必须用ConcurrentHashMap或者给结果集加锁。

第三，任务状态的可见性。Master需要知道所有任务是否都已完成，Worker需要知道是否还有新任务要处理。这些状态变量如果不用volatile或原子类修饰，一个线程的修改可能永远不被另一个线程看到。

我见过太多人在这三个地方翻车。下面我会逐一拆解每个环节的设计考量和实现细节。

2.3 方案选型：为什么用BlockingQueue而不是自己写wait/notify

有人可能会想：我用synchronized加wait/notify自己实现一个队列不就行了？理论上可以，但实际项目中强烈不建议。原因有三：

• 正确性难以保证：wait/notify的虚假唤醒问题、通知丢失问题、锁的粒度控制，每一个都是坑。JDK的BlockingQueue已经帮你处理了所有这些边界情况。
• 性能优化：LinkedBlockingQueue内部用了两把锁（入队锁和出队锁分离），ArrayBlockingQueue用了一把锁但配合条件队列，这些优化是自己手写很难达到的。
• 可维护性：标准库的API语义清晰，后来人接手代码时不需要重新理解你的自定义同步逻辑。

所以我的建议很明确：任务队列直接用LinkedBlockingQueue，结果集直接用ConcurrentHashMap，状态标志用AtomicBoolean或volatile。把精力放在任务拆分策略和结果汇总逻辑上，而不是重复造同步轮子。

3. 核心细节解析：每个环节的线程安全设计要点

3.1 任务队列的选型与容量设计

LinkedBlockingQueue和ArrayBlockingQueue是最常用的两个实现。怎么选？

LinkedBlockingQueue基于链表，默认容量是Integer.MAX_VALUE，相当于无界队列。它的入队和出队用了两把不同的锁，并发吞吐量更高。但无界队列有个致命风险：如果Worker处理速度跟不上Master的生产速度，队列会无限膨胀，最终OOM。

ArrayBlockingQueue基于数组，必须在构造时指定容量，是有界队列。它只用一把锁控制入队和出队，吞吐量略低，但内存可控，能天然形成背压——当队列满时，Master的put操作会阻塞，直到有Worker取走任务。

我的实操建议是：生产环境优先用有界队列。容量怎么定？一个经验公式是：队列容量 = Worker数量 × 单任务平均处理时间 / 任务提交间隔。比如你有8个Worker，每个任务平均处理200ms，Master每50ms提交一个任务，那么队列容量至少要是8 × 200 / 50 = 32。实际设置时可以在这个基础上乘以1.5到2倍作为缓冲。

// 推荐：有界队列，容量根据实际压测调整 BlockingQueue<Task> taskQueue = new ArrayBlockingQueue<>(64); // 如果确实需要高吞吐且能保证生产消费平衡，可以用LinkedBlockingQueue指定容量 BlockingQueue<Task> taskQueue = new LinkedBlockingQueue<>(128);

注意：千万不要用new LinkedBlockingQueue<>()不传容量，那个默认值是Integer.MAX_VALUE，等于埋了一颗OOM的雷。

3.2 结果集的线程安全方案对比

结果集的写入是另一个并发热点。假设每个Worker处理完任务后要把结果放到一个Map里，key是任务ID，value是处理结果。可选方案有：

我一般直接用ConcurrentHashMap。它在JDK8之后改成了CAS + synchronized的方式，锁粒度细化到单个桶，并发写入性能很好。如果结果集需要保证顺序，可以用ConcurrentSkipListMap，它基于跳表实现，天然有序。

// 结果集：任务ID -> 处理结果 ConcurrentHashMap<String, Result> resultMap = new ConcurrentHashMap<>(); // 如果需要有序结果 ConcurrentSkipListMap<String, Result> orderedResultMap = new ConcurrentSkipListMap<>();

还有一种更极致的做法：每个Worker维护自己的局部结果列表，等所有Worker执行完毕后，由Master统一合并。这样做的好处是Worker之间完全没有竞争，缺点是内存占用会翻倍，而且合并阶段可能成为瓶颈。适合Worker数量不多（比如不超过16个）且结果集较大的场景。

3.3 任务对象的不可变性设计

任务对象本身应该是不可变对象。什么意思？就是任务一旦创建，它的所有字段都不能再被修改。这样做的好处是：多个Worker即使同时读取同一个任务对象，也不会有任何线程安全问题，因为没有人能改它。

public final class Task { private final String taskId; private final String data; private final int priority; public Task(String taskId, String data, int priority) { this.taskId = taskId; this.data = data; this.priority = priority; } // 只提供getter，不提供setter public String getTaskId() { return taskId; } public String getData() { return data; } public int getPriority() { return priority; } }

如果任务对象确实需要可变状态（比如记录重试次数），那这个可变字段必须用原子类或者加锁保护。但更好的做法是：把可变状态抽出来放到一个单独的TaskContext对象里，任务本身保持不可变。

3.4 Worker的优雅停止机制

Worker线程不能是死循环一直跑，必须有一个优雅停止的机制。常见做法有两种：

第一种是毒丸对象（Poison Pill）。Master在任务全部提交完毕后，往队列里放入与Worker数量相同的特殊任务对象，Worker取到这个对象后就退出循环。这种方式的优点是简单直观，缺点是如果队列里还有未处理的任务，毒丸会被排在后面，Worker会先把前面的任务处理完才退出。

第二种是中断标志 + 超时轮询。用一个AtomicBoolean作为停止标志，Worker在循环中检查这个标志，同时用poll(timeout, unit)而不是take()来获取任务，这样即使没有新任务，Worker也能定期醒来检查停止标志。

public class Worker extends Thread { private final BlockingQueue<Task> taskQueue; private final ConcurrentHashMap<String, Result> resultMap; private final AtomicBoolean running; public Worker(BlockingQueue<Task> taskQueue, ConcurrentHashMap<String, Result> resultMap, AtomicBoolean running) { this.taskQueue = taskQueue; this.resultMap = resultMap; this.running = running; } @Override public void run() { while (running.get()) { try { Task task = taskQueue.poll(100, TimeUnit.MILLISECONDS); if (task == null) { continue; // 超时未取到任务，回去检查running标志 } Result result = process(task); resultMap.put(task.getTaskId(), result); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } private Result process(Task task) { // 具体的业务处理逻辑 return new Result(task.getTaskId(), "processed"); } }

我倾向于第二种方案，因为它对中断响应更及时，而且不需要关心毒丸的投放数量是否精确匹配。

4. 完整实操：从零搭建一个可运行的Master-Worker框架

4.1 整体类结构设计

先把整个框架的类图理清楚。我们需要以下几个核心类：

• Task：任务对象，不可变。
• Result：结果对象，不可变。
• Master：核心调度器，负责拆分任务、启动Worker、提交任务、等待完成、汇总结果。
• Worker：工作线程，从队列取任务、执行、写结果。
• MasterWorkerDemo：测试入口，模拟一批任务并验证结果。

这个结构足够简单，没有过度设计，但覆盖了Master-Worker模式的所有关键要素。

4.2 Master节点的实现细节

Master的职责比较重，我们一步步来。

首先是构造方法，需要指定Worker数量和任务队列容量：

public class Master { private final int workerCount; private final BlockingQueue<Task> taskQueue; private final ConcurrentHashMap<String, Result> resultMap; private final List<Worker> workers; private final AtomicBoolean running; public Master(int workerCount, int queueCapacity) { this.workerCount = workerCount; this.taskQueue = new ArrayBlockingQueue<>(queueCapacity); this.resultMap = new ConcurrentHashMap<>(); this.workers = new ArrayList<>(workerCount); this.running = new AtomicBoolean(false); }

然后是启动方法，创建并启动所有Worker线程：

public void start() { running.set(true); for (int i = 0; i < workerCount; i++) { Worker worker = new Worker(taskQueue, resultMap, running); worker.setName("worker-" + i); workers.add(worker); worker.start(); } }

提交任务的方法，这里要注意：如果队列满了，put会阻塞，这是背压机制在起作用：

public void submit(Task task) throws InterruptedException { if (!running.get()) { throw new IllegalStateException("Master is not running"); } taskQueue.put(task); }

等待所有任务完成的方法，这里用了一个简单的轮询检查：

public void waitForCompletion() throws InterruptedException { // 先等待队列排空 while (!taskQueue.isEmpty()) { Thread.sleep(50); } // 再等待所有Worker处理完手头的任务 // 这里用一个简单的方式：等待所有Worker线程结束 // 实际项目中可以用CountDownLatch或Phaser做更精确的控制 Thread.sleep(200); }

等等，这里有个问题：taskQueue.isEmpty()返回true并不代表所有任务都处理完了，因为可能还有任务正在Worker手里执行。更严谨的做法是用一个AtomicInteger来统计已提交和已完成的任务数：

private final AtomicInteger submittedCount = new AtomicInteger(0); private final AtomicInteger completedCount = new AtomicInteger(0); public void submit(Task task) throws InterruptedException { taskQueue.put(task); submittedCount.incrementAndGet(); } // Worker处理完后调用 public void onTaskCompleted() { completedCount.incrementAndGet(); } public void waitForCompletion() throws InterruptedException { while (completedCount.get() < submittedCount.get()) { Thread.sleep(50); } }

这样就能精确知道所有任务是否都已完成。AtomicInteger的incrementAndGet是CAS操作，线程安全且性能很好。

最后是停止方法和结果获取方法：

public void stop() throws InterruptedException { running.set(false); for (Worker worker : workers) { worker.join(1000); } } public Map<String, Result> getResults() { return Collections.unmodifiableMap(resultMap); }

4.3 Worker线程的执行逻辑与异常处理

Worker的run方法前面已经给过基本框架，这里补充异常处理的细节。Worker在执行任务时可能会抛出各种异常，不能让一个任务的异常导致整个Worker线程死掉：

@Override public void run() { while (running.get()) { Task task = null; try { task = taskQueue.poll(100, TimeUnit.MILLISECONDS); if (task == null) { continue; } Result result = process(task); resultMap.put(task.getTaskId(), result); master.onTaskCompleted(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { // 单个任务失败不影响Worker继续工作 if (task != null) { resultMap.put(task.getTaskId(), new Result(task.getTaskId(), "ERROR: " + e.getMessage())); master.onTaskCompleted(); } // 记录日志 System.err.println("Worker " + getName() + " failed on task: " + e.getMessage()); } } }

这里有个关键点：异常必须被捕获并记录到结果集中，否则Master的waitForCompletion会永远等下去，因为completedCount永远不会增加到submittedCount。

4.4 测试验证：模拟100个任务、8个Worker

写一个简单的测试来验证整个框架：

public class MasterWorkerDemo { public static void main(String[] args) throws Exception { Master master = new Master(8, 32); master.start(); int taskCount = 100; for (int i = 0; i < taskCount; i++) { master.submit(new Task("task-" + i, "data-" + i, i % 3)); } long start = System.currentTimeMillis(); master.waitForCompletion(); long elapsed = System.currentTimeMillis() - start; Map<String, Result> results = master.getResults(); System.out.println("Total tasks: " + taskCount); System.out.println("Completed: " + results.size()); System.out.println("Elapsed: " + elapsed + "ms"); // 验证结果完整性 for (int i = 0; i < taskCount; i++) { String taskId = "task-" + i; if (!results.containsKey(taskId)) { System.err.println("Missing result for: " + taskId); } } master.stop(); } }

跑下来应该能看到100个任务全部完成，结果集大小正好是100。如果出现结果数少于100，说明有任务丢失，需要检查队列容量和Worker的异常处理逻辑。

4.5 性能压测：不同Worker数量下的吞吐对比

我实际跑了一组压测数据，任务总数10000，每个任务模拟10ms的处理耗时，队列容量固定为64，调整Worker数量：

从数据可以看出，Worker数量不是越多越好。在8核机器上，8个Worker基本就能跑满CPU，再增加Worker数量反而会因为线程上下文切换导致性能下降。一般建议Worker数量设置为CPU核心数的1到2倍，具体要根据任务是CPU密集型还是IO密集型来调整。

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

5.1 任务丢失：为什么结果集里少了几个任务

这是最常见的问题。排查思路按以下顺序进行：

第一，检查Worker的异常处理。如果process方法抛了异常但没有被捕获，Worker线程会直接死掉，后续任务没人处理。用jstack看线程状态，如果Worker线程数量少于预期，基本就是这个原因。

第二，检查completedCount和submittedCount的递增逻辑。如果某个分支下onTaskCompleted没有被调用，Master会一直等下去。可以在waitForCompletion里加超时机制：

public boolean waitForCompletion(long timeout, TimeUnit unit) throws InterruptedException { long deadline = System.currentTimeMillis() + unit.toMillis(timeout); while (completedCount.get() < submittedCount.get()) { if (System.currentTimeMillis() > deadline) { return false; // 超时 } Thread.sleep(50); } return true; }

第三，检查队列容量。如果队列太小且Master提交速度远快于Worker处理速度，put会阻塞，但不会丢任务。真正会丢任务的情况是用了offer且没有处理返回值——offer在队列满时直接返回false，任务就被丢弃了。

5.2 死锁：Master等待Worker，Worker等待Master

有一种隐蔽的死锁场景：Master在提交任务时调用了taskQueue.put，而队列已满，Master阻塞；同时所有Worker都在执行任务，但任务执行过程中需要回调Master的某个同步方法，而那个方法被Master持有锁。这就形成了循环等待。

避免方法很简单：Master在持有锁的时候不要调用可能阻塞的操作。具体来说，submit方法不要加synchronized，put的阻塞是队列内部的锁，和Master对象的锁无关。另外，Worker的回调方法（如onTaskCompleted）只用原子类操作，不要加锁。

5.3 结果覆盖：相同taskId的任务互相覆盖

如果任务ID有重复，后完成的任务会覆盖先完成的结果。这在ConcurrentHashMap的put操作中是正常行为。解决办法是在put之前用putIfAbsent检查：

Result existing = resultMap.putIfAbsent(task.getTaskId(), result); if (existing != null) { // 记录重复任务告警 System.err.println("Duplicate taskId: " + task.getTaskId()); }

更好的做法是在任务拆分阶段就保证ID唯一，比如用UUID或者原始任务ID + 子任务序号的组合。

5.4 常见问题速查表

实操心得：在开发阶段，我习惯在Master的waitForCompletion里加一个每秒打印一次进度的日志，输出completedCount/submittedCount和队列剩余大小。这样一旦卡住，一眼就能看出是生产慢了还是消费慢了。

6. 进阶优化：从能跑到跑得好的几个关键改进

6.1 用CountDownLatch替代轮询等待

前面的waitForCompletion用的是Thread.sleep轮询，虽然简单但不够优雅，而且有延迟。更好的方式是用CountDownLatch：

private final CountDownLatch completionLatch; public Master(int workerCount, int queueCapacity) { // ... this.completionLatch = new CountDownLatch(1); // 只需要一个信号 } public void submit(Task task) throws InterruptedException { taskQueue.put(task); submittedCount.incrementAndGet(); } public void onTaskCompleted() { if (completedCount.incrementAndGet() == submittedCount.get()) { completionLatch.countDown(); } } public void waitForCompletion() throws InterruptedException { completionLatch.await(); }

注意这里submittedCount.get()的读取和completedCount.incrementAndGet()之间没有原子性保证，在极端情况下可能提前触发或延迟触发。更严谨的做法是用一个AtomicInteger同时维护提交和完成，或者用Phaser。但对于大多数场景，这个微小的时间窗口是可以接受的。

6.2 支持任务优先级调度

如果任务有优先级，可以把ArrayBlockingQueue换成PriorityBlockingQueue。但要注意：PriorityBlockingQueue是无界的，需要自己控制提交速度。另外，任务对象需要实现Comparable接口：

public final class Task implements Comparable<Task> { // ... 其他字段 private final int priority; @Override public int compareTo(Task other) { return Integer.compare(other.priority, this.priority); // 优先级高的排前面 } }

6.3 动态调整Worker数量

固定Worker数量在流量波动大的场景下不够灵活。可以做一个简单的动态扩缩容：当队列大小持续超过阈值时增加Worker，当队列持续为空时减少Worker。但要注意，线程的创建和销毁有开销，扩缩容的频率不能太高，一般设置一个最小间隔（比如30秒）。

这个特性实现起来比较复杂，我的建议是：如果队列积压是常态，直接一开始就设置足够的Worker数量；如果只是偶发峰值，用有界队列加背压让上游感知到压力，比动态扩缩容更简单可靠。

6.4 结果集的分批处理

如果任务量非常大（比如百万级），把所有结果都放在一个ConcurrentHashMap里会占用大量内存。这时候可以做一个批处理机制：每完成N个任务，就把结果集里已完成的部分取出来持久化或传给下游，然后从Map里移除。

private static final int BATCH_SIZE = 1000; private final AtomicInteger batchCounter = new AtomicInteger(0); public void onTaskCompleted() { int completed = completedCount.incrementAndGet(); if (batchCounter.incrementAndGet() >= BATCH_SIZE) { flushResults(); batchCounter.set(0); } if (completed == submittedCount.get()) { completionLatch.countDown(); } } private synchronized void flushResults() { // 把resultMap里的结果批量输出，然后清空 // 注意：这里加synchronized是为了防止多个Worker同时flush // 更好的做法是用一个专门的flush线程 }

这里flushResults加了synchronized，因为多个Worker可能同时触发flush。但加锁会影响性能，更好的做法是用一个BlockingQueue把完成的结果传给一个专门的输出线程，由那个线程负责批量写入。

7. 我在实际项目中的几点体会

这个Master-Worker框架我在两个项目里实际用过，一个是日志清洗，一个是报表数据预计算。踩过的坑和总结出来的经验大概有这么几条。

第一，不要过早优化。一开始就想着支持优先级、动态扩缩容、结果分批，代码会变得非常复杂。先用最简单的ArrayBlockingQueue+ConcurrentHashMap+ 固定Worker数量跑通，确认功能正确后再根据压测结果做针对性优化。我第一个版本就是直接上了PriorityBlockingQueue，结果发现任务优先级其实根本没用上，反而因为无界队列导致了一次内存告警。

第二，监控比功能更重要。Master-Worker模式一旦跑起来，你很难直观看到内部状态。一定要在关键路径上加监控：队列当前大小、已完成任务数、每个Worker的处理速率、任务平均耗时。这些指标能帮你在出问题时快速定位是生产端的问题还是消费端的问题。

第三，异常处理要区分对待。不是所有异常都应该被吞掉。如果是业务逻辑异常（比如数据格式不对），记录到结果集里继续处理下一个任务；如果是系统级异常（比如OOM），应该让Worker退出并通知Master停止整个流程。我一般会定义一个FatalException，只有这个异常才会触发全局停止。

第四，测试并发问题不能靠运气。写单元测试的时候，不要只跑一遍就认为没问题。用CountDownLatch模拟高并发场景，让多个线程同时提交任务、同时读取结果，跑个几百遍。我写过一个测试，循环1000次创建Master、提交100个任务、验证结果，跑了一晚上才复现出一个结果丢失的bug——原因是completedCount和submittedCount的读取顺序在某些时序下会出错。

这个框架后续还可以往几个方向扩展：支持任务超时中断、支持Worker级别的任务重试、支持结果集的流式输出。但核心的线程安全设计就是这些，把队列、结果集、状态标志这三个点守住了，整个框架就不会出大问题。