企业官网建设流程全解析

1. 从一次线上事故说起：为什么要自己造一个Master-Worker并发组件

去年帮一个朋友排查他们系统里的一个诡异问题：一个批量对账任务，单线程跑要四十多分钟，改成多线程之后，速度确实上去了，但每隔几天就会在结果文件里发现几条重复记录，偶尔还会丢数据。日志翻了个底朝天，没有异常堆栈，没有报错，就是结果对不上。最后定位到问题出在他们自己写的一个简易任务分发逻辑上——主线程往队列里塞任务，几个工作线程从队列里取，用的是普通的ArrayList加一个synchronized块，锁的粒度没控制好，加上任务状态标记和结果收集是分开的两步操作，中间存在竞态窗口。

这个场景其实非常典型。但凡做过批量数据处理、爬虫调度、文件并行处理、订单异步核销这类需求的人，大概率都动过“自己写一个并发任务框架”的念头。而Master-Worker 模式就是这类需求里最常被拿出来说的一种并发组件设计思路。它的核心模型很朴素：一个 Master 负责接收和拆分任务，把子任务放进一个共享的任务队列；若干个 Worker 线程不断从队列里领取任务、执行、把结果写回结果集；Master 最后负责汇总。

听起来简单，但真正落地的时候，线程安全才是那个决定成败的命门。队列的并发读写、任务状态的原子性、结果收集的可见性、Worker 的优雅退出、异常任务的隔离——每一个点处理不好，都会变成那种“平时没事、压力一大就出鬼”的隐蔽 bug。

这篇内容就是围绕“线程安全”这个核心，把 Master-Worker 并发组件的设计思路、关键实现、踩坑经验完整地拆一遍。我会从整体架构讲到具体的数据结构选型，从锁策略讲到无锁队列的取舍，从任务分发讲到结果汇总的可见性保证，最后给出一份可以直接参考复现的实现方案和一份常见问题速查表。适合有一定 Java 并发基础、正在做批量任务调度或者想深入理解并发组件设计的读者。如果你只是想知道ExecutorService怎么用，那网上教程一大把；但如果你想搞清楚“为什么这么设计”“哪里会出问题”“怎么验证它真的线程安全”，那这篇应该能给你一些实在的东西。

2. 整体架构设计：Master-Worker 到底该怎么拆

2.1 核心角色划分与职责边界

Master-Worker 模式的结构，用一句话概括就是“一个调度中心加一组执行单元”。但落到代码层面，角色划分的粒度直接决定了后面线程安全的复杂度。我见过不少实现，把 Master 的职责和结果汇总揉在一起，Worker 又同时负责取任务和更新任务状态，职责交叉之后，锁的范围就变得很难界定。

比较清晰的一种拆法是四个角色：

• Master（调度者）：负责接收原始任务集合，按某种策略拆分成子任务，投递到任务队列，并在所有任务完成后触发结果汇总。它不参与具体任务执行。
• Worker（执行者）：从任务队列中竞争获取子任务，执行具体的业务逻辑，把执行结果写入结果容器。多个 Worker 并行运行。
• Task Queue（任务队列）：Master 和 Worker 之间的缓冲区，是并发竞争最激烈的地方，也是线程安全设计的重中之重。
• Result Set（结果集）：Worker 写、Master 读，需要保证写入的可见性和最终一致性。

这里有个设计上的关键取舍：任务队列和结果集是否要合并成一个对象。有些实现图省事，让 Worker 执行完直接把结果塞回任务对象本身，然后 Master 遍历任务集合取结果。这样做的问题是，任务对象同时承担了“待执行”和“已完成”两种状态，状态转换和结果写入如果不是原子的，就会出现 Master 读到半成品结果的情况。我的建议是任务队列和结果集物理分离，任务对象只承载输入数据，结果单独用一个并发容器收集，职责单一，线程安全边界也清晰。

