Disruptor系列3：Disruptor样例实战

本文主要是介绍Disruptor系列3：Disruptor样例实战，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

章节回顾：
- Disruptor系列1：初识Disruptor
- Disruptor系列2：Disruptor原理剖析

本章节是Disruptor样例实战，依据Disruptor的工作流依次执行的特性，实现各种样例。如果想了解Disruptor是什么，可以查看章节 Disruptor系列1：初识Disruptor ，如果想深层次了解Disruptor，可以查看章节 Disruptor系列2：Disruptor原理剖析。通过本章节，希望让大家对如何使用Disruptor有个初步认识，看看它能够解决哪些情况。

具体而言，它可以解决如下方面：
- 并行计算实现；
- 串行依次执行；
- 菱形方式执行；
- 链式并行计算。
并且基于以上情况，每种类型的消费者都可以池化，默认初始化多个同一类型的消费者实例，并行处理，提高系统吞吐量。

本样例是一个生产者生产一个Long类型的数值，消费者对该数值进行处理的操作。本样例对以上各种情况的实现只是disruptor注册消费者的方式不同，因此，我们先把事件类、事件工厂类、消费者类、事件转换类和主函数贴出来。

事件类

public class LongEvent {private Long number;public Long getNumber() {return number;}public void setNumber(Long number) {this.number = number;}
}

事件工厂类

public class LongEventFactory implements EventFactory<LongEvent> {@Overridepublic LongEvent newInstance() {return new LongEvent();}
}

事件转换类

public class LongEventTranslator implements EventTranslatorOneArg<LongEvent, Long> {@Overridepublic void translateTo(LongEvent event, long sequence, Long arg0) {event.setNumber(arg0);}
}

C1-1消费者类

该消费者执行将数值+10的操作。可以看到该消费者同时实现了EventHandler和WorkHandler两个接口。如果不需要池化，只需要实现EventHandler类即可。如果需要池化，只需要实现WorkHandler类即可。本例为了能够同时讲解池化和非池化的实现，因此同时实现了两个类，当然，也没啥问题。

public class C11EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number += 10;System.out.println(System.currentTimeMillis()+": c1-1 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number += 10;System.out.println(System.currentTimeMillis()+": c1-1 consumer finished.number=" + number);}
}

C1-2消费者类

该消费者类执行将数值乘以10的操作。

public class C12EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number *= 10;System.out.println(System.currentTimeMillis()+": c1-2 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number *= 10;System.out.println(System.currentTimeMillis()+": c1-2 consumer finished.number=" + number);}
}

c2-1消费者类

该消费者类负责将数值+20.

public class C21EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number += 20;System.out.println(System.currentTimeMillis()+": c2-1 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number += 20;System.out.println(System.currentTimeMillis()+": c2-1 consumer finished.number=" + number);}
}

C2-2消费者类

该消费者类负责将数值*20

public class C22EventHandler implements EventHandler<LongEvent>,WorkHandler<LongEvent> {@Overridepublic void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {long number = event.getNumber();number *= 20;System.out.println(System.currentTimeMillis()+": c2-2 consumer finished.number=" + number);}@Overridepublic void onEvent(LongEvent event) throws Exception {long number = event.getNumber();number *= 20;System.out.println(System.currentTimeMillis()+": c2-2 consumer finished.number=" + number);}
}

主函数

public class Main {public static void main(String[] args) {int bufferSize = 1024*1024;//环形队列长度，必须是2的N次方EventFactory<LongEvent> eventFactory = new LongEventFactory();/*** 定义Disruptor，基于单生产者，阻塞策略*/Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory,bufferSize, Executors.defaultThreadFactory(),ProducerType.SINGLE,new BlockingWaitStrategy());/////////////////////////////////////////////////////////////////////XXX(disruptor);//这里是调用各种不同方法的地方./////////////////////////////////////////////////////////////////////RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();/*** 输入10*/ringBuffer.publishEvent(new LongEventTranslator(),10L);ringBuffer.publishEvent(new LongEventTranslator(),100L);}
}

并行计算实现

并行计算就是消费者之间互相不依赖，并行执行，执行开始时间是一样的。

/*** 并行计算实现,c1,c2互相不依赖* <br/>* p --> c11*   --> c21*/public static void parallel(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler(),new C21EventHandler());disruptor.start();}

串行计算，依次执行

/*** 串行依次执行* <br/>* p --> c11 --> c21* @param disruptor*/public static void serial(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler()).then(new C21EventHandler());disruptor.start();}

菱形方式执行

/*** 菱形方式执行* <br/>*   --> c11* p          --> c21*   --> c12* @param disruptor*/public static void diamond(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler(),new C12EventHandler()).then(new C21EventHandler());disruptor.start();}

链式并行计算

/*** 链式并行计算* <br/>*   --> c11 --> c12* p*   --> c21 --> c22* @param disruptor*/public static void chain(Disruptor<LongEvent> disruptor){disruptor.handleEventsWith(new C11EventHandler()).then(new C12EventHandler());disruptor.handleEventsWith(new C21EventHandler()).then(new C22EventHandler());disruptor.start();}

上面的实例，每一种消费者都只有一个实例，如果想多个实例形成一个线程池并发处理多个任务怎么办？如果使用disruptor.handleEventWith(new C11EventHandler(),new C11EventHandler(),...)这种，会造成重复消费同一个数据，不是我们想要的。我们想要的是同一个类的实例消费不同的数据，怎么办？
- 首先，消费者类需要实现WorkHandler接口，而不是EventHandler接口。为了方便，我们同时实现了这两个接口。
- 其次，disruptor调用handleEventsWithWorkerPool方法，而不是handleEventsWith方法
- 最后，实例化多个事件消费类。

并行计算实现,c1,c2互相不依赖,同时C1，C2分别有2个实例

/*** 并行计算实现,c1,c2互相不依赖,同时C1，C2分别有2个实例* <br/>* p --> c11*   --> c21*/public static void parallelWithPool(Disruptor<LongEvent> disruptor){disruptor.handleEventsWithWorkerPool(new C11EventHandler(),new C11EventHandler());disruptor.handleEventsWithWorkerPool(new C21EventHandler(),new C21EventHandler());disruptor.start();}

串行依次执行,同时C11，C21分别有2个实例

/*** 串行依次执行,同时C11，C21分别有2个实例* <br/>* p --> c11 --> c21* @param disruptor*/public static void serialWithPool(Disruptor<LongEvent> disruptor){disruptor.handleEventsWithWorkerPool(new C11EventHandler(),new C11EventHandler()).then(new C21EventHandler(),new C21EventHandler());disruptor.start();}

这篇关于Disruptor系列3：Disruptor样例实战的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---