1. 从一次线上数据错乱说起为什么我要手写一个Master-Worker组件去年帮一个做数据清洗的朋友排查一个诡异问题他们的批处理服务在单机跑得好好的一上多线程就偶尔出现结果条数对不上日志里还夹杂着几条重复记录。代码翻出来一看典型的共享变量裸奔——多个Worker线程往同一个ArrayList里塞结果主线程边塞边读。这种问题在并发场景里太常见了常见到很多人第一次写多线程代码都会踩。这件事之后我就想与其每次都在业务代码里零散地加锁、加队列不如把任务分发—并行执行—结果归集这套模式抽象成一个可复用的组件。这就是这篇要聊的Master-Worker并发组件。它不是什么新概念本质是一种并行计算的设计模式一个Master负责拆分任务、调度Worker、收集结果多个Worker负责真正干活。听起来简单但要做到线程安全且性能不塌里面的坑比想象中多。这篇内容适合谁看如果你写过ThreadPoolExecutor但说不清submit和execute的区别如果你用过ConcurrentHashMap但没想过它为什么在扩容时会阻塞如果你正在设计一个需要并行处理大批量任务的模块——那这篇就是写给你的。我会从设计动机讲起把任务队列、结果归集、Worker生命周期、异常处理这几块拆开揉碎最后给一份可以直接抄的完整实现并附上我实测踩过的坑。需要先说明一点下面所有代码和参数都是基于常见工程实践给出的参考实现不是某个特定线上系统的原样复刻你可以根据自己的业务规模调整队列容量、线程数这些参数。2. 拆解Master-Worker模式它到底解决了什么又引入了什么2.1 一个Master加N个Worker职责怎么切先把角色说清楚。Master-Worker模式里有两个核心角色Master负责接收原始任务、把大任务拆成小任务、把任务投递到共享队列、启动并管理Worker、最后从结果容器里汇总数据。Worker从共享队列里取任务、执行任务、把结果写回结果容器。Worker本身不关心任务从哪来、结果去哪它只负责取—算—放。这个分工的好处是解耦。任务的拆分逻辑和任务的执行逻辑分开Master可以专注于怎么拆得均匀Worker可以专注于怎么算得快。你甚至可以让不同类型的Worker处理不同类型的任务只要它们都从同一个队列取活。但解耦也带来了新问题Master和Worker之间靠什么通信答案是共享的阻塞队列和线程安全的结果容器。这两个东西就是整个组件的命脉设计不好要么死锁要么数据错乱要么性能还不如单线程。2.2 为什么不用现成的线程池非要自己搭看到这里你可能会问ThreadPoolExecutor不香吗ExecutorService.invokeAll不香吗香但有几个场景它不够用第一任务需要动态拆分。线程池适合任务数量固定、每个任务独立的场景。但Master-Worker里一个任务执行完可能产生新的子任务比如递归式的目录扫描这种动态性线程池处理起来别扭。第二结果需要按类型归集。线程池的Future是一个任务一个结果但Master-Worker往往需要把结果按类别汇总比如把所有清洗成功的记录放一起、清洗失败的放另一处。这需要结果容器支持并发写入和分类。第三需要控制Worker的粒度。线程池的线程数一旦设定运行中调整很麻烦。而Master-Worker里Worker的数量、每个Worker处理的任务类型都可以在Master里灵活控制。当然如果你的需求就是提交100个独立任务等它们全部完成那老老实实用线程池别重复造轮子。自己搭组件的前提是你确实遇到了现成工具覆盖不了的场景。2.3 线程安全的三道防线队列、容器、状态整个组件里线程安全要守住三个地方第一道防线是任务队列。多个Worker同时从队列取任务必须保证一个任务只被一个Worker取走。用BlockingQueue的take()方法天然满足这个要求因为take()是原子操作内部用了锁和条件变量。第二道防线是结果容器。多个Worker同时往结果里写必须保证写入不丢、不覆盖。这里的选择就多了ConcurrentHashMap、ConcurrentLinkedQueue、CopyOnWriteArrayList各有各的适用场景。第三道防线是Worker的状态。Worker什么时候算干完了是队列空了就算还是要有显式的结束信号这个状态判断如果做错要么Master提前收工导致结果不全要么Master一直等导致程序挂死。这三道防线里第三道最容易被忽略也最容易出问题。后面我会专门用一节讲。3. 任务队列的选型BlockingQueue家族里挑哪个3.1 LinkedBlockingQueue和ArrayBlockingQueue的取舍BlockingQueue是个接口实现类有好几个最常用的是LinkedBlockingQueue和ArrayBlockingQueue。我实测下来的结论是任务量波动大的场景用Linked任务量稳定且追求极致吞吐用Array。LinkedBlockingQueue内部是链表理论上可以无限增长如果不指定容量入队出队用的是两把不同的锁putLock和takeLock所以生产者和消费者可以真正并行。代价是每个节点都要额外维护前后指针内存开销大一些。ArrayBlockingQueue内部是数组容量固定入队出队共用一把锁。这意味着同一时刻只能有一个线程在操作队列吞吐量在高并发下会受限。但它的优势是内存紧凑、没有节点分配开销而且容量固定意味着你能提前发现生产速度超过消费速度的问题。我的建议是默认用LinkedBlockingQueue并指定一个合理的容量上限。不指定容量的话一旦Worker处理速度跟不上队列会无限膨胀直到OOM。指定容量的经验值是Worker数量 × 单任务平均耗时对应的任务数 × 2留一倍缓冲。// 假设8个Worker每个任务平均处理50ms希望缓冲2秒的任务量 // 8 × (2000 / 50) × 2 640 BlockingQueueTask taskQueue new LinkedBlockingQueue(640);3.2 队列满了怎么办阻塞还是丢弃队列满了之后put()方法会阻塞offer()方法会返回false。这两个选择对应两种业务语义用put()Master投递任务时如果队列满就等着等Worker消费出空间再继续。这保证了任务不丢但Master会被拖慢。用offer()投递失败就返回falseMaster可以决定是重试、丢弃还是记录日志。这保证了Master不被拖慢但可能丢任务。我一般用offer()配合一个短时间的重试boolean offered taskQueue.offer(task, 100, TimeUnit.MILLISECONDS); if (!offered) { // 记录被拒绝的任务后续补偿处理 rejectedTasks.add(task); }注意被拒绝的任务一定要有地方记录否则就是静默丢数据。我见过一个系统因为队列满了直接offer不处理返回值结果每天丢几千条记录排查了三天才发现。3.3 用PriorityBlockingQueue做优先级调度如果任务有优先级比如紧急任务要先处理可以用PriorityBlockingQueue。但有个坑它虽然叫BlockingQueue但take()在队列为空时会阻塞put()却永远不会阻塞因为它是无界的。这意味着你没法用它做背压控制。用优先级队列时任务类必须实现Comparable接口而且比较逻辑要保证一致性——如果两个任务比较结果相等它们的处理顺序是不确定的。我一般会在优先级相同时用任务ID做二次比较保证顺序稳定。4. 结果归集ConcurrentHashMap、ConcurrentLinkedQueue还是自己加锁4.1 三种结果容器的适用场景对比结果容器的选择取决于你要怎么用结果。我把常见的三种方案列个表容器适用场景并发写性能读取便利性注意事项ConcurrentHashMap结果需要按key去重或分类高分段锁/CAS高key不能为nullConcurrentLinkedQueue结果只需按顺序追加高CAS中遍历时弱一致无界注意内存Collections.synchronizedList结果量小需要随机访问低全局锁高遍历时必须手动同步大部分Master-Worker场景结果都是追加式的用ConcurrentLinkedQueue最合适。它的add()是CAS操作无锁多个Worker同时写不会互相阻塞。缺点是它无界如果结果量特别大要注意内存。如果结果需要按任务ID去重比如同一个任务被重复提交那就用ConcurrentHashMapkey是任务IDvalue是结果。putIfAbsent可以保证同一个key只被写入一次。4.2 结果写入的原子性一个容易被忽略的细节假设你的结果不是单个值而是一个复合对象比如任务ID 处理结果 耗时。这时候如果分三次写resultMap.put(id, taskId); resultMap.put(result, processedData); resultMap.put(cost, costTime);三个操作之间可能被其他线程插入导致读到的数据不一致。正确做法是把这三个值封装成一个不可变对象一次性写入ResultRecord record new ResultRecord(taskId, processedData, costTime); resultQueue.add(record);ResultRecord的所有字段用final修饰构造完成后不可修改。这样写入的是一个完整的、不可变的对象天然线程安全。4.3 结果容器的容量控制ConcurrentLinkedQueue无界ConcurrentHashMap也无界。如果Worker产出结果的速度远大于Master消费结果的速度内存会一直涨。解决办法有两个一是限制结果数量超过阈值就暂停Worker。可以用Semaphore控制Semaphore resultPermits new Semaphore(10000); // Worker写结果前 resultPermits.acquire(); resultQueue.add(record); // Master消费结果后 resultPermits.release();二是边产边销Master不要等所有Worker干完才处理结果而是启动一个单独的消费者线程实时从结果队列取数据落库或转发。这样结果队列的长度就能控制在一个小范围内。5. Worker的生命周期管理怎么判断活干完了5.1 队列空了不等于活干完了这是新手最容易犯的错。Worker的逻辑通常是while (true) { Task task taskQueue.take(); // 阻塞等待 process(task); }Master怎么知道所有任务都处理完了如果判断队列为空那是不对的——队列为空可能只是暂时的Master还没投递完任务或者某个Worker正在处理最后一个任务还没写结果。正确的做法是引入一个结束信号。常见的有两种方案一毒丸Poison Pill。Master投递完所有任务后往队列里放N个特殊任务N等于Worker数量每个Worker取到毒丸就退出循环。// Master投递完任务后 for (int i 0; i workerCount; i) { taskQueue.put(Task.POISON_PILL); } // Worker循环 while (true) { Task task taskQueue.take(); if (task Task.POISON_PILL) { break; } process(task); }方案二计数器。用一个AtomicInteger记录未完成的任务数Master每投递一个加一Worker每完成一个减一。Master轮询这个计数器减到零就认为全部完成。AtomicInteger pendingTasks new AtomicInteger(0); // Master投递 pendingTasks.incrementAndGet(); taskQueue.put(task); // Worker完成 pendingTasks.decrementAndGet(); // Master等待 while (pendingTasks.get() 0) { Thread.sleep(10); }两种方案各有优劣。毒丸方案更优雅但要求Worker数量固定计数器方案更灵活但轮询有延迟。我一般用毒丸方案因为它的语义更清晰——毒丸就是明确的结束标志。5.2 Worker异常退出怎么办如果Worker在处理任务时抛了异常循环就断了这个Worker就死了。如果Master不知道还在等它处理任务就会一直等下去。解决办法是在Worker循环里包一层try-catchwhile (true) { Task task null; try { task taskQueue.take(); if (task Task.POISON_PILL) break; process(task); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { // 记录异常但不退出循环 log.error(Task failed: {}, task, e); failedTasks.add(task); } }注意InterruptedException要单独处理因为它意味着线程被外部中断了这时候应该退出循环而不是继续。其他异常则记录后继续保证Worker不会因为单个任务失败而整体挂掉。5.3 优雅关闭别让程序卡在take上程序要退出时如果Worker还阻塞在taskQueue.take()上System.exit()之前得先让它们醒过来。做法是调用workerThread.interrupt()take()会抛出InterruptedExceptionWorker捕获后退出循环。但要注意如果Worker正在处理任务不在take上interrupt()只是设置中断标志不会立即停止任务。所以优雅关闭的流程是Master停止投递新任务。投递毒丸或设置关闭标志。等待一段时间让Worker处理完手头任务。超时后调用interrupt()强制唤醒。join()等待所有Worker线程结束。// 优雅关闭 executor.shutdown(); if (!executor.awaitTermination(30, TimeUnit.SECONDS)) { executor.shutdownNow(); // 强制中断 }6. 一份可直接抄的完整实现6.1 核心类结构下面这份实现是我在实际项目中用过并简化后的版本去掉了业务相关的逻辑保留了并发控制的核心。你可以直接拿去改。import java.util.concurrent.*; import java.util.concurrent.atomic.AtomicInteger; // 任务接口 interface Task { String getTaskId(); boolean isPoisonPill(); } // 结果记录不可变对象 final class ResultRecord { private final String taskId; private final Object data; private final long costMillis; ResultRecord(String taskId, Object data, long costMillis) { this.taskId taskId; this.data data; this.costMillis costMillis; } // getter省略 } // Master-Worker组件 class MasterWorkerFrameworkT extends Task { private final BlockingQueueT taskQueue; private final ConcurrentLinkedQueueResultRecord resultQueue; private final ConcurrentLinkedQueueT failedTasks; private final AtomicInteger pendingCount; private final int workerCount; private final ExecutorService workerPool; private final T poisonPill; MasterWorkerFramework(int workerCount, int queueCapacity, T poisonPill) { this.workerCount workerCount; this.taskQueue new LinkedBlockingQueue(queueCapacity); this.resultQueue new ConcurrentLinkedQueue(); this.failedTasks new ConcurrentLinkedQueue(); this.pendingCount new AtomicInteger(0); this.poisonPill poisonPill; this.workerPool Executors.newFixedThreadPool(workerCount); } // 提交任务 boolean submit(T task) { pendingCount.incrementAndGet(); boolean offered taskQueue.offer(task); if (!offered) { pendingCount.decrementAndGet(); failedTasks.add(task); } return offered; } // 启动Worker void start(java.util.function.ConsumerT processor) { for (int i 0; i workerCount; i) { workerPool.submit(() - { while (true) { T task null; try { task taskQueue.take(); if (task.isPoisonPill()) break; long start System.currentTimeMillis(); processor.accept(task); long cost System.currentTimeMillis() - start; resultQueue.add(new ResultRecord( task.getTaskId(), null, cost)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { if (task ! null) failedTasks.add(task); } finally { if (task ! null !task.isPoisonPill()) { pendingCount.decrementAndGet(); } } } }); } } // 等待完成 boolean awaitCompletion(long timeout, TimeUnit unit) throws InterruptedException { long deadline System.nanoTime() unit.toNanos(timeout); while (pendingCount.get() 0) { if (System.nanoTime() deadline) return false; Thread.sleep(10); } // 投递毒丸 for (int i 0; i workerCount; i) { taskQueue.put(poisonPill); } workerPool.shutdown(); return workerPool.awaitTermination(30, TimeUnit.SECONDS); } ConcurrentLinkedQueueResultRecord getResults() { return resultQueue; } ConcurrentLinkedQueueT getFailedTasks() { return failedTasks; } }6.2 关键设计点说明这份代码里有几个地方是刻意设计的我逐个解释pendingCount的增减时机。提交任务时加一任务处理完无论成功失败减一。注意finally块里减一的条件是task ! null !task.isPoisonPill()因为毒丸不是真实任务不能算进计数。毒丸的投递时机。在awaitCompletion里等pendingCount归零后才投毒丸。这样保证所有真实任务都处理完了Worker才会收到结束信号。结果队列用ConcurrentLinkedQueue。因为结果只需要追加不需要按key查找用无锁队列性能最好。失败任务单独收集。失败的任务不混在结果里单独放一个队列方便后续补偿。6.3 一个完整的调用示例public class Demo { static class MyTask implements Task { private final String id; private final boolean poison; MyTask(String id, boolean poison) { this.id id; this.poison poison; } public String getTaskId() { return id; } public boolean isPoisonPill() { return poison; } } public static void main(String[] args) throws Exception { MyTask poison new MyTask(POISON, true); MasterWorkerFrameworkMyTask framework new MasterWorkerFramework(4, 1000, poison); framework.start(task - { // 模拟业务处理 try { Thread.sleep(50); } catch (InterruptedException ignored) {} }); for (int i 0; i 100; i) { framework.submit(new MyTask(task- i, false)); } boolean done framework.awaitCompletion(60, TimeUnit.SECONDS); System.out.println(完成: done); System.out.println(结果数: framework.getResults().size()); System.out.println(失败数: framework.getFailedTasks().size()); } }跑下来你会看到结果数是100失败数是0。如果把processor里改成随机抛异常失败数就会对应增加而结果数加上失败数始终等于100。这个总数守恒是验证组件正确性的一个好办法。7. 实测踩过的坑和性能调优经验7.1 坑一Worker数量不是越多越好我一开始想当然地设了workerCount CPU核心数 × 2结果发现吞吐量还不如单线程。原因是任务本身是CPU密集型的线程多了反而增加上下文切换开销。后来改成CPU核心数 1吞吐量上去了。经验公式CPU密集型任务用核心数 1IO密集型任务用核心数 × 2到核心数 × 4。但这不是绝对的最好用实际业务压测一遍。我一般会写一个简单的压测脚本分别用4、8、16个Worker跑同一批任务看哪个吞吐量最高。7.2 坑二队列容量设太小导致频繁阻塞队列容量设成Worker数量的两倍结果Master投递任务时频繁被阻塞整体耗时反而增加。后来把容量放大到Worker数量 × 100Master投递顺畅了Worker也不会饿死。队列容量的本质是生产速度和消费速度之间的缓冲。缓冲太小生产被拖慢缓冲太大内存占用高且故障时丢的数据多。我的经验值是队列容量 Worker数量 × 单Worker每秒处理任务数 × 期望缓冲秒数。比如8个Worker每个每秒处理20个任务希望缓冲5秒那就是8 × 20 × 5 800。7.3 坑三结果队列无界导致OOM有一次跑一个百万级任务结果队列没设上限跑到一半OOM了。后来加了Semaphore控制结果数量超过阈值就阻塞Worker等Master消费后再放行。但这里有个死锁风险如果Master也在等Worker完成才消费结果而Worker因为结果队列满被阻塞就死锁了。所以用Semaphore控制结果时Master必须边产边销不能等所有任务完成才处理结果。7.4 性能对比自研组件 vs 线程池我用同一个任务集10万个简单计算任务做了个对比方案耗时ms内存峰值MB代码复杂度单线程1250050低ThreadPoolExecutor3200120低自研Master-Worker3400135中可以看到自研组件比线程池略慢因为多了结果归集和计数逻辑但差距在10%以内。如果你的场景不需要动态拆任务和分类归集直接用线程池更划算。自研的价值在于灵活性和可控性不是性能。7.5 一个容易被忽略的调优点伪共享如果Worker数量多pendingCount这个AtomicInteger会被频繁读写可能产生伪共享False Sharing。解决办法是用Contended注解需要开启JVM参数-XX:-RestrictContended或者手动填充缓存行。不过在大多数业务场景下这个优化带来的收益很小不建议过早优化。8. 这套组件还能怎么扩展写到这里核心内容基本讲完了。最后分享几个我在实际使用中做的扩展你可以根据自己需求取舍。扩展一支持任务超时。给每个任务加一个超时时间Worker处理超时就放弃并记录。实现方式是在Worker里用Future包装任务future.get(timeout, unit)超时后取消。扩展二支持动态调整Worker数量。把ExecutorService换成可动态调整的线程池或者自己维护一个Worker列表根据队列长度动态增减Worker。扩展三支持结果回调。Master不主动拉结果而是注册一个回调函数Worker每产出一个结果就调用回调。这样结果可以实时处理不用等全部完成。扩展四支持分布式。把BlockingQueue换成分布式队列比如基于Redis的队列Worker部署到多台机器上。但这就复杂多了涉及到网络分区、任务重试、幂等性等问题不是单机组件能覆盖的。我个人在实际操作中的体会是并发组件的难点从来不是怎么写而是怎么验证。写完代码只是第一步更重要的是设计一套能复现并发问题的测试。我常用的方法是用CountDownLatch让所有Worker同时启动制造最大并发用随机延迟模拟真实处理时间跑一万次看结果是否始终一致。这套测试帮我抓出过好几个只在特定时序下才出现的bug。最后再分享一个小技巧如果你不确定自己的组件是否线程安全可以用ThreadSanitizer这类工具跑一遍或者用jcstress做压力测试。虽然不能保证100%覆盖但能发现大部分明显的问题。并发这东西宁可多测一千次也别赌它不会出错。