2.2 为什么选 Master-Worker 而不是直接上线程池

这个问题我被问过很多次。既然 Java 有ThreadPoolExecutor，为什么还要自己搭一套 Master-Worker？答案在于任务的可拆分性和结果的聚合需求。

ThreadPoolExecutor解决的是“把一堆独立任务丢给线程池执行”的问题，它不关心任务之间的从属关系，也不提供结果汇总的语义。而 Master-Worker 模式天然适合那种“一个大任务拆成 N 个子任务、并行执行、最后合并结果”的场景，比如：

• 一个大文件按行分片，多个线程并行处理不同片段，最后合并统计结果；
• 一批订单按商户分组，每组并行核销，最后汇总成功失败数量；
• 一个爬取任务按 URL 分片，多线程抓取，最后统一入库。

这些场景的共同点是：有明确的拆分阶段、并行执行阶段和汇总阶段。用 Master-Worker 把这几个阶段显式建模出来，代码的可读性和可维护性会比硬塞进线程池好很多。当然，底层执行单元完全可以复用线程池，Master-Worker 负责的是调度逻辑和结果聚合，线程池负责的是线程的生命周期管理，两者不冲突。

2.3 线程安全设计的三个层次

在动手写代码之前，得先想清楚线程安全要解决哪些层次的问题。我把它分成三层：

第一层是数据结构的线程安全。任务队列和结果集是共享可变状态，多个线程同时读写，必须用线程安全的容器或者加锁保护。这一层是最基础的，也是最容易想到的。

第二层是复合操作的原子性。比如“判断队列是否为空 + 取出一个任务”这两步，如果分开做，中间就可能被其他线程插入或取走，导致空指针或者任务丢失。这一层是很多隐蔽 bug 的来源。

第三层是内存可见性和有序性。一个线程写入的结果，另一个线程能不能立刻看到？任务完成标记的写入和结果数据的写入，会不会被指令重排导致顺序错乱？这一层最抽象，但出问题时最难排查。

后面几章会围绕这三层逐一展开，把每个层次的具体实现和踩坑点讲透。

3. 任务队列的线程安全实现：从加锁到无锁的取舍

3.1 用 BlockingQueue 做任务队列的正确姿势

任务队列最省心的选型就是BlockingQueue。Java 并发包里提供了好几种实现，选哪个直接影响到吞吐量和线程安全特性。我整理了一个对比表，方便你根据场景选：

对于 Master-Worker 场景，我最常用的是LinkedBlockingQueue，并且一定要显式指定容量。原因很简单：如果 Worker 处理速度跟不上 Master 投递速度，无界队列会一直堆积任务对象，直到内存撑爆。指定一个合理的容量，比如Worker 数量 × 2到Worker 数量 × 4，既能起到缓冲作用，又能在积压时通过阻塞 Master 形成背压，避免内存失控。

// 推荐的任务队列初始化方式 int workerCount = Runtime.getRuntime().availableProcessors() * 2; BlockingQueue<Task> taskQueue = new LinkedBlockingQueue<>(workerCount * 4);

这里有个细节：LinkedBlockingQueue的put()方法在队列满时会阻塞，offer()方法在队列满时返回 false。Master 投递任务时用哪个，取决于你希望的行为。如果希望 Master 在队列满时等待，用put()；如果希望 Master 感知到积压并做其他处理（比如记录日志、降级），用offer()配合超时。

3.2 复合操作的原子性陷阱

用BlockingQueue之后，单个take()和put()是线程安全的，但复合操作仍然可能出问题。最典型的就是 Worker 的取任务循环：

// 错误示范：先判断再取，中间有竞态窗口 while (!taskQueue.isEmpty()) { Task task = taskQueue.take(); // 可能阻塞，也可能抛异常 process(task); }

