高性能队列Disruptor在测试中应用
高性能队列Disruptor在测试中应用
最近在研究goreplay的源码的过程中，感觉有些思路还是很值得借鉴。所以自己立了一个flag，实现一个千万级日志回放功能。但是在这个实现的过程中遇到一个棘手的问题：Java自带的LinkedBlockingQueue比较难以直接满足需求场景和性能要求。
熟悉goreplay的测友应该清楚Go语言chanel在goreplay这个框架中应用是十分广泛的，加上Go语言自身较高的性能，可以说双剑合并。所以我也想照葫芦画瓢写一个类似思路的实现。这个后面会有专题讲这个。
基于此，我搜到了Disruptor这个高性能队列。Disruptor是英国外汇交易公司LMAX开发的一个高性能队列，研发的初衷是解决内存队列的延迟问题（在性能测试中发现竟然与I/O操作处于同样的数量级）。基于Disruptor开发的系统单线程能支撑每秒600万订单。
测试使用Disruptor时候不用像Springboot框架中那样，创建各类对象，抽象各种对象方法，我的原则就是怎么简单怎么来，下面分享一下Disruptor在测试中的基础实践和简单案例演示。
依赖
// https://mvnrepository.com/artifact/com.lmax/disruptor
implementation group: 'com.lmax', name: 'disruptor', version: '3.4.2'只列出了Gradle的，版本建议使用3+，有些Lambda语法支持需要。
Event对象
首先我们要定义一个Event类型，当然也可以直接使用类似java.lang.String使用已经存在的类，但是在设置Event对象时候，需要使用new关键字以及构造新的Event时，使用set方法总比直接=赋值的一致性更好一些。单独写一个Event类，可以更加简单，是的代码逻辑性更强，不用收到其他类的影响。
这里我定义了简单的Event类：
public static class FunEvent {
String id;
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
}
Disruptor创建
Disruptor对象创建首先需要我们定义一个Event类型，调用构造方法，参数有五个。1.Event初始化方法；2.ringbuffsize，这里设置需要是2的整数倍；3.threadfactory，创建线程的工厂类，这里我用了com.funtester.frame.execute.ThreadPoolUtil#getFactory()；4.生产者模式，分成单生产者和多生产者枚举类；5.等待策略，官方提供了一些实现类供选择，我使用了com.lmax.disruptor.YieldingWaitStrategy。
创建方法如下：
Disruptor<FunEvent> disruptor = new Disruptor<FunEvent>(
FunEvent::new,
1024 * 1024,
ThreadPoolUtil.getFactory(),
ProducerType.MULTI,
new YieldingWaitStrategy()
);生产者
对于消息队列来讲，需要两个重要的角色，生产者和消费者。这里先将一下Disruptor生产者，我搜到不少资料，都是需要创建一个生产者的类，然后实现一个方法，这个方法内容基本一致的，内容如下：
long sequence = ringBuffer.next();
try {
FunEvent funEvent = ringBuffer.get(sequence);
funEvent.setId(orderId);
} finally {
ringBuffer.publish(sequence);
}然后使用生产者对象调用这个方法，我觉得有点多此一举了，幸好有一篇文章介绍了Disruptor一些新特性的时候提到支持了Lambda语法，这下就可以不用创建生产者对象了。语法如下：
ringBuffer.publishEvent((Event, sequence) -> Event.setId(StringUtil.getString(10)));消费者
消费者创建需要实现两个接口com.lmax.disruptor.EventHandler和com.lmax.disruptor.WorkHandler，这俩一个是处理单消费者模式，另外一个多消费者模式。
创建方法如下：
/**
* 消费者
*/
private static class FunEventHandler implements EventHandler<FunEvent>, WorkHandler<FunEvent> {
public void onEvent(FunEvent Event, long sequence, boolean endOfBatch) {
output("消费消息:" + Event.getId() + TAB + sequence);
}
public void onEvent(FunEvent Event) {
output("消费消息:" + Event.getId());
}
}配置handler
这里分两类：配置单个消费者和配置多个消费者。
单个消费者：
disruptor.handleEventsWith(new FunEventHandler());
多个消费者：
disruptor.handleEventsWithWorkerPool(new FunEventHandler(), new FunEventHandler());
不管是单个还是多个，每次调用都会产生一个com.lmax.disruptor.dsl.EventHandlerGroup，每个com.lmax.disruptor.dsl.EventHandlerGroup都会完整消费每一个生产者产生的Event，如果设置了5次，那么一个Event就会被消费5次，每个com.lmax.disruptor.dsl.EventHandlerGroup消费一次，而且是阻塞的，加入某一个com.lmax.disruptor.dsl.EventHandlerGroup对象消费慢了，会阻塞其他消费者消费下一个Event。
启动
组装完成之后就可以启动了Disruptor了，语法如下：disruptor.start();，关闭语法disruptor.shutdown();，此处的关闭不会清空getRingBuffer已经存在的Event，看官方文档应该是停止生产，然后等待消费。
演示Demo
Java版本
public static void main(String[] args) {
Disruptor<FunEvent> disruptor = new Disruptor<FunEvent>(
FunEvent::new,
1024 * 1024,
ThreadPoolUtil.getFactory(),
ProducerType.MULTI,
new YieldingWaitStrategy()
);
disruptor.handleEventsWithWorkerPool(new FunEventHandler(), new FunEventHandler());
disruptor.handleEventsWith(new FunEventHandler());
disruptor.start();
RingBuffer<FunEvent> ringBuffer = disruptor.getRingBuffer();
for (int i = 0; i < 3; i++) {
ringBuffer.publishEvent((event, sequence) -> event.setId(StringUtil.getString(10)));
}
sleep(5.0);
disruptor.shutdown();
}控制台输出：
INFO-> main 当前用户：oker，工作目录：/Users/oker/IdeaProjects/funtester/,系统编码格式:UTF-8,系统Mac OS X版本:10.16
INFO-> main
###### # # # # ####### ###### ##### ####### ###### #####
# # # ## # # # # # # # #
#### # # # # # # #### ##### # #### #####
# # # # # # # # # # # # #
# ##### # # # ###### ##### # ###### # #
INFO-> F-3 消费消息:i3OrH2ZnxD 0
INFO-> F-1 消费消息:i3OrH2ZnxD
INFO-> F-2 消费消息:whhoxoMxmR
INFO-> F-3 消费消息:whhoxoMxmR 1
INFO-> F-2 消费消息:IeP9fIRpKp
INFO-> F-3 消费消息:IeP9fIRpKp 2
Process finished with exit code 0可以看到，每个消息会消费了两次。其中F-3线程消费量=F-1和F-2线程消费量总和，这就跟家理解了com.lmax.disruptor.dsl.EventHandlerGroup的功能。
Groovy+异步版本
public static void main(String[] args) {
Disruptor<FunEvent> disruptor = new Disruptor<FunEvent>(
FunEvent::new,
1024 * 1024,
ThreadPoolUtil.getFactory(),
ProducerType.MULTI,
new YieldingWaitStrategy()
)
disruptor.handleEventsWithWorkerPool(new FunEventHandler(), new FunEventHandler())
disruptor.handleEventsWith(new FunEventHandler())
disruptor.start()
RingBuffer<FunEvent> ringBuffer = disruptor.getRingBuffer();
def funtester = {
fun {
100.times {ringBuffer.publishEvent((event, sequence) -> event.setId(StringUtil.getString(10)));}
}
}
10.times {funtester()}
sleep(5.0)
disruptor.shutdown()
}
![disruptor （史上最全）[通俗易懂]](https://ask.qcloudimg.com/http-save/yehe-8223537/f1ac6c6c2d6495aa87aeaf5f23a6e12d.png)
腾讯云开发者
