Disruptor—1.原理和使用简介
大纲1.Disruptor简介
2.Disruptor和BlockingQueue的压测对比
3.Disruptor的编程模型
4.Disruptor的数据结构与生产消费模型
5.RingBuffer + Disruptor + Sequence相关类
6.Disruptor的WaitStrategy消费者等待策略
7.EventProcessor + EventHandler等类
8.Disruptor的运行原理图
9.复杂业务需求下的编码方案和框架
10.Disruptor的串行操作
11.Disruptor的并行操作
12.Disruptor的多边形操作
13.Disruptor的多生产者和多消费者
1.Disruptor简介
(1)Disruptor是什么
(2)Disruptor的特点
(3)Disruptor的核心
(1)Disruptor是什么
Martin Fowler在自己网站上写了一篇LMAX架构的文章，在文章中他介绍了LMAX是一种新型零售金融交易平台，能够以很低的延迟产生大量的交易。LMAX是建立在JVM平台上，其核心是一个业务逻辑处理器，能够在一个线程里每秒处理6百万订单。LMAX业务逻辑处理器完全是运行在内存中，使用事件驱动方式，其核心是Disruptor。
(2)Disruptor的特点
大大简化了并发程序开发的难度，性能上比Java提供的一些并发包还好。
Disruptor是一个高性能异步处理框架，实现了观察者模式。Disruptor是无锁的、是CPU友好的。Disruptor不会清除缓存中的数据，只会覆盖缓存中的数据，不需要进行垃圾回收。Disruptor业务逻辑是纯内存操作，使用事件驱动方式。
(3)Disruptor的核心
Disruptor核心是一个RingBuffer，RingBuffer是一个数组，没有首尾指针。RingBuffer是一个首尾相接的环，用于在不同线程之间传递数据。
如果RingBuffer满了，是继续覆盖还是等待消费，由生产者和消费者决定。假设RingBuffer满了，生产者有两个选择：选择一是等待RingBuffer有空位再填充，选择二是直接覆盖。同时消费者也有两种选择：选择一是等待RingBuffer满了再消费，选择二是RingBuffer填充一个就消费一个。
RingBuffer有一个序号Sequence，这个序号指向数组中下一个可用元素。随着数据不断地填充这个数组，这个序号会一直增长，直到绕过这个环。序号指向的元素，可以通过mod计算：序号 % 长度 = 索引。建议将长度设为2的n次方，有利于二进制计算：序号 & (长度 - 1) = 索引。
Sequence通过顺序递增的序号来进行编号，以此管理正在进行交换的数据(事件)。对数据处理的过程总是沿着需要逐个递增处理，从而实现线程安全。一个Sequence用于跟踪标识某个特定的事件处理者的处理进度。
2.Disruptor和BlockingQueue的压测对比
Disruptor的性能是ArrayBlockingQueue的3倍+，这里的测试代码都是基于单线程的单生产者单消费者模式运行的。但是Disruptor本身就支持多生产者多消费者模型，测试中使用单线程明显降低了其性能。而ArrayBlockingQueue在多生产者多消费者场景下，其性能又会比单生产者单消费者场景下更低。因此，在实际应用中，Disruptor的性能会是ArrayBlockingQueue的3倍+。
public interface Constants {
int EVENT_NUM_OHM = 100000000;
int EVENT_NUM_FM = 50000000;
int EVENT_NUM_OM = 10000000;
}
public class ArrayBlockingQueue4Test {
public static void main(String[] args) {
//初始化一个大小为100000000的有界队列ArrayBlockingQueue，为了避免在测试时由于扩容影响性能，所以一开始就初始化大小为1亿
final ArrayBlockingQueue<Data> queue = new ArrayBlockingQueue<Data>(100000000);
//开始时间
final long startTime = System.currentTimeMillis();
//向容器中添加元素
new Thread(new Runnable() {
public void run() {
long i = 0;
//首先把数据投递到有界队列ArrayBlockingQueue，单线程的生产者
while (i < Constants.EVENT_NUM_OHM) {
Data data = new Data(i, "c" + i);
try {
queue.put(data);
} catch (InterruptedException e) {
e.printStackTrace();
}
i++;
}
}
}).start();
//从容器中取出元素
new Thread(new Runnable() {
public void run() {
int k = 0;
//然后才开始消费有界队列中的数据，单线程的消费者
while (k < Constants.EVENT_NUM_OHM) {
try {
queue.take();
} catch (InterruptedException e) {
e.printStackTrace();
}
k++;
}
//结束时间
long endTime = System.currentTimeMillis();
//整个main函数就是单线程运行，处理1千万数据，大概耗时3.6秒
System.out.println("ArrayBlockingQueue costTime = " + (endTime - startTime) + "ms");
}
}).start();
}
}
public class DisruptorSingle4Test {
public static void main(String[] args) {
int ringBufferSize = 65536;
final Disruptor<Data> disruptor = new Disruptor<Data>(
new EventFactory<Data>() {
public Data newInstance() {
return new Data();
}
},
ringBufferSize,
//设置为单线程运行
Executors.newSingleThreadExecutor(),
//单生产者模式
ProducerType.SINGLE,
//new BlockingWaitStrategy()
new YieldingWaitStrategy()
);
//创建一个消费者事件处理器
DataConsumer consumer = new DataConsumer();
//消费数据
disruptor.handleEventsWith(consumer);
disruptor.start();
//单线程的消费者
new Thread(new Runnable() {
public void run() {
RingBuffer<Data> ringBuffer = disruptor.getRingBuffer();
for (long i = 0; i < Constants.EVENT_NUM_OHM; i++) {
long seq = ringBuffer.next();
Data data = ringBuffer.get(seq);
data.setId(i);
data.setName("c" + i);
//发布一个数据被消费的事件
ringBuffer.publish(seq);
}
}
}).start();
}
}
public class DataConsumer implements EventHandler<Data> {
private long startTime;
private int i;
public DataConsumer() {
this.startTime = System.currentTimeMillis();
}
public void onEvent(Data data, long seq, boolean bool) throws Exception {
i++;
if (i == Constants.EVENT_NUM_OHM) {
long endTime = System.currentTimeMillis();
//处理1千万的数据，大概耗时1.1秒
System.out.println("Disruptor costTime = " + (endTime - startTime) + "ms");
//可见Disruptor的性能是ArrayBlockingQueue的3倍+
}
}
}
3.Disruptor的编程模型
(1)Disruptor的使用步骤
(2)Disruptor的使用演示
(1)Disruptor的使用步骤
步骤一：建立一个Event工厂类，用于创建数据(Event类实例对象)
步骤二：建立一个监听事件类(Event处理器)，用于处理数据(Event类实例对象)
步骤三：创建Disruptor实例，配置一系列参数
步骤四：编写生产者组件，向Disruptor容器投递数据(2)Disruptor的使用演示
一.引入pom依赖
<dependency>
<groupId>com.lmax</groupId>
disruptor</artifactId>
<version>3.3.2</version>
</dependency>二.建立Event工厂类用于创建数据
Event工厂类创建的数据就是Event类实例对象。
public class OrderEvent {
//订单的价格
private long value;
public long getValue() {
return value;
}
public void setValue(long value) {
this.value = value;
}
}
public class OrderEventFactory implements EventFactory<OrderEvent> {
//返回一个空的数据对象(OrderEvent对象实例)
public OrderEvent newInstance() {
return new OrderEvent();
}
}三.建立监听事件类用于处理数据
监听事件类就是Event处理器，处理的数据就是Event类实例对象。
public class OrderEventHandler implements EventHandler<OrderEvent> {
public void onEvent(OrderEvent event, long sequence, boolean endOfBatch) throws Exception {
Thread.sleep(1000);
System.err.println("消费者: " + event.getValue());
}
}四.创建Disruptor对象实例
public class Main {
public static void main(String[] args) {
//参数准备
OrderEventFactory orderEventFactory = new OrderEventFactory();
int ringBufferSize = 4;
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
//参数一：eventFactory，消息(Event)工厂对象
//参数二：ringBufferSize，容器的长度
//参数三：executor，线程池(建议使用自定义线程池)，RejectedExecutionHandler
//参数四：ProducerType，单生产者还是多生产者
//参数五：waitStrategy，等待策略
//1.实例化Disruptor对象
Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(
orderEventFactory,
ringBufferSize,
executor,
ProducerType.SINGLE,//单生产者
new BlockingWaitStrategy()
);
//2.添加Event处理器，用于处理事件
//也就是构建Disruptor与消费者的一个关联关系
disruptor.handleEventsWith(new OrderEventHandler());
//3.启动disruptor
disruptor.start();
...
}
}五.编写生产者组件向Disruptor容器投递数据
public class Main {
public static void main(String[] args) {
//参数准备
OrderEventFactory orderEventFactory = new OrderEventFactory();
int ringBufferSize = 4;
ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
//参数一：eventFactory，消息(Event)工厂对象
//参数二：ringBufferSize，容器的长度
//参数三：executor，线程池(建议使用自定义线程池)，RejectedExecutionHandler
//参数四：ProducerType，单生产者还是多生产者
//参数五：waitStrategy，等待策略
//1.实例化Disruptor对象
Disruptor<OrderEvent> disruptor = new Disruptor<OrderEvent>(
orderEventFactory,
ringBufferSize,
executor,
ProducerType.SINGLE,
new BlockingWaitStrategy()
);
//2.添加Event处理器，用于处理事件
//也就是构建Disruptor与消费者的一个关联关系
disruptor.handleEventsWith(new OrderEventHandler());
//3.启动disruptor
disruptor.start();
//4.获取实际存储数据的容器: RingBuffer
RingBuffer<OrderEvent> ringBuffer = disruptor.getRingBuffer();
OrderEventProducer producer = new OrderEventProducer(ringBuffer);
ByteBuffer bb = ByteBuffer.allocate(8);
for (long i = 0; i < 5; i++) {
bb.putLong(0, i);
//向容器中投递数据
producer.sendData(bb);
}
disruptor.shutdown();
executor.shutdown();
}
}
public class OrderEventProducer {
private RingBuffer<OrderEvent> ringBuffer;
public OrderEventProducer(RingBuffer<OrderEvent> ringBuffer) {
this.ringBuffer = ringBuffer;
}
public void sendData(ByteBuffer data) {
//1.在生产者发送消息时, 首先需要从ringBuffer里获取一个可用的序号
long sequence = ringBuffer.next();
try {
//2.根据这个序号, 找到具体的"OrderEvent"元素
//注意：此时获取的OrderEvent对象是一个没有被赋值的"空对象"
OrderEvent event = ringBuffer.get(sequence);
//3.进行实际的赋值处理
event.setValue(data.getLong(0));
} finally {
//4.提交发布操作
ringBuffer.publish(sequence);
}
}
}
4.Disruptor的数据结构与生产消费模型
(1)Disruptor的核心与原理
(2)Disruptor的RingBuffer数据结构
(3)Disruptor的生产消费模型
(1)Disruptor的核心与原理
Disruptor的核心是RingBuffer，生产者向RingBuffer中写入元素，消费者从RingBuffer中消费元素。
(2)Disruptor的RingBuffer数据结构
RingBuffer是一个首尾相接的环(数组)，用于在不同上下文(线程)之间传递数据。
RingBuffer拥有一个序号，这个序号指向数组中下一个可用的元素。随着生产者不停地往RingBuffer写入元素，这个序号也会一直增长，直到这个序号绕过这个环。
要找到RingBuffer数组中当前序号指向的元素，可以通过mod操作：序号 % 数组长度 = 数组索引。建议将长度设为2的n次方，有利于二进制计算：序号 & (长度 - 1) = 索引。
(3)Disruptor的生产消费模型
一.消费快生产慢
如果消费者从RingBuffer消费元素的速度大于生产者写入元素的速度，那么当消费者发现RingBuffer没有元素时，就要停下等待生产者写入元素。
二.生产快消费慢
如果生产者向RingBuffer写入元素的速度大于消费者消费元素的速度，那么当生产者发现RingBuffer已经满了，就要停下等待消费者消费元素。
因为RingBuffer数组的长度是有限的，生产者写入到RingBuffer的末尾时，会从RingBuffer的开始位置继续写入，这时候生产者就可能会追上消费者。
5.RingBuffer + Disruptor + Sequence相关类
(1)RingBuffer类
(2)Disruptor类
(3)Sequence类
(4)Sequencer接口
(5)SequenceBarrier类
(1)RingBuffer类
RingBuffer不仅是基于数组的缓存，也是创建Sequencer与定义WaitStrategy的入口。
(2)Disruptor类
Disruptor类可认为是一个持有RingBuffer、消费者线程池、消费者集合等引用的辅助类。
(3)Sequence类
通过顺序递增的序号来编号，管理正在进行交换的数据(事件)。对数据(事件)的处理总是沿着序号逐个递增，所以能够实现多线程下的并发安全与原子性。
一个Sequence用于跟踪标识某个特定的事件处理者的处理进度，也就是事件处理者在RingBuffer中的处理进度。每一个Producer和Consumer都有一个自己的Sequence。
Sequence可以看成是一个AtomicLong类型字段，用于标识进度。Sequence还可以防止不同Sequence之间CPU缓存的伪共享问题。
Sequence的两个作用：
作用一：用于递增标识进度
作用二：用于消除伪共享
(4)Sequencer接口
一.Sequencer包含Sequence
二.Sequencer接口有两个实现类
第一个实现类是SingleProducerSequencer
第二个实现类是MultiProducerSequencer
(5)SequenceBarrier类
作用一：用于保持对RingBuffer的生产者和消费者之间的平衡关系，比如让生产者或消费者进行等待、唤醒生产者或消费者
作用二：决定消费者是否还有可处理的事件
6.Disruptor的WaitStrategy消费者等待策略
(1)WaitStrategy接口的作用
(2)消费者等待策略的种类
(3)BlockingWaitStrategy
(4)SleepingWaitStrategy
(5)YieldingWaitStrategy
(1)WaitStrategy接口的作用
决定一个消费者将会如何等待生产者将Event投递到Disruptor。
(2)消费者等待策略的种类
BlockingWaitStrategy，通过阻塞的方式进行等待
SleepingWaitStrategy，通过休眠的方式进行等待
YieldingWaitStrategy，通过线程间的切换的方式进行等待(3)BlockingWaitStrategy
BlockingWaitStrategy是最低效的等待策略，但是对CPU的消耗最小，并且在各种不同部署环境中能提供一致的性能表现。该策略需要使用到Java中的锁，也就是会通过ReentrantLock来阻塞消费者线程。而Disruptor本身是一个无锁并发框架，所以如果追求高性能，就不要选择这种策略。
(4)SleepingWaitStrategy
SleepingWaitStrategy是性能一般的等待策略，其性能表现和BlockingWaitStrategy差不多。但由于SleepingWaitStrategy是无锁的，所以对生产者线程的影响最小。该策略对CPU的消耗一般，通过在单个线程循环 + yield切换线程实现，所以这种策略特别适合于异步日志类似的场景。
(5)YieldingWaitStrategy
YieldingWaitStrategy的性能是最好的，适合于低延迟的系统。不过该策略对CPU的消耗最高，因为完全基于yield切换线程来实现。推荐用于要求高性能且事件处理线程数小于CPU逻辑核心数的场景中，尤其是当CPU开启了超线程特性的时候。
7.EventProcessor + EventHandler等类
(1)Event对象
(2)EventProcessor接口
(3)EventHandler接口
(4)WorkProcessor类
(1)Event对象
Disruptor中的Event指的是从生产者到消费者过程中所处理的数据对象。Disruptor中没有代码表示Event，它用泛型表示，完全由用户定义。比如创建一个RingBuffer对象时，其中的泛型就表示着这个Event对象。
(2)EventProcessor接口
EventProcessor用于处理Disruptor中的Event，拥有消费者的Sequence，它有一个实现类叫BatchEventProcessor。
由于EventProcessor接口继承自Runnable接口，所以BatchEventProcessor类会实现Runnable接口的run()方法。
其实BatchEventProcessor类是Disruptor框架中最核心的类，因为它的run()方法会不断轮询并获取数据对象，然后把数据对象(Event)交给消费者去处理，也就是即回调EventHandler接口的实现类对象的onEvent()方法。
(3)EventHandler接口
EventHandler是由用户实现的并且代表了Disruptor中的一个消费者接口，也就是消费者逻辑需要在EventHandler接口的onEvent()方法实现。
(4)WorkProcessor类
WorkProcessor类可确保每个Sequence只被一个Processor消费。注意：在单消费者模式下，使用的是EventHandler，对应于EventProcessor。在多消费者模式下，使用的是WorkHandler，对应于WorkProcessor。
8.Disruptor的运行原理图
9.复杂业务需求下的编码方案和框架
(1)方案选择
(2)框架选择
(1)方案选择
方案一：完全解耦的模式，比如一个子业务线也开一个项目，此时重复代码会比较多。
方案二：模版方法模式，如果业务快速迭代，可能也会需要经常重构底层的模版方法。
(2)框架选择
一.使用有限状态机框架
二.使用Disruptor框架
10.Disruptor的串行操作
Disruptor的串行操作，可以通过链式调用handleEventsWith()方法来实现。
如果使用RingBuffer对象来发布事件，那么需要先从RingBuffer对象中获取一个可用的序号，然后根据序号获取Event对象并对Event对象赋值，最后调用RingBuffer的publish()方法发布事件。
如果使用Disruptor对象来发布事件，那么直接调用Disruptor的publishEvent()方法发布事件即可。
此外，实际应用中不建议通过Executors来创建线程池，而应通过ThreadPoolExecutor构造函数具体指定线程池的每一个参数。因为Executors创建的线程池还是可能有安全隐患，比如Executors的newFixedThreadPool()方法使用的是无界队列，其使用的LinkedBlockingQueue是一个可选是否有界的阻塞队列。
//Disruptor中的Event
public class Trade {
private String id;
private String name;
private double price;
private AtomicInteger count = new AtomicInteger(0);
public Trade() {
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public double getPrice() {
return price;
}
public void setPrice(double price) {
this.price = price;
}
public AtomicInteger getCount() {
return count;
}
public void setCount(AtomicInteger count) {
this.count = count;
}
}
public class Main {
@SuppressWarnings("unchecked")
public static void main(String[] args) throws Exception {
//实际应用中不建议这样创建线程池，而应通过ThreadPoolExecutor构造函数具体指定每个参数
//因为这种创建的线程池还是有安全隐患，比如newFixedThreadPool()使用的是无界队列
//LinkedBlockingQueue是一个可选是否有界的阻塞队列
ExecutorService es1 = Executors.newFixedThreadPool(8);
//构建一个线程池用于提交任务
ExecutorService es2 = Executors.newFixedThreadPool(1);
//1.构建Disruptor
Disruptor<Trade> disruptor = new Disruptor<Trade>(
new EventFactory<Trade>() {
public Trade newInstance() {
return new Trade();
}
},
1024 * 1024,
es1,
ProducerType.SINGLE,
new BusySpinWaitStrategy()
);
//2.把消费者设置到Disruptor中，也就是使用Disruptor.handleEventsWith()方法
//串行操作，通过链式编程实现
disruptor.handleEventsWith(new Handler1())
.handleEventsWith(new Handler2())
.handleEventsWith(new Handler3());
//3.启动disruptor并获取RingBuffer
RingBuffer<Trade> ringBuffer = disruptor.start();
CountDownLatch latch = new CountDownLatch(1);
long begin = System.currentTimeMillis();
//通过线程池向Disruptor发布事件(生产数据)
es2.submit(new TradePublisher(latch, disruptor));
latch.await();
disruptor.shutdown();
es1.shutdown();
es2.shutdown();
System.err.println("总耗时: " + (System.currentTimeMillis() - begin));
}
}
public class TradePublisher implements Runnable {
private static int PUBLISH_COUNT = 10;
private Disruptor<Trade> disruptor;
private CountDownLatch latch;
public TradePublisher(CountDownLatch latch, Disruptor<Trade> disruptor) {
this.disruptor = disruptor;
this.latch = latch;
}
public void run() {
TradeEventTranslator eventTranslator = new TradeEventTranslator();
for (int i = 0; i < PUBLISH_COUNT; i++) {
//新的发布事件的方式，另一种方式就是通过传入的RingBuffer的publish()方法发布事件
disruptor.publishEvent(eventTranslator);
}
latch.countDown();
}
}
class TradeEventTranslator implements EventTranslator<Trade> {
private Random random = new Random();
public void translateTo(Trade event, long sequence) {
this.generateTrade(event);
}
private void generateTrade(Trade event) {
event.setPrice(random.nextDouble() * 9999);
}
}
public class Handler1 implements EventHandler<Trade>, WorkHandler<Trade> {
//实现EventHandler的onEvent()方法，可以监听生产者发布的事件
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
this.onEvent(event);
}
//实现WorkHandler的onEvent()方法，也可以监听生产者发布的事件
public void onEvent(Trade event) throws Exception {
System.err.println("handler 1 : SET NAME");
Thread.sleep(1000);
event.setName("H1");
}
}
public class Handler2 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 2 : SET ID");
Thread.sleep(2000);
event.setId(UUID.randomUUID().toString());
}
}
public class Handler3 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 3 : NAME: " + event.getName() + ", ID: " + event.getId() + ", PRICE: " + event.getPrice() + " INSTANCE : " + event.toString());
}
}
11.Disruptor的并行操作
Disruptor的并行操作可以有两种方式实现：方式一是调用handleEventsWith()方法时传入多个handler对象，方式二是分别多次调用handleEventsWith()方法。
public class Main {
@SuppressWarnings("unchecked")
public static void main(String[] args) throws Exception {
//实际应用中不建议这样创建线程池，而应通过ThreadPoolExecutor构造函数具体指定每个参数
//因为这种创建的线程池还是有安全隐患，比如newFixedThreadPool()使用的是无界队列
//LinkedBlockingQueue是一个可选是否有界的阻塞队列
ExecutorService es1 = Executors.newFixedThreadPool(8);
//构建一个线程池用于提交任务
ExecutorService es2 = Executors.newFixedThreadPool(1);
//1.构建Disruptor
Disruptor<Trade> disruptor = new Disruptor<Trade>(
new EventFactory<Trade>() {
public Trade newInstance() {
return new Trade();
}
},
1024 * 1024,
es1,
ProducerType.SINGLE,
new BusySpinWaitStrategy()
);
//2.把消费者设置到Disruptor中，也就是使用Disruptor.handleEventsWith()方法
//Disruptor的并行操作可以有两种方式实现
//方式一：调用handleEventsWith方法时传入多个handler对象
disruptor.handleEventsWith(new Handler1(), new Handler2(), new Handler3());
//方式二：分别多次调用handleEventsWith()方法
//disruptor.handleEventsWith(new Handler1());
//disruptor.handleEventsWith(new Handler2());
//disruptor.handleEventsWith(new Handler3());
//3.启动disruptor并获取RingBuffer
RingBuffer<Trade> ringBuffer = disruptor.start();
CountDownLatch latch = new CountDownLatch(1);
long begin = System.currentTimeMillis();
//通过线程池向Disruptor发布事件(生产数据)
es2.submit(new TradePublisher(latch, disruptor));
latch.await();
disruptor.shutdown();
es1.shutdown();
es2.shutdown();
System.err.println("总耗时: " + (System.currentTimeMillis() - begin));
}
}
public class TradePublisher implements Runnable {
private static int PUBLISH_COUNT = 10;
private Disruptor<Trade> disruptor;
private CountDownLatch latch;
public TradePublisher(CountDownLatch latch, Disruptor<Trade> disruptor) {
this.disruptor = disruptor;
this.latch = latch;
}
public void run() {
TradeEventTranslator eventTranslator = new TradeEventTranslator();
for (int i = 0; i < PUBLISH_COUNT; i++) {
//新的发布事件的方式，另一种方式就是通过传入的RingBuffer的publish()方法发布事件
disruptor.publishEvent(eventTranslator);
}
latch.countDown();
}
}
class TradeEventTranslator implements EventTranslator<Trade> {
private Random random = new Random();
public void translateTo(Trade event, long sequence) {
this.generateTrade(event);
}
private void generateTrade(Trade event) {
event.setPrice(random.nextDouble() * 9999);
}
}
public class Handler1 implements EventHandler<Trade>, WorkHandler<Trade> {
//实现EventHandler的onEvent()方法，可以监听生产者发布的事件
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
this.onEvent(event);
}
//实现WorkHandler的onEvent()方法，也可以监听生产者发布的事件
public void onEvent(Trade event) throws Exception {
System.err.println("handler 1 : SET NAME");
Thread.sleep(1000);
event.setName("H1");
}
}
public class Handler2 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 2 : SET ID");
Thread.sleep(2000);
event.setId(UUID.randomUUID().toString());
}
}
public class Handler3 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 3 : NAME: " + event.getName() + ", ID: " + event.getId() + ", PRICE: " + event.getPrice() + " INSTANCE : " + event.toString());
}
}
12.Disruptor的多边形操作
(1)Disruptor的菱形操作
(2)Disruptor的六边形操作
Disruptor可以实现串并行同时编码。
(1)Disruptor的菱形操作
可以理解为先并行执行，然后再串行执行，类似于CyclicBarrier。
菱形操作方式一：调用handleEventsWith()方法时传入多个参数 + 链式调用。
菱形操作方式二：调用handleEventsWith()方法时传入多个参数 + 使用then()方法。
public class Main {
@SuppressWarnings("unchecked")
public static void main(String[] args) throws Exception {
//实际应用中不建议这样创建线程池，而应通过ThreadPoolExecutor构造函数具体指定每个参数
//因为这种创建的线程池还是有安全隐患，比如newFixedThreadPool()使用的是无界队列
//LinkedBlockingQueue是一个可选是否有界的阻塞队列
ExecutorService es1 = Executors.newFixedThreadPool(8);
//构建一个线程池用于提交任务
ExecutorService es2 = Executors.newFixedThreadPool(1);
//1.构建Disruptor
Disruptor<Trade> disruptor = new Disruptor<Trade>(
new EventFactory<Trade>() {
public Trade newInstance() {
return new Trade();
}
},
1024 * 1024,
es1,
ProducerType.SINGLE,
new BusySpinWaitStrategy()
);
//2.把消费者设置到Disruptor中，也就是使用Disruptor.handleEventsWith()方法
//菱形操作一
disruptor.handleEventsWith(new Handler1(), new Handler2())
.handleEventsWith(new Handler3());
//菱形操作二
//EventHandlerGroup<Trade> ehGroup = disruptor.handleEventsWith(new Handler1(), new Handler2());
//ehGroup.then(new Handler3());
//3.启动disruptor并获取RingBuffer
RingBuffer<Trade> ringBuffer = disruptor.start();
CountDownLatch latch = new CountDownLatch(1);
long begin = System.currentTimeMillis();
//通过线程池向Disruptor发布事件(生产数据)
es2.submit(new TradePublisher(latch, disruptor));
latch.await();
disruptor.shutdown();
es1.shutdown();
es2.shutdown();
System.err.println("总耗时: " + (System.currentTimeMillis() - begin));
}
}
public class TradePublisher implements Runnable {
private static int PUBLISH_COUNT = 10;
private Disruptor<Trade> disruptor;
private CountDownLatch latch;
public TradePublisher(CountDownLatch latch, Disruptor<Trade> disruptor) {
this.disruptor = disruptor;
this.latch = latch;
}
public void run() {
TradeEventTranslator eventTranslator = new TradeEventTranslator();
for (int i = 0; i < PUBLISH_COUNT; i++) {
//新的发布事件的方式，另一种方式就是通过传入的RingBuffer的publish()方法发布事件
disruptor.publishEvent(eventTranslator);
}
latch.countDown();
}
}
class TradeEventTranslator implements EventTranslator<Trade> {
private Random random = new Random();
public void translateTo(Trade event, long sequence) {
this.generateTrade(event);
}
private void generateTrade(Trade event) {
event.setPrice(random.nextDouble() * 9999);
}
}
public class Handler1 implements EventHandler<Trade>, WorkHandler<Trade> {
//实现EventHandler的onEvent()方法，可以监听生产者发布的事件
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
this.onEvent(event);
}
//实现WorkHandler的onEvent()方法，也可以监听生产者发布的事件
public void onEvent(Trade event) throws Exception {
System.err.println("handler 1 : SET NAME");
Thread.sleep(1000);
event.setName("H1");
}
}
public class Handler2 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 2 : SET ID");
Thread.sleep(2000);
event.setId(UUID.randomUUID().toString());
}
}
public class Handler3 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 3 : NAME: " + event.getName() + ", ID: " + event.getId() + ", PRICE: " + event.getPrice() + " INSTANCE : " + event.toString());
}
}
(2)Disruptor的六边形操作
通过Disruptor的after()方法 + 菱形操作，可实现六边形操作。
注意在单消费者模式下：一个EventHandler会对应一个BatchEventProcessor，所以如果有n个EventHandler监听Disruptor，那么初始化Disruptor时的线程池就要有n个线程，否则可能导致多边形操作失效。
在单消费者模式下，如果有非常多EventHandler，就需要非常多线程。此时是不合理的，所以如果有很多EventHandler，可采用多消费者模式。
public class Main {
@SuppressWarnings("unchecked")
public static void main(String[] args) throws Exception {
//实际应用中不建议这样创建线程池，而应通过ThreadPoolExecutor构造函数具体指定每个参数
//因为这种创建的线程池还是有安全隐患，比如newFixedThreadPool()使用的是无界队列
//LinkedBlockingQueue是一个可选是否有界的阻塞队列
ExecutorService es1 = Executors.newFixedThreadPool(8);
//构建一个线程池用于提交任务
ExecutorService es2 = Executors.newFixedThreadPool(1);
//1.构建Disruptor
Disruptor<Trade> disruptor = new Disruptor<Trade>(
new EventFactory<Trade>() {
public Trade newInstance() {
return new Trade();
}
},
1024 * 1024,
es1,
ProducerType.SINGLE,
new BusySpinWaitStrategy()
);
//2.把消费者设置到Disruptor中，也就是使用Disruptor.handleEventsWith()方法
//六边形操作
Handler1 h1 = new Handler1();
Handler2 h2 = new Handler2();
Handler3 h3 = new Handler3();
Handler4 h4 = new Handler4();
Handler5 h5 = new Handler5();
disruptor.handleEventsWith(h1, h4);
disruptor.after(h1).handleEventsWith(h2);
disruptor.after(h4).handleEventsWith(h5);
disruptor.after(h2, h5).handleEventsWith(h3);
//3.启动disruptor并获取RingBuffer
RingBuffer<Trade> ringBuffer = disruptor.start();
CountDownLatch latch = new CountDownLatch(1);
long begin = System.currentTimeMillis();
//通过线程池向Disruptor发布事件(生产数据)
es2.submit(new TradePublisher(latch, disruptor));
latch.await();
disruptor.shutdown();
es1.shutdown();
es2.shutdown();
System.err.println("总耗时: " + (System.currentTimeMillis() - begin));
}
}
public class TradePublisher implements Runnable {
private static int PUBLISH_COUNT = 10;
private Disruptor<Trade> disruptor;
private CountDownLatch latch;
public TradePublisher(CountDownLatch latch, Disruptor<Trade> disruptor) {
this.disruptor = disruptor;
this.latch = latch;
}
public void run() {
TradeEventTranslator eventTranslator = new TradeEventTranslator();
for (int i = 0; i < PUBLISH_COUNT; i++) {
//新的发布事件的方式，另一种方式就是通过传入的RingBuffer的publish()方法发布事件
disruptor.publishEvent(eventTranslator);
}
latch.countDown();
}
}
class TradeEventTranslator implements EventTranslator<Trade> {
private Random random = new Random();
public void translateTo(Trade event, long sequence) {
this.generateTrade(event);
}
private void generateTrade(Trade event) {
event.setPrice(random.nextDouble() * 9999);
}
}
public class Handler1 implements EventHandler<Trade>, WorkHandler<Trade> {
//实现EventHandler的onEvent()方法，可以监听生产者发布的事件
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
this.onEvent(event);
}
//实现WorkHandler的onEvent()方法，也可以监听生产者发布的事件
public void onEvent(Trade event) throws Exception {
System.err.println("handler 1 : SET NAME");
Thread.sleep(1000);
event.setName("H1");
}
}
public class Handler2 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 2 : SET ID");
Thread.sleep(2000);
event.setId(UUID.randomUUID().toString());
}
}
public class Handler3 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 3 : NAME: " + event.getName() + ", ID: " + event.getId() + ", PRICE: " + event.getPrice() + " INSTANCE : " + event.toString());
}
}
public class Handler4 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 4 : SET PRICE");
Thread.sleep(1000);
event.setPrice(17.0);
}
}
public class Handler5 implements EventHandler<Trade> {
public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {
System.err.println("handler 5 : GET PRICE: " + event.getPrice());
Thread.sleep(1000);
event.setPrice(event.getPrice() + 3.0);
}
}
13.Disruptor的多生产者和多消费者
注意一：使用多消费者模式时，每个消费者都需要实现WorkHandler接口，而不是EventHandler接口。单消费者模式，使用的是EventHandler，对应于EventProcessor。多消费者模式，使用的是WorkHandler，对应于WorkProcessor。
注意二：使用多消费者模式时，需要构建消费者工作池WorkerPool。
注意三：使用多消费者模式时，每个消费者需要一个Sequence来标记当前消费的最小序号。这样生产者投递消息时才能遍历消费者的Sequence找出最小的序号，然后写到最小的序号位置进行阻塞等待。
比如下图中，在某一时刻：消费者1消费了序号0和2，但序号1还没有消费完毕。消费者2消费了序号3和4，消费者3消费了序号5。此时，在RingBuffer中，虽然序号0、2、3、4、5都可以覆盖了，但由于序号1还没被消费，所以生产者最多只能覆盖到序号0的位置。然后等待序号1被消费者1消费完毕后，才能继续往RingBuffer投递消息。
//Disruptor中的 Event
public class Order {
private String id;
private String name;
private double price;
public Order() {
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public double getPrice() {
return price;
}
public void setPrice(double price) {
this.price = price;
}
}
public class Main {
public static void main(String[] args) throws InterruptedException {
//1.创建RingBuffer
RingBuffer<Order> ringBuffer = RingBuffer.create(
ProducerType.MULTI,//多生产者
new EventFactory<Order>() {
public Order newInstance() {
return new Order();
}
},
1024 * 1024,
new YieldingWaitStrategy()
);
//2.通过ringBuffer创建一个屏障
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
//3.创建消费者数组，每个消费者Consumer都需要实现WorkHandler接口
Consumer[] consumers = new Consumer;
for (int i = 0; i < consumers.length; i++) {
consumers = new Consumer("C" + i);
}
//4.构建多消费者工作池WorkerPool，因为多消费者模式下需要使用WorkerPool
WorkerPool<Order> workerPool = new WorkerPool<Order>(
ringBuffer,
sequenceBarrier,
new EventExceptionHandler(),
consumers
);
//5.设置多个消费者的sequence序号，用于单独统计每个消费者的消费进度, 并且设置到RingBuffer中
ringBuffer.addGatingSequences(workerPool.getWorkerSequences());
//6.启动workerPool
workerPool.start(Executors.newFixedThreadPool(5));
final CountDownLatch latch = new CountDownLatch(1);
for (int i = 0; i < 100; i++) {
final Producer producer = new Producer(ringBuffer);
new Thread(new Runnable() {
public void run() {
try {
latch.await();
} catch (Exception e) {
e.printStackTrace();
}
for (int j = 0; j < 100; j++) {
producer.sendData(UUID.randomUUID().toString());
}
}
}).start();
}
Thread.sleep(2000);
System.err.println("----------等待线程创建完毕，才开始生产数据----------");
latch.countDown();
Thread.sleep(10000);
System.err.println("任务总数:" + consumers.getCount());
}
static class EventExceptionHandler implements ExceptionHandler<Order> {
public void handleEventException(Throwable ex, long sequence, Order event) {
}
public void handleOnStartException(Throwable ex) {
}
public void handleOnShutdownException(Throwable ex) {
}
}
}
public class Consumer implements WorkHandler<Order> {
private static AtomicInteger count = new AtomicInteger(0);
private String consumerId;
private Random random = new Random();
public Consumer(String consumerId) {
this.consumerId = consumerId;
}
public void onEvent(Order event) throws Exception {
Thread.sleep(1 * random.nextInt(5));
System.err.println("当前消费者: " + this.consumerId + ", 消费信息ID: " + event.getId());
count.incrementAndGet();
}
public int getCount() {
return count.get();
}
}
public class Producer {
private RingBuffer<Order> ringBuffer;
public Producer(RingBuffer<Order> ringBuffer) {
this.ringBuffer = ringBuffer;
}
public void sendData(String uuid) {
long sequence = ringBuffer.next();
try {
Order order = ringBuffer.get(sequence);
order.setId(uuid);
} finally {
ringBuffer.publish(sequence);
}
}
}
来源：程序园用户自行投稿发布，如果侵权，请联系站长删除
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！
页:
[1]