这段代码的问题在于，isEmpty()返回 false 之后、take()执行之前，另一个 Worker 可能已经把最后一个任务取走了。此时take()会一直阻塞，Worker 线程就卡死了。正确的做法是用poll()配合超时，或者用一个“毒丸”标记来表示任务结束：

// 正确示范：用 poll 超时避免永久阻塞 while (running) { Task task = taskQueue.poll(100, TimeUnit.MILLISECONDS); if (task == null) { continue; // 队列暂时为空，继续循环检查 running 状态 } if (task == POISON_PILL) { break; // 收到结束标记，退出循环 } process(task); }

毒丸方案的好处是，Master 在投递完所有任务后，往队列里放 N 个毒丸（N 等于 Worker 数量），每个 Worker 取到一个毒丸就退出。这样既保证了所有任务都被处理完，又能让 Worker 优雅退出，不会出现“任务还没处理完就被强制中断”的情况。

注意：毒丸对象的类型必须和正常任务兼容，通常用一个特殊的 Task 子类或者用Optional包装。如果队列里混入了非任务对象，Worker 的类型转换会抛异常。

3.3 无锁队列的适用边界

有些追求极致性能的场景会考虑无锁队列，比如基于 CAS 的ConcurrentLinkedQueue。它的优势是入队出队不需要加锁，吞吐量在高并发下比BlockingQueue更好。但用在 Master-Worker 里有两个问题：

第一，ConcurrentLinkedQueue不提供阻塞语义。Worker 取任务时如果队列为空，poll()直接返回 null，需要自己写自旋等待逻辑。自旋等待如果控制不好，会空耗 CPU。

第二，无锁队列的size()方法是 O(n) 的，遍历整个链表才能算出大小。如果你的调度逻辑依赖队列大小做决策，用ConcurrentLinkedQueue会是个性能陷阱。

我的经验是：除非压测数据明确显示BlockingQueue的锁竞争是瓶颈，否则不要轻易换无锁队列。大多数业务场景下，任务执行本身的耗时远大于队列操作的耗时，锁竞争根本不是瓶颈。过早优化反而引入复杂度。

4. Worker 执行与结果收集的线程安全细节

4.1 Worker 线程的生命周期管理

Worker 线程的创建和销毁，建议交给线程池管理，而不是手动new Thread()。手动创建线程的问题在于：线程数量不可控、异常处理不统一、退出时机难协调。用ThreadPoolExecutor或者ForkJoinPool来承载 Worker，可以复用线程、统一异常处理策略。

// 用线程池承载 Worker int workerCount = Runtime.getRuntime().availableProcessors() * 2; ExecutorService workerPool = new ThreadPoolExecutor( workerCount, workerCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(workerCount), new ThreadFactory() { private final AtomicInteger counter = new AtomicInteger(0); @Override public Thread newThread(Runnable r) { Thread t = new Thread(r, "master-worker-" + counter.incrementAndGet()); t.setDaemon(false); // 非守护线程，保证任务执行完才退出 return t; } }, new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：调用者执行 );

这里有几个细节值得说。线程命名很重要，出问题时通过jstack能快速定位是哪个 Worker 卡住了。守护线程的设置要看场景，如果希望 JVM 在所有任务完成后才退出，就用非守护线程；如果希望主线程退出后 Worker 立即结束，就用守护线程。拒绝策略选CallerRunsPolicy是为了在 Worker 池满时让 Master 自己执行任务，形成天然的背压，避免任务丢失。

4.2 结果收集的可见性保证

结果收集是线程安全最容易翻车的地方。Worker 执行完任务后，把结果写入结果集，Master 最后读取结果集做汇总。这里涉及两个问题：写入的原子性和读取的可见性。

如果结果集用普通的HashMap，多个 Worker 同时put会直接导致死循环或者数据丢失（JDK 7 的HashMap在并发扩容时会形成环形链表，JDK 8 虽然修复了这个问题，但仍然不是线程安全的）。必须用ConcurrentHashMap或者加锁的Collections.synchronizedMap。

