【Java 开发日记】我们来说一下无锁队列 Disruptor 的原理

核心操作（生产与消费）完全无锁（Lock-Free），通过内存屏障（Memory Barrier） 和 CAS（Compare-And-Swap） 操作实现线程安全。
生产者之间通过 CAS 竞争下一个可写的槽位。
生产者和消费者之间通过序列（Sequence） 的协调来工作，消费者通过等待策略（Wait Strategy） 来感知新数据的到来。

3. 消除伪共享（Cache Line Padding）：

识别出会被多个线程频繁写入的关键变量（如生产者的 cursor，各个消费者的 Sequence）。
通过在这些变量前后添加无意义的填充字节（padding），确保每个核心变量独占一个完整的 CPU 缓存行（通常为64字节），防止它们被意外地加载到同一个缓存行中，从而避免一个线程的写入使另一个线程的整个缓存行失效。

4. 批量与依赖关系：

支持批量处理事件，能极大提高吞吐量。
可以显式地构建消费者之间的依赖关系图（如 A->B->C 或 A,B 都完成 -> C），实现高效的工作流。

三、核心组件与原理

1. 环形缓冲区（Ring Buffer）

这是 Disruptor 的物理存储核心。它是一个固定大小的 Object[] 数组。每个位置被称为一个“槽”（slot）。

size：必须是2的幂次（如 1024）。这样 sequence % size 可以通过 sequence & (size - 1) 位运算高效完成。
cursor：生产者发布事件的序列号。它代表最后成功发布的事件的位置。这是一个 Sequence 对象。
缓冲区本身不维护“头”和“尾”指针，头和尾的概念由生产者和消费者的 Sequence 共同决定。

2. 序列（Sequence）

Disruptor 的灵魂。它是一个使用 padding 封装的长整型（long）值。

所有需要追踪进度的组件都有自己的 Sequence：
- Ring Buffer 有 cursor（一个 Sequence）。
- 每个 EventProcessor（消费者）有自己的 Sequence，表示自己已处理完成的位置。
- 每个 Producer（如果是多生产者）也有自己的 Sequence。
Sequence 的值单调递增，代表对应组件在环形缓冲区中的位置。
通过比较不同 Sequence 的值，就能知道生产和消费的进度关系。

3. 序列屏障（Sequence Barrier）

消费者用来协调工作、控制进度的核心工具。

它持有：
1. 生产者（或上游消费者）的 cursor 引用。
2. 所有它所依赖的消费者的 Sequence 引用（用于构建依赖图）。
当一个消费者想要消费事件时，它会询问它的 SequenceBarrier：“我可以安全消费的下一个事件是什么？”
SequenceBarrier 的逻辑是：返回 min(生产者cursor, 所有依赖的消费者的Sequence) 。这确保了消费者不会超越其依赖者，从而实现了无锁的有序消费。

4. 等待策略（Wait Strategy）

定义了消费者如何等待新事件的到来。这是影响延迟和 CPU 占用的关键。

BlockingWaitStrategy：使用锁和条件变量。最节省CPU，但延迟最高。适用于异步日志等场景。
SleepingWaitStrategy：先自旋，后 Thread.yield()，最后使用 LockSupport.parkNanos(1)。平衡延迟和CPU。
YieldingWaitStrategy：先自旋100次，然后调用 Thread.yield()。延迟低，但会占用较多CPU。适用于要求极高吞吐、线程数小于CPU核心数的场景。
BusySpinWaitStrategy：纯自旋。延迟最低，但疯狂消耗CPU。必须在绑定核心、线程数少于物理核心数的场景下使用。

5. 事件处理器（EventProcessor）

消费者的执行体。通常指 BatchEventProcessor。

