企业官网建设流程全解析

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 BlockingQueue<Task> taskQueue = new LinkedBlockingQueue<>(640);

3.2 队列满了怎么办：阻塞还是丢弃

队列满了之后，put()方法会阻塞，offer()方法会返回false。这两个选择对应两种业务语义：

• 用put()：Master投递任务时如果队列满就等着，等Worker消费出空间再继续。这保证了任务不丢，但Master会被拖慢。
• 用offer()：投递失败就返回false，Master可以决定是重试、丢弃还是记录日志。这保证了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 三种结果容器的适用场景对比

结果容器的选择取决于你要怎么用结果。我把常见的三种方案列个表：

大部分Master-Worker场景，结果都是"追加式"的，用ConcurrentLinkedQueue最合适。它的add()是CAS操作，无锁，多个Worker同时写不会互相阻塞。缺点是它无界，如果结果量特别大，要注意内存。

如果结果需要按任务ID去重（比如同一个任务被重复提交），那就用ConcurrentHashMap，key是任务ID，value是结果。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-catch：

while (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()会抛出InterruptedException，Worker捕获后退出循环。

但要注意，如果Worker正在处理任务（不在take上），interrupt()只是设置中断标志，不会立即停止任务。所以优雅关闭的流程是：

1. Master停止投递新任务。
2. 投递毒丸或设置关闭标志。
3. 等待一段时间让Worker处理完手头任务。
4. 超时后调用interrupt()强制唤醒。
5. 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 MasterWorkerFramework<T extends Task> { private final BlockingQueue<T> taskQueue; private final ConcurrentLinkedQueue<ResultRecord> resultQueue; private final ConcurrentLinkedQueue<T> 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.Consumer<T> 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); } ConcurrentLinkedQueue<ResultRecord> getResults() { return resultQueue; } ConcurrentLinkedQueue<T> 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); MasterWorkerFramework<MyTask> 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密集型任务用核心数 + 1，IO密集型任务用核心数 × 2到核心数 × 4。但这不是绝对的，最好用实际业务压测一遍。我一般会写一个简单的压测脚本，分别用4、8、16个Worker跑同一批任务，看哪个吞吐量最高。

7.2 坑二：队列容量设太小导致频繁阻塞

队列容量设成Worker数量的两倍，结果Master投递任务时频繁被阻塞，整体耗时反而增加。后来把容量放大到Worker数量 × 100，Master投递顺畅了，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万个简单计算任务）做了个对比：

可以看到，自研组件比线程池略慢（因为多了结果归集和计数逻辑），但差距在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%覆盖，但能发现大部分明显的问题。并发这东西，宁可多测一千次，也别赌它不会出错。