Disruptor是什么?
Disruptor是一个高性能的异步处理框架,或者可以认为是线程间通信的高效低延时的内存消息组件,它最大特点是高性能,其LMAX架构可以获得每秒6百万订单,用1微秒的延迟获得吞吐量为100K+。
它是如何实现高性能的呢?它由于JDK内置的队列有什么区别呢?JDK内置内存队列?
我们知道,Java内置了几种内存消息队列,如下所示:
队列 | 加锁方式 | 是否有界 | 数据结构 |
---|---|---|---|
ArrayBlockingQueue | 加锁 | 有界 | ArrayList |
LinkedBlockingQueue | 加锁 | 无界 | LinkedList |
ConcurrentLinkedQueue | CAS | 无界 | LinkedList |
LinkedTransferQueue | CAS | 无界 | LinkedList |
我们知道CAS算法比通过加锁实现同步性能高很多,而上表可以看出基于CAS实现的队列都是无界的,而有界队列是通过同步实现的。在系统稳定性要求比较高的场景下,为了防止生产者速度过快,如果采用无界队列会最终导致内存溢出,只能选择有界队列。而有界队列只有ArrayBlockingQueue
,该队列是通过加锁实现的,在请求锁和释放锁时对性能开销很大,这时候基于有界队列的高性能的Disruptor就应运而生。
Disruptor如何实现高性能?
Disruptor实现高性能主要体现了去掉了锁,采用CAS算法,同时内部通过环形队列实现有界队列。
- 环形数据结构 为了避免垃圾回收,采用数组而非链表。同时,数组对处理器的缓存机制更加友好。
- 元素位置定位 数组长度2^n,通过位运算,加快定位的速度。下标采取递增的形式。不用担心index溢出的问题。index是long类型,即使100万QPS的处理速度,也需要30万年才能用完。
- 无锁设计 每个生产者或者消费者线程,会先申请可以操作的元素在数组中的位置,申请到之后,直接在该位置写入或者读取数据。整个过程通过原子变量CAS,保证操作的线程安全。
Disruptor可以用来做什么?
当前业界开源组件使用Disruptor的包括Log4j2、Apache Storm等,它可以用来作为高性能的有界内存队列,基于生产者消费者模式,实现一个/多个生产者对应多个消费者。它也可以认为是观察者模式的一种实现,或者发布订阅模式。
同时,Disruptor还允许开发者使用多线程技术去创建基于任务的工作流。Disruptor能用来并行创建任务,同时保证多个处理过程的有序性,并且它是没有锁的。
为什么要使用Disruptor?
使用Disruptor,主要用于对性能要求高、延迟低的场景,它通过“榨干”机器的性能来换取处理的高性能。如果你的项目有对性能要求高,对延迟要求低的需求,并且需要一个无锁的有界队列,来实现生产者/消费者模式,那么Disruptor是你的不二选择。
怎么用Disruptor?
要学会基于Disruptor进行编程,我们先了解下大概流程示意图,其中绿色部分是表示我们需要编写和实现的类。
import com.lmax.disruptor.BlockingWaitStrategy;import com.lmax.disruptor.EventFactory;import com.lmax.disruptor.EventHandler;import com.lmax.disruptor.EventTranslatorOneArg;import com.lmax.disruptor.ExceptionHandler;import com.lmax.disruptor.RingBuffer;import com.lmax.disruptor.dsl.Disruptor;import com.lmax.disruptor.dsl.ProducerType;import java.util.concurrent.ThreadFactory;/** * @author: chuanyi.huang **/public class DisruptorTest { /* * 消息事件类 */ public static class MessageEvent { /** * 原始消息 */ private String message; public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } } /* * 消息事件工厂类 */ public static class MessageEventFactory implements EventFactory{ public MessageEvent newInstance() { return new MessageEvent(); } } /* * 消息转换类, 负责将消息转换成事件 */ public static class MessageEventTranslator implements EventTranslatorOneArg { public void translateTo(MessageEvent messageEvent, long l, String s) { messageEvent.setMessage(s); } } /* * 消费者线程工厂类 */ public static class MessageThreadFactory implements ThreadFactory { public Thread newThread(Runnable r) { return new Thread(r, "Simple Discruptor Test Thread"); } } /* * 消息事件处理类 */ public static class MessageEventHandler implements EventHandler { public void onEvent(MessageEvent messageEvent, long l, boolean b) throws Exception { System.out.println(messageEvent.getMessage()); } } /* * 异常处理类 */ public static class MessageExceptionHandler implements ExceptionHandler { public void handleEventException(Throwable throwable, long l, MessageEvent messageEvent) { throwable.printStackTrace(); } public void handleOnStartException(Throwable throwable) { throwable.printStackTrace(); } public void handleOnShutdownException(Throwable throwable) { throwable.printStackTrace(); } } /* * 消息生产者类 */ public static class MessageEventProducer { private RingBuffer ringBuffer; public MessageEventProducer(RingBuffer ringBuffer) { this.ringBuffer = ringBuffer; } /** * 将接收到的消息输出到ringBuffer */ public void onData(String message) { EventTranslatorOneArg translator = new MessageEventTranslator(); ringBuffer.publishEvent(translator, message); } } public static void main(String[] args) { String message = "Hello Disruptor"; int ringBufferSize = 1024; Disruptor disruptor = new Disruptor (new MessageEventFactory(), ringBufferSize, new MessageThreadFactory(), ProducerType.SINGLE, new BlockingWaitStrategy()); disruptor.handleEventsWith(new MessageEventHandler()); disruptor.setDefaultExceptionHandler(new MessageExceptionHandler()); RingBuffer start = disruptor.start(); MessageEventProducer messageEventProducer = new MessageEventProducer(start); messageEventProducer.onData(message); }}