一、Disruptor是什么
Disruptor是一个开源的并发框架，能够在无锁的情况下实现网络的Queue并发操作。
Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量级JMS），也可以认为是一个观察者模式的实现，或者事件监听模式的实现。
二、Disruptor核心概念
RingBuffer：
被看做Disruptor最主要组件，然而从3.0开始RingBuffer仅仅负责存储和更新在Disruptor中流通的数据。对一些特殊的使用场景能够被用户（其他数据结构）完全替代。
Sequence：
Disruptor使用Sequence来表示一个特殊组件处理的序号。和Disruptor一样，每个消费者（EventProcessor）都维持着一个Sequence。大部分的并发代码依赖这些Sequence值得运转，因此Sequence支持多种当前为AtomicLong类的特性。
Sequencer：
这是Disruptor真正的核心。实现了这个接口的两种生产者（单生产者和多生产者）均实现了所有的并发算法，为了在生产者和消费者之间进行准确快速的数据传递。
SequenceBarrier：
有Sequence生成，并且包含了已经发布的Sequence的引用，这些Sequence源于Sequenceer和一些独立的消费者的Sequence。它包含了决定是否有供消费者来消费者的Event的逻辑。
WaitStrategy：
决定了一个消费者将如何等待生产者将Event置于Disruptor。
Event：
从生产者到消费者过程中所处理的数据单元。Disruptor中没有代码表示Event，因此它完全是由用户定义。
EventProcessor：
主要时间循环，处理Disruptor中的Event，并且拥有消费者的Sequence。他有一个实现类是BatchEventProcessor，包含了event loop有效的实现，并且将回调到一个EventHandler接口的实现对象。
EventHandler：
由用户实现并且代表了Disruptor中的一个消费者的接口。
Producer：
由用户实现，它调用RingBuffer来插入事件（Event），在Disruptor中没有相应的实现代码，由用户实现。
WorkProcessor：
确保每个sequence只被一个processor消费，在同一个WorkPool中的处理多个WorkProcessor不会消费同样的sequence。
WorkerPool：
一个WorkProcessor池，其中WorkProcessor将消费Sequence，所以任务可以在实现WorkHandler接口的worker之间移交。
LifecycleAware：
当BatchEventProcessor启动和停止时，实现这个接口用于接收通知。
三、Disruptor工作原理
Disruptor的核心是RingBuffer，主要用来存储和更新数据，生产者负责往RingBuffer中写数据，消费者从RingBuffer中消费数据。
RingBuffer其实就是一个环，首尾连接的一个环，只维护一个next()指向的下一个元素，当数据超过最大的限制的时候，其实就是一个覆盖操作，打个比方现在RingBuffer的长度为2^3=8，将下标从0到7的索引填满之后，下一个索引应该是8，这个时候其实只需要进行简单的取模操作即可，8%8就是覆盖第0个元素对象的数据，12的话就是12%8=4，就覆写索引下标为4的位置。
之前有说RingBuffer中槽的大小需要是2^n，这样是因为假如是2的n次方的话，进行操作完全可以用位运算，这是速度非常快的，比如: 8 >> 1 = 4 8 << 1 = 16
四、Disruptor使用
在Disruptor开发过程中，主要需要实现如下几个步骤：
1.定义事件
事件(Event)就是通过 Disruptor 进行交换的数据类型
public class LongEvent {
private long value;
public long getValue() {
return value;
}
public void setValue(long value) {
this.value = value;
}
}
2.定义事件工厂
事件工厂(Event Factory)定义了如何实例化前面第1步中定义的事件(Event)，需要实现接口 com.lmax.disruptor.EventFactory。
Disruptor 通过 EventFactory 在 RingBuffer 中预创建 Event 的实例。
一个 Event 实例实际上被用作一个“数据槽”，发布者发布前，先从 RingBuffer 获得一个 Event 的实例，然后往 Event 实例中填充数据，之后再发布到 RingBuffer 中，之后由 Consumer 获得该 Event 实例并从中读取数据。
/**
* 需要让disruptor为我们创建事件，我们同时还声明一个EventFactory来实例化Event对象
* */
public class LongEventFactory implements EventFactory{
@Override
public Object newInstance() {
return new LongEvent();
}
}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
3.定义事件监听器类
通过实现接口 com.lmax.disruptor.EventHandler 定义事件处理的具体实现。
/**
* 事件消费者：
* 也就是一个事件处理器，这个事件处理器简单地把事件中存储的数据打印到终端
* */
public class LongEventHandler implements EventHandler {
@Override
public void onEvent(LongEvent arg0, long arg1, boolean arg2) throws Exception {
System.out.println(arg0.getValue());
}
}
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
4.定义用于事件处理的线程池
Disruptor 通过 java.util.concurrent.ExecutorService 提供的线程来触发 Consumer 的事件处理。例如：
ExecutorService executor = Executors.newCachedThreadPool();
1.
5.指定等待策略
Disruptor 定义了 com.lmax.disruptor.WaitStrategy 接口用于抽象 Consumer 如何等待新事件，这是策略模式的应用。
Disruptor 提供了多个 WaitStrategy 的实现，每种策略都具有不同性能和优缺点，根据实际运行环境的 CPU 的硬件特点选择恰当的策略，并配合特定的 JVM 的配置参数，能够实现不同的性能提升。
例如，BlockingWaitStrategy、SleepingWaitStrategy、YieldingWaitStrategy 等，其中，
BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；
SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；
YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。
WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
1.
2.
3.
6.启动 Disruptor
EventFactory eventFactory = new LongEventFactory();
ExecutorService executor = Executors.newSingleThreadExecutor();
int ringBufferSize = 1024 * 1024; // RingBuffer 大小，必须是 2 的 N 次方；
Disruptor disruptor = new Disruptor(eventFactory,
ringBufferSize, executor, ProducerType.SINGLE,
new YieldingWaitStrategy());
EventHandler eventHandler = new LongEventHandler();
disruptor.handleEventsWith(eventHandler);
disruptor.start();
1.
2.
3.
4.
5.
6.
7.
8.
9.
10.
11.
12.
7.发布事件
Disruptor 的事件发布过程是一个两阶段提交的过程：
第一步：先从 RingBuffer 获取下一个可以写入的事件的序号；
第二步：获取对应的事件对象，将数据写入事件对象；
第三步：将事件提交到 RingBuffer;
事件只有在提交之后才会通知 EventProcessor 进行处理；
// 发布事件；
RingBuffer ringBuffer = disruptor.getRingBuffer();
long sequence = ringBuffer.next();//请求下一个事件序号；
try {
LongEvent event = ringBuffer.get(sequence);//获取该序号对应的事件对象；
long data = getEventData();//获取要通过事件传递的业务数据；
event.set(data);
} finally{
ringBuffer.publish(sequence);//发布事件；
}
注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
8.关闭 Disruptor
disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；
1.
2.
4~8步主要是实例化Disruptor实例，定义用于事件处理的线程池，指定等待策略，配置一系列参数。然后我们对Disruptor实例绑定监听事件类，接收并处理数据。
public class LongEventMain {
public static void main(String[] args) {
//创建线程池
ExecutorService executor = Executors.newCachedThreadPool();
//创建工程
LongEventFactory factory = new LongEventFactory();
//创建bufferSize，也就是RingBuffer大小，必须是2的N次方
int ringBufferSize = 1024 * 1024;
/**
BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景
WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性
WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
*/
/* 创建disruptor
* 1.第一个参数为工厂类对象，用于创建一个个的LongEvent，LongEvent是实际的消费数据。
* 2.第二个参数为缓存区大小
* 3.第三个参数为线程池，进行Disruptor内部的数据接收处理调度
* 4.第四个参数ProducerType.SINGLE(表示一个生产者) 和 ProducerType.MULTI(多个生产者)
* 5.第五个参数是一种策略，就是生产和消费的策略
* */
Disruptor disruptor = new Disruptor(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
//连接消费事件方法
disruptor.handleEventsWith(new LongEventHandler());
//启动
disruptor.start();
//Disruptor的事件发布过程是一个两阶段提交过程
//使用该方法获得具体存放数据的容器RingBuffer（环形结构）
RingBuffer ringBuffer = disruptor.getRingBuffer();
LongEventProducer producer = new LongEventProducer(ringBuffer);
ByteBuffer byteBuffer = ByteBuffer.allocate(8);
for(long input = 0; input < 100; input++){
byteBuffer.putLong(0,input);
producer.onData(byteBuffer);
}
disruptor.shutdown(); //关闭disruptor，方法会阻塞，直到所有的时间得到处理
executor.shutdown(); //关闭executor，disruptor不会自动关闭executor
}
}
public class LongEventProducer {
private final RingBuffer ringBuffer;
public LongEventProducer(RingBuffer ringBuffer){
this.ringBuffer = ringBuffer;
}
/**
* onData用来发布事件，每调用一次就发布一次事件
* 他的参数会用该事件传递给消费者
* */
public void onData(ByteBuffer input){
//1.把ringBuffer看做一个事件队列，那么next就是得到下一个事件槽
long sequence = ringBuffer.next();
try{
//2.用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）
LongEvent event = ringBuffer.get(sequence);
//3.获取要通过事件传递的业务数据
event.setValue(input.getLong(0));
} finally {
//4.发布事件
//注意，最后的ringBuffer.publish 方法必须包含在finally中以确保必须得到调用
//如果某个请求的sequence未被提交，则对应的消费者获取不了数据
ringBuffer.publish(sequence);
}
}
}
注意：对Disruptor实例绑定监听事件类有两种方式，使用handleEventsWithWorkerPool可以完成不重复消费，使用handleEventsWith就是重复消费。
handleEventsWith
//连接消费事件方法
disruptor.handleEventsWith(new LongEventHandler());
1.
2.
handleEventsWithWorkerPool
这次的消费者需要实现不同的接口，实现的是WorkHandler接口
import com.lmax.disruptor.WorkHandler;
public class Consumer implements WorkHandler {
@Override
public void onEvent(LongEvent longEvent) throws Exception {
System.out.println(Thread.currentThread().getName() + “消费者消费了消息：” + longEvent.toString());
}
}
登录后复制
public class Test {
public static void main(String[] args) {
ThreadFactory producerFactory = Executors.defaultThreadFactory();
// 创建缓冲池
LongEventFactory eventFactory = new LongEventFactory();
// 创建bufferSize ,也就是RingBuffer大小，必须是2的N次方
int ringBufferSize = 1024 * 1024;
Disruptor disruptor = new Disruptor(eventFactory, ringBufferSize, producerFactory,
ProducerType.SINGLE, new BlockingWaitStrategy());
RingBuffer ringBuffer = disruptor.getRingBuffer();
// 创建10个消费者来处理同一个生产者发的消息(这10个消费者不重复消费消息)
Consumer[] consumers = new Consumer[10];
for (int i = 0; i < consumers.length; i++) {
consumers[i] = new Consumer();
}
disruptor.handleEventsWithWorkerPool(consumers);
disruptor.start();
LongEventProducer longEventProducer = new LongEventProducer(ringBuffer);
ByteBuffer bb = ByteBuffer.allocate(8);
for (long i = 0; i < 100L; i++) {
bb.putLong(0, i);
longEventProducer.onData(bb);
}
disruptor.shutdown();
}
}