// 结果集推荐用 ConcurrentHashMap ConcurrentHashMap<String, TaskResult> resultMap = new ConcurrentHashMap<>(); // Worker 写入结果 resultMap.put(task.getTaskId(), taskResult); // Master 读取结果 for (Map.Entry<String, TaskResult> entry : resultMap.entrySet()) { // 汇总逻辑 }

ConcurrentHashMap保证了单个put操作的原子性和可见性，但复合操作仍然需要注意。比如“如果不存在则写入”这种逻辑，要用putIfAbsent()而不是先containsKey()再put()。再比如“累加计数”这种操作，要用LongAdder或者AtomicLong，而不是普通的int变量。

还有一个容易被忽略的点：结果对象的不可变性。如果 Worker 写入结果后，又修改了结果对象的内部字段，Master 读到的可能是修改后的值，也可能是修改前的值，取决于内存可见性。最稳妥的做法是结果对象设计成不可变对象，所有字段用final修饰，构造完成后不再修改。

4.3 任务状态标记的原子性

有些实现需要跟踪每个任务的执行状态（待执行、执行中、已完成、失败）。这个状态标记如果处理不好，会出现“任务被执行了两次”或者“任务状态显示已完成但结果没写入”的问题。

状态标记的更新必须是原子的。用AtomicInteger或者AtomicReference来承载状态，通过 CAS 操作做状态转换：

// 任务状态定义 enum TaskStatus { PENDING, RUNNING, COMPLETED, FAILED } // 用 AtomicReference 保证状态转换的原子性 AtomicReference<TaskStatus> status = new AtomicReference<>(TaskStatus.PENDING); // Worker 领取任务时，用 CAS 把状态从 PENDING 改为 RUNNING boolean claimed = status.compareAndSet(TaskStatus.PENDING, TaskStatus.RUNNING); if (!claimed) { // 任务已被其他 Worker 领取，跳过 return; }

这种 CAS 领取任务的方式，比“从队列取出就算领取”更严谨。因为从队列取出任务和标记任务状态是两步操作，中间如果 Worker 崩溃，任务状态就会不一致。用 CAS 把“领取”和“标记”合并成一个原子操作，能避免重复执行。

5. 完整实操：手写一个线程安全的 Master-Worker 组件

5.1 任务对象与结果对象定义

先把基础的数据结构定下来。任务对象承载输入数据，结果对象承载输出数据，两者都设计成不可变对象。

// 任务对象：不可变 public final class Task<T> { private final String taskId; private final T payload; public Task(String taskId, T payload) { this.taskId = taskId; this.payload = payload; } public String getTaskId() { return taskId; } public T getPayload() { return payload; } } // 结果对象：不可变 public final class TaskResult<R> { private final String taskId; private final boolean success; private final R data; private final String errorMsg; public TaskResult(String taskId, boolean success, R data, String errorMsg) { this.taskId = taskId; this.success = success; this.data = data; this.errorMsg = errorMsg; } // getter 省略 }

任务对象和结果对象都用final修饰字段，构造完成后不可修改。这样做的好处是，一旦对象被安全发布（通过并发容器或者 volatile 引用），所有线程看到的都是同一个不可变状态，不需要额外的同步。

5.2 Master 调度器的实现

Master 的职责是拆分任务、投递任务、等待完成、汇总结果。核心逻辑如下：

