Disruptor是LMAX公司开源的一款高性能的多线程通信库。Java的队列在高并发场景下会带来延迟,而LMAX目标是成为世界上最快的交易平台,也就是系统要有非常低的延迟和很好的吞吐量。为了优化Java队列的延迟问题,LMAX研发了Disruptor。
Disruptor不是仅仅为金融领域专用的,在解决并发编程的难题上,它都是可以适用的。
在普通的并发编程中,CPU级别的缓存未命中,还有内核对锁的操作都是有很大的开销,Disruptor则是无锁的。
实战
在Disruptor入门中,我们会考虑用一个简单的例子来理解它,从producer中传递一个Long类型的值,然后在consumer中将数字输出。
引入依赖:
<dependency> <groupId>com.lmax</groupId> <artifactId>disruptor</artifactId> <version>3.2.0</version> </dependency>
首先,定义要携带数据的事件对象。
public class LongEvent { private long value; public void set(long value) { this.value = value; } }
为了Disrutpor可以预先分配这些对象,还需要创建一个事件工厂。
import com.lmax.disruptor.EventFactory; public class LongEventFactory implements EventFactory<LongEvent> { public LongEvent newInstance() { return new LongEvent(); } }
当我们的事件工厂创建完毕以后,我们需要创建consume来处理这些事件,本次我们只需要在控制台中输出事件携带的数据就好。
import com.lmax.disruptor.EventHandler; public class LongEventHandler implements EventHandler<LongEvent> { public void onEvent(LongEvent event, long sequence, boolean endOfBatch) { System.out.println("Event: " + event); } }
发布事件,在Disruptor 3.0之后,更喜欢使用Event Publisher/Event Translator来发布事件。使用这种方式的好处是,translator代码可以独立出来,并且更加容易进行测试,Disruptor也提供了很多内置的(EventTranslator, EventTranslatorOneArg, EventTranslatorTwoArg)等,
publicclassLongEventProducerWithTranslator{ privatefinalRingBuffer<LongEvent>ringBuffer; publicLongEventProducerWithTranslator(RingBuffer<LongEvent>ringBuffer) { this.ringBuffer=ringBuffer; } privatestaticfinalEventTranslatorOneArg<LongEvent, ByteBuffer>TRANSLATOR= newEventTranslatorOneArg<LongEvent, ByteBuffer>() { @Override publicvoidtranslateTo(LongEventevent, longsequence, ByteBufferbb) { event.set(bb.getLong(0)); } }; publicvoidonData(ByteBufferbb) { ringBuffer.publishEvent(TRANSLATOR, bb); } }
除了使用Translator,还有一种更加常见的方法,但是这种方法在多发布者的场景中,可能会出现一些无法预期的问题,因此还是建议使用Translator。
// 不建议使用 @Deprecated public class LongEventProducer { private final RingBuffer<LongEvent> ringBuffer; public LongEventProducer(RingBuffer<LongEvent> ringBuffer) { this.ringBuffer = ringBuffer; } public void onData(ByteBuffer bb) { long sequence = ringBuffer.next(); // Grab the next sequence try { LongEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor // for the sequence event.set(bb.getLong(0)); // Fill with data } finally { ringBuffer.publish(sequence); } } }
将整个过程串联起来。
publicclassLongEventMain{ publicstaticvoidmain(String[] args) throwsInterruptedException{ // 构建Disrutor对象 Disruptor<LongEvent>disruptor=newDisruptor<LongEvent>(newLongEventFactory(), 1024*1024, Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()), ProducerType.SINGLE, newBusySpinWaitStrategy() ); // 设置消费者 disruptor.handleEventsWith(newLongEventHandler()); // 启动Disrutor disruptor.start(); RingBuffer<LongEvent>ringBuffer=disruptor.getRingBuffer(); LongEventProducerWithTranslatorproducer=newLongEventProducerWithTranslator(ringBuffer); ByteBufferbb=ByteBuffer.allocate(8); for(longl=0; true; l++) { bb.putLong(0, l); producer.onData(bb); Thread.sleep(1000); } } }
查看运行结果
用法就是上面那些了,还有一些Disrutor的具体使用,可以结合实际的业务场景。
扩展
在创建Disrutor对象的时候,需要指定RingBuffer的等待策略,默认的测试是BlockingWaitStrategy,阻塞等待。也就是使用锁,然后等待其它的线程进行唤醒。除了阻塞等待之外,还有一些系统提供的策略,当然也可以自定义等待策略
SleepingWaitStrategy 在不需要低延迟的场景下,可以使用它,内部用的是LockSupport.parkNanos。
YieldingWaitStrategy 内部使用的是Threa.yield,允许其他的线程先进行,比较推荐的使用方式
BusySpinWaitStrategy 一般在实际的消费者数量小于核心线程数的时候使用,因为它会不断的循环,占用CPU资源
其它用法
我看到内部的系统再使用Disrutor的时候,在消费事件的时候,将事件类型进行转换,再次异步处理。
publicclassAlarmDisruptorConsumerimplementsWorkHandler<AlarmDisruptorVO>{ @Override publicvoidonEvent(AlarmDisruptorVOevent) throwsException{ if(event!=null) { process(event.getAlarmMsgVO()); process(event.getAlarmTaskVO()); } } publicvoidprocess(AlarmMsgVOmsgVO) { if(msgVO==null) { return; } AlarmMsgHandlerServicealarmMsgHandlerService=getAlarmMsgHandlerService(); // 将消息转换为任务 alarmMsgHandlerService.alarmMsg2Task(msgVO); } publicvoidprocess(AlarmTaskVOmsgVO) { if(msgVO==null) { return; } AlarmMsgHandlerServicealarmMsgHandlerService=getAlarmMsgHandlerService(); // 处理任务 alarmMsgHandlerService.alarmTask2Process( msgVO); } privateAlarmMsgHandlerServicegetAlarmMsgHandlerService(){ return(AlarmMsgHandlerService) SpringApplicationContext.getBean("alarmMsgHandlerService"); } }
最后
关于Disrutor的使用就说到这了,如果JDK内部的并发编程框架不能满足你对性能上的要求,那么这是一个值得尝试的方案。