并行写入场景下如何通过毒丸模式优雅停止所有工作线程？

可行方案参考 #

方案1：投放与并行度等量的毒丸 #

这是最简洁的实现方案，不需要修改现有BlockingQueueItemReader的逻辑，只需要在上游读取步骤完成所有业务数据入队后，向阻塞队列中投放和step2并行度数量一致的毒丸即可。你当前step2配置的throttleLimit为8，直接投放8个POISON对象，每个工作线程读取到毒丸后会自行退出，不会出现剩余线程无限等待的问题。

• 优点：无额外线程安全处理逻辑，代码改动量最小
• 注意点：后续调整step2的并行度时，需要同步修改毒丸的投放数量，保证两者一致。

方案2：使用全局原子停止标志位 #

如果不想将毒丸数量和并行度绑定，可以引入线程安全的原子标志位作为全局停止信号，所有读取器实例共享该标志位，任意线程读取到毒丸后修改标志位，其余线程下次读取时感知到信号直接退出。
代码实现参考如下：

@RequiredArgsConstructor
public class BlockingQueueItemReader<T> implements ItemReader<T> {

    private final BlockingQueue<T> queue;
    private final T poisonPill;
    private final int timeoutSeconds;
    // 线程安全的原子标志位，所有读取线程共享
    private final AtomicBoolean exhausted = new AtomicBoolean(false);

    @Nullable
    @Override
    public T read() throws Exception {
        if (exhausted.get()) {
            return null;
        }
        T taken = queue.poll(timeoutSeconds, TimeUnit.SECONDS);
        if (poisonPill.equals(taken)) {
            exhausted.set(true);
            return null;
        }
        return taken;
    }
}

• 优点：和并行度完全解耦，只需投放1个毒丸即可触发所有线程停止，逻辑更简洁
• 注意点：你原有实现中的exhausted是普通实例变量且未加volatile修饰，存在多线程可见性问题，部分线程可能感知不到变量变更，必须用AtomicBoolean或者加volatile修饰保证可见性。

如果需要兼容队列临时无数据（非耗尽）的超时场景，可以在poll返回null时额外判断上游读取步骤是否已完成，避免误判退出。

内容的提问来源于stack exchange，提问作者flywell