public class Master<T, R> { private final BlockingQueue<Task<T>> taskQueue; private final ConcurrentHashMap<String, TaskResult<R>> resultMap; private final List<Worker<T, R>> workers; private final ExecutorService workerPool; private final int workerCount; private volatile boolean running = false; public Master(int workerCount, int queueCapacity) { this.workerCount = workerCount; this.taskQueue = new LinkedBlockingQueue<>(queueCapacity); this.resultMap = new ConcurrentHashMap<>(); this.workerPool = Executors.newFixedThreadPool(workerCount); this.workers = new ArrayList<>(workerCount); } // 提交任务 public void submit(Task<T> task) throws InterruptedException { taskQueue.put(task); } // 启动 Worker 并等待所有任务完成 public Map<String, TaskResult<R>> execute() throws InterruptedException { running = true; CountDownLatch startLatch = new CountDownLatch(1); CountDownLatch doneLatch = new CountDownLatch(workerCount); for (int i = 0; i < workerCount; i++) { Worker<T, R> worker = new Worker<>(taskQueue, resultMap, startLatch, doneLatch); workers.add(worker); workerPool.submit(worker); } startLatch.countDown(); // 统一启动所有 Worker doneLatch.await(); // 等待所有 Worker 完成 running = false; workerPool.shutdown(); return resultMap; } }

这里用了两个CountDownLatch。startLatch用来让所有 Worker 同时开始，避免先启动的 Worker 把任务抢光、后启动的 Worker 空转。doneLatch用来让 Master 等待所有 Worker 退出。这种“统一启动、统一等待”的模式，在需要精确控制并发起点的场景里很实用。

5.3 Worker 执行单元的实现

Worker 的核心是一个循环：从队列取任务、执行、写结果、处理异常。关键是要保证循环能正确退出，异常不会导致 Worker 崩溃。

public class Worker<T, R> implements Runnable { private final BlockingQueue<Task<T>> taskQueue; private final ConcurrentHashMap<String, TaskResult<R>> resultMap; private final CountDownLatch startLatch; private final CountDownLatch doneLatch; private final TaskProcessor<T, R> processor; public Worker(BlockingQueue<Task<T>> taskQueue, ConcurrentHashMap<String, TaskResult<R>> resultMap, CountDownLatch startLatch, CountDownLatch doneLatch) { this.taskQueue = taskQueue; this.resultMap = resultMap; this.startLatch = startLatch; this.doneLatch = doneLatch; this.processor = new DefaultTaskProcessor<>(); } @Override public void run() { try { startLatch.await(); // 等待统一启动信号 while (true) { Task<T> task = taskQueue.poll(100, TimeUnit.MILLISECONDS); if (task == null) { // 队列暂时为空，检查是否应该退出 if (Thread.currentThread().isInterrupted()) { break; } continue; } if (task == Task.POISON_PILL) { break; // 收到毒丸，退出 } processTask(task); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); // 恢复中断状态 } finally { doneLatch.countDown(); // 通知 Master 自己已退出 } } private void processTask(Task<T> task) { try { R result = processor.process(task.getPayload()); resultMap.put(task.getTaskId(), new TaskResult<>(task.getTaskId(), true, result, null)); } catch (Exception e) { // 单个任务失败不影响其他任务 resultMap.put(task.getTaskId(), new TaskResult<>(task.getTaskId(), false, null, e.getMessage())); } } }

这段代码有几个关键点。第一，poll()带超时，避免永久阻塞，同时给 Worker 一个检查中断状态的机会。第二，任务执行用 try-catch 包裹，单个任务失败只记录错误结果，不会导致整个 Worker 线程崩溃。第三，finally块里countDown()，保证无论正常退出还是异常退出，Master 都能感知到。

5.4 毒丸退出机制与优雅关闭

毒丸机制是 Master-Worker 优雅退出的关键。Master 在投递完所有任务后，往队列里放 N 个毒丸，每个 Worker 取到一个就退出。

// Master 投递完任务后，放入毒丸 public void shutdown() throws InterruptedException { for (int i = 0; i < workerCount; i++) { taskQueue.put(Task.POISON_PILL); } }

毒丸对象的定义要注意类型兼容。因为BlockingQueue<Task<T>>的泛型约束，毒丸必须是一个Task<T>实例。可以用一个静态常量：

public final class Task<T> { // 毒丸标记 @SuppressWarnings("rawtypes") public static final Task POISON_PILL = new Task<>("__POISON__", null); // ... }

用==比较毒丸，而不是equals()，因为毒丸是单例，引用相等就够了，而且==比较更快。

注意：毒丸方案要求 Master 在投递毒丸之前，确保所有正常任务都已经投递完毕。如果 Master 是多线程投递任务的，需要先等待所有投递线程结束，再投递毒丸。否则可能出现 Worker 收到毒丸退出后，还有任务没被处理的情况。

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

6.1 任务丢失与重复执行的排查思路

任务丢失和重复执行是 Master-Worker 最典型的两类问题，排查思路不太一样。

任务丢失通常有几个原因：一是队列容量满了之后，Master 用offer()投递失败但没有处理返回值，任务被静默丢弃；二是 Worker 取到任务后、执行前发生异常，任务没有被记录到结果集；三是毒丸投递过早，Worker 退出时还有任务在队列里。

排查方法：在 Master 投递任务时记录投递计数，在 Worker 完成时记录完成计数，最后对比两个计数是否一致。如果不一致，差额就是丢失的任务数。再结合日志定位是投递阶段丢的还是执行阶段丢的。

重复执行的原因通常是任务被多个 Worker 同时领取。如果用BlockingQueue，take()本身是原子的，不会重复领取。但如果任务状态标记和领取是分开的两步操作，就可能出现重复。排查方法是给每个任务加一个执行计数器，用AtomicInteger记录被执行次数，如果大于 1 就打印告警日志。

6.2 内存可见性导致的“幽灵 bug”

内存可见性问题最难排查，因为它的表现是“偶发”“不可复现”“换台机器就好了”。典型场景是：Worker 写入了结果，Master 读到的却是旧值或者 null。

排查这类问题的第一步是检查共享变量有没有用volatile或者并发容器保护。如果用了普通的HashMap或者普通对象字段，基本可以确定是可见性问题。第二步是检查有没有“先写标记、后写数据”的顺序依赖，如果有，需要保证标记的写入在数据写入之后（用volatile写或者锁释放来保证 happens-before）。

一个实用的技巧是：在关键路径上加日志，记录线程名和时间戳。如果发现 Master 读到的结果时间戳早于 Worker 写入的时间戳，那就是可见性问题。

6.3 常见问题速查表

6.4 几个我踩过的坑

第一个坑是队列容量设得太大。有次为了“避免阻塞”，把队列容量设成了 100 万，结果 Worker 处理速度跟不上，任务对象在队列里堆了几十万个，每个任务对象还带着大字段，直接把堆内存撑爆了。后来改成Worker 数量 × 4，配合监控队列大小，一旦超过阈值就告警，问题就解决了。

第二个坑是结果集用Collections.synchronizedMap。这个包装类虽然线程安全，但所有操作共用一把锁，Worker 数量一多，结果写入就成了瓶颈。换成ConcurrentHashMap之后，吞吐量提升了将近一倍。

第三个坑是忘记处理中断。Worker 在poll()阻塞时，如果外部调用了shutdownNow()，会抛InterruptedException。如果捕获后不恢复中断状态，上层就感知不到中断信号，导致关闭流程卡住。正确的做法是在 catch 块里调用Thread.currentThread().interrupt()恢复中断状态。

第四个坑是毒丸和正常任务混用同一个队列。有次毒丸投递之后，Master 又投递了几个正常任务，结果这些任务排在毒丸后面，Worker 取到毒丸就退出了，后面的任务没人处理。后来改成 Master 先投递完所有任务，再统一投毒丸，并且用一个AtomicBoolean标记“是否已开始投毒丸”，防止投递顺序错乱。

7. 性能调优与扩展方向

7.1 Worker 数量的确定

Worker 数量不是越多越好。对于 CPU 密集型任务，Worker 数量设为 CPU 核心数或核心数加一比较合适；对于 IO 密集型任务，可以设为 CPU 核心数的两到四倍。但这不是绝对的，最好通过压测来确定。

我通常的做法是：先设一个初始值（比如CPU 核心数 × 2），然后逐步增加 Worker 数量，观察吞吐量和 CPU 使用率。如果吞吐量不再增长、CPU 使用率接近 100%，说明 Worker 数量已经够了；如果 CPU 使用率不高但吞吐量上不去，可能是锁竞争或者 IO 瓶颈，需要进一步分析。

7.2 结果集的批量写入优化

如果任务数量很大（比如几十万），每个任务完成都往ConcurrentHashMap里put一次，会有一定的开销。可以考虑批量写入：Worker 在本地缓存一批结果，攒够一定数量或者等待一定时间后，一次性写入结果集。这样能减少并发容器的操作次数，提升吞吐量。

但批量写入会带来一个问题：Master 在汇总时，可能有一部分结果还在 Worker 的本地缓存里没写入。解决办法是在 Worker 退出前，强制把本地缓存刷入结果集。或者在 Master 汇总前，等待所有 Worker 的doneLatch归零，确保所有缓存都已刷入。

7.3 任务优先级与超时控制

有些场景下任务有优先级，比如紧急任务要优先处理。这时候可以用PriorityBlockingQueue替代LinkedBlockingQueue，任务对象实现Comparable接口，按优先级排序。但要注意PriorityBlockingQueue是无界的，需要自己控制任务数量。

超时控制是另一个常见的扩展需求。如果某个任务执行时间过长，需要强制中断并标记为超时失败。实现方式是在 Worker 里用Future包装任务执行，通过Future.get(timeout, unit)设置超时。超时后调用Future.cancel(true)中断任务线程。

// 带超时的任务执行 Future<R> future = executor.submit(() -> processor.process(task.getPayload())); try { R result = future.get(30, TimeUnit.SECONDS); resultMap.put(task.getTaskId(), new TaskResult<>(task.getTaskId(), true, result, null)); } catch (TimeoutException e) { future.cancel(true); resultMap.put(task.getTaskId(), new TaskResult<>(task.getTaskId(), false, null, "timeout")); }

这里要注意，cancel(true)只是发送中断信号，任务代码本身要响应中断才能真的停下来。如果任务代码里有阻塞操作（比如Thread.sleep()、wait()），会抛InterruptedException；如果是纯计算循环，需要自己检查Thread.currentThread().isInterrupted()。

8. 写在最后：线程安全没有银弹

做并发组件这些年，我最大的体会是：线程安全不是靠某个“万能容器”或者“高级锁”就能解决的，它是一整套设计约束和编码纪律。用ConcurrentHashMap不代表就安全了，如果复合操作没处理好，照样出问题；用synchronized也不代表就低效，如果锁的范围合理，性能完全够用。

Master-Worker 这个模式本身不复杂，复杂的是并发环境下的各种边界情况。我的建议是：先把单线程版本跑通，确认业务逻辑正确；然后逐步引入并发，每引入一个并发点，就想清楚它的线程安全边界在哪里；最后用压测和日志验证，确保在高并发下行为符合预期。

如果你正在做类似的东西，不妨从最简单的BlockingQueue加ConcurrentHashMap开始，把毒丸退出和异常隔离做好，大概率就能覆盖百分之八十的场景。等真的遇到性能瓶颈了，再考虑无锁队列、批量写入这些优化。过早优化除了增加复杂度，没什么别的好处。

最后分享一个我常用的验证技巧：在测试环境把 Worker 数量调到 CPU 核心数的十倍，然后跑一个包含各种边界情况（空任务、超长任务、异常任务、大量小任务）的测试集。如果这个配置下结果依然正确、没有死锁、没有内存泄漏，那基本可以认为线程安全设计是过关的。这个“十倍压力测试”帮我提前发现过好几次隐蔽的竞态问题，比单纯看代码靠谱得多。