它是一个线程，其 run() 方法内部是一个循环：
1. 通过 SequenceBarrier.waitFor(nextSequence) 等待自己可用的最大 nextSequence。
2. 获取到 availableSequence 后，从自己的当前 sequence 到 availableSequence 批量处理事件。
3. 调用 EventHandler.onEvent() 处理每个事件。
4. 处理完毕后，更新自己的消费者 Sequence 值。

6. 生产者（Producer）

负责向 Ring Buffer 发布事件。分为单生产者（Single Producer） 和多生产者（Multi Producer） 两种模式。

发布过程（两阶段提交）：
1. 申请空间（Claim）：
  - 单生产者：直接 nextSequence = cursor + 1（无竞争，无需CAS）。
  - 多生产者：通过 CAS 操作竞争递增一个 nextSequence。
2. 发布（Publish）：
  - 生产者将数据写入 nextSequence 对应的 slot。
  - 写入完成后，必须调用 RingBuffer.publish(sequence)。
  - publish 方法会先添加内存屏障（store-store barrier，确保数据写入先于 cursor 更新），然后将 cursor 更新到 sequence。
  - cursor 的更新会通知所有在 SequenceBarrier 上等待的消费者。

四、工作流程示例（单生产者 -> 单消费者）

初始化：
- Ring Buffer 大小为 8，cursor = -1。
- 消费者 Sequence = -1。
生产者发布事件：
- 生产者需要发布事件 A。它申请下一个位置：next = cursor + 1 = 0。
- 它将事件 A 的数据写入 RingBuffer[0 & 7]，即 RingBuffer[0]。
- 写入完成后，调用 publish(0)，更新 cursor = 0。
消费者消费事件：
- 消费者线程（BatchEventProcessor）在循环中调用 SequenceBarrier.waitFor(0)。
- SequenceBarrier 发现 cursor (0) >= 0，且没有依赖者，于是返回 availableSequence = 0。
- 消费者知道自己当前的 sequence (-1) < availableSequence (0)，于是处理 RingBuffer[0] 的事件 A。
- 处理完成后，将自己的 Sequence 更新为 0。
循环继续：生产者发布事件 B 到 slot 1，更新 cursor=1。消费者等待并处理，如此往复。

五、多消费者与依赖关系

这是 Disruptor 最强大的部分。例如，我们有三个消费者：C1（数据持久化），C2（数据统计），C3（发送消息，必须在 C1 和 C2 都完成后进行）。

C3 的 SequenceBarrier 会持有 RingBuffer.cursor、C1.sequence 和 C2.sequence。
当 C3 调用 waitFor 时，SequenceBarrier 返回的是 min(生产者cursor, C1.sequence, C2.sequence)。
这意味着，即使生产者已经发布了事件 10，但只要 C1 才处理到 5，C3 最多也只能拿到 5。这样就保证了 C3 不会跑到 C1 前面去，完全无锁地实现了依赖。

构建依赖图：

RingBuffer -> C1 -> C2 -> C3 (依赖 C1 和 C2)

六、总结：Disruptor 高性能的秘诀

预分配内存，消除GC：环形数组 + 对象复用。
无锁并发：CAS + 内存屏障，取代重量级锁。
消除伪共享：对关键序列进行缓存行填充。
批量处理：一次等待，处理多个事件，摊薄开销。
依赖关系感知：通过序列比较实现无锁的消费者协调，避免了“线程间握手”的开销。
关注点分离：将并发控制（Sequence， Barrier）、等待逻辑（WaitStrategy）、业务处理（EventHandler）清晰地解耦。

Disruptor 本质上是一种精心设计的内存队列，它将共享变量的数量降到最低（核心就是那几个 Sequence），并通过硬件友好的方式（缓存行填充、内存屏障）来操作它们，从而在软件层面最大限度地压榨出现代 CPU 和内存子系统的性能。它特别适用于金融交易、高频计算、事件溯源等对延迟和吞吐有极端要求的领域。

如果小假的内容对你有帮助，请点赞，评论，收藏。创作不易，大家的支持就是我坚持下去的动力！