深入理解 Disruptor
1. 概述
Disruptor 是一个高性能、低延迟的无锁队列替代方案，最初由 LMAX 公司开发，专为处理高吞吐量和低延迟的消息传递系统而设计。它利用环形缓冲区（RingBuffer）和无锁的生产者-消费者模型，大幅提升并发性能。
相比传统的基于 java.util.concurrent 的队列（如 ArrayBlockingQueue、LinkedBlockingQueue），Disruptor 通过避免锁竞争、减少 CPU 缓存行无效（cache invalidation）等方式提高吞吐量。
2. 核心概念
2.1 RingBuffer（环形缓冲区）
Disruptor 的核心数据结构是环形缓冲区（RingBuffer），它类似于一个固定大小的数组，数据结构如下：
+----+----+----+----+----+----+----+----+
| 0 | 1 | 2 | 3 | 4 | 5 | 6 | 7 |
+----+----+----+----+----+----+----+----+
RingBuffer 通过索引递增的方式循环使用元素，避免内存分配和垃圾回收的开销。
2.2 Sequence（序列号）
在 Disruptor 中，所有读写操作都基于 Sequence，用于跟踪当前生产和消费的位置。它主要包括：
-
Cursor：指向 RingBuffer 中最后一个被写入的位置。 -
SequenceBarrier：用于协调生产者和消费者的进度，确保消费者不会读取尚未发布的数据。 -
Sequencer：用于管理 RingBuffer 的序列。
2.3 Producer（生产者）
生产者向 RingBuffer 写入数据，通常采用 ClaimStrategy 申请空间，然后写入数据并发布。
2.4 Consumer（消费者）
消费者从 RingBuffer 读取数据，并可以设置多个消费者进行并行处理，支持 WorkerPool 模式。
2.5 WaitStrategy（等待策略）
Disruptor 通过 WaitStrategy 来决定消费者如何等待新的数据到达。常见策略包括：
-
BusySpinWaitStrategy：自旋等待，适用于低延迟应用，但 CPU 开销较大。 -
SleepingWaitStrategy：适当休眠，减少 CPU 占用。 -
YieldingWaitStrategy：让出 CPU 时间片，适用于高吞吐场景。
3. Disruptor 的优势
3.1 无锁设计
传统队列使用 ReentrantLock 或 synchronized 来保证线程安全，而 Disruptor 通过 CAS（Compare-And-Swap）机制更新 Sequence，避免锁的开销。
3.2 高效的 CPU 缓存利用
Disruptor 采用 伪共享（False Sharing） 避免 CPU 缓存行竞争，并使用 缓存行填充（Cache Line Padding） 来减少缓存行失效。
3.3 生产者-消费者模型优化
Disruptor 允许多种消费者模式：
- 单消费者：一个消费者处理所有数据。
- 多消费者并行消费：多个消费者共同消费数据，提高吞吐量。
- 菱形依赖消费：一个消费者的输出作为另一个消费者的输入。
4. 使用示例
4.1 引入依赖
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.4.2</version>
</dependency>
4.2 创建事件类
public class LongEvent {
private long value;
public void set(long value) {
this.value = value;
}
public long getValue() {
return value;
}
}
4.3 定义事件工厂
import com.lmax.disruptor.EventFactory;
public class LongEventFactory implements EventFactory<LongEvent> {
@Override
public LongEvent newInstance() {
return new LongEvent();
}
}
4.4 事件处理器
import com.lmax.disruptor.EventHandler;
public class LongEventHandler implements EventHandler<LongEvent> {
@Override
public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
System.out.println("Event: " + event.getValue());
}
}
4.5 配置 Disruptor
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class DisruptorExample {
public static void main(String[] args) {
ExecutorService executor = Executors.newCachedThreadPool();
LongEventFactory factory = new LongEventFactory();
int bufferSize = 1024;
Disruptor<LongEvent> disruptor = new Disruptor<>(
factory, bufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
disruptor.handleEventsWith(new LongEventHandler());
disruptor.start();
RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
LongEventProducer producer = new LongEventProducer(ringBuffer);
for (long i = 0; i < 10; i++) {
producer.onData(i);
}
}
}
5. 适用场景
Disruptor 适用于以下场景：
- 高吞吐量、低延迟的消息队列
- 日志系统（如
log4j2采用 Disruptor 作为日志处理引擎） - 交易撮合系统
- 事件驱动架构
6. 总结
今天先初步了解Disruptor的简单用法，后续会继续介绍Disruptor的特性，为什么性能秒杀JDK提供的队列，以及相关原理分析。
最后
欢迎follow加瓦点灯，每天推送干货知识！
本文由mdnice多平台发布
