关键词:
前言
介绍高性能队列Disruptor原理以及使用例子。
Disruptor是什么?
Disruptor是外汇和加密货币交易所运营商 LMAX group 建立高性能的金融交易所的结果。用于解决生产者、消费者及其数据存储的设计问题的高性能队列实现。可以对标JDK中的ArrayBlockingQueue。是目前单机且基于内存存储的最高性能的队列实现。见 与ArrayBlockingQueue性能对比。
Disruptor高性能秘诀
使用CAS代替锁
锁非常昂贵,因为它们在竞争时需要仲裁。这种仲裁是通过到操作系统内核的上下文切换来实现的,该内核将挂起等待锁的线程,直到它被释放。系统提供的原子操作CAS(Compare And Swap/Set)是很好的锁替代方案,Disruptor中同步就是使用的这种。
比如多生产者模式中com.lmax.disruptor.MultiProducerSequencer就是用了Java里sun.misc.Unsafe类基于CAS实现的API。
等待策略com.lmax.disruptor.BlockingWaitStrategy使用了基于CAS实现的ReentrantLock。
独占缓存行
为了提高效率CPU硬件不会以字节或字为单位移动内存,而是以缓存行,通常大小为 32-256 字节的缓存行,最常见的缓存行是 64 字节。这意味着,如果两个变量在同一个缓存行中,并且由不同的线程写入,那么它们会出现与单个变量相同的写入争用问题。为了获得高性能,如果要最小化争用,那么确保独立但同时写入的变量不共享相同的缓存行是很重要的。
比如com.lmax.disruptor.RingBuffer中属性前后都用未赋值的long来独占。com.lmax.disruptor.SingleProducerSequencerPad也有相同处理方式。
环形队列
- 使用有界队列,减少线程争用
队列相比链表在访问速度上占据优势,而有界队列相比可动态扩容的无界队列则避免扩容产生的同步问题效率更高。Disruptor和JDK中的ArrayBlockingQueue一样使用有界队列。队列长度要设为2的n次幂,有利于二进制计算。
- 使用环形数组,避免生产和消费速度差异导致队列头和尾争用
Disruptor在逻辑上将数组的的头尾看成是相连的,即一个环形数组(RingBuffer)。
- Sequence
生产和消费都需要维护自增序列值(Sequence),从0开始。
生产方只维护一个代表生产的最后一个元素的序号。代表生产的最后一个元素的序号。每次向Disruptor发布一个元素都调用Sequenced.next()来获取下个位置的写入权。
在单生产者模式(SINGLE)由于不存在并发写入,则不需要解决同步问题。在多生产者模式(MULTI)就需要借助JDK中基于CAS(Compare And Swap/Set)实现的API来保证线程安全。
多个消费者各自维护自己的消费序列值(Sequence)保存数组中。
而环形通过与运算(sequence & indexMask)实现的,indexMask就是环形队列的长度-1。以环形队列长度8为例,第9个元素Sequence为8,8 & 7 = 0,刚好又回到了数组第1个位置。
见com.lmax.disruptor.RingBuffer.elementAt(long sequence)
预分配内存
环形队列存放的是Event对象,而且是在Disruptor创建的时候调用EventFactory创建并一次将队列填满。Event保存生产者生产的数据,消费也是通过Event获取,后续生产则只需要替换掉Event中的属性值。这种方式避免了重复创建对象,降低JVM的GC产频率。
见com.lmax.disruptor.RingBuffer.fill(EventFactory
消费者8种等待策略
当消费速度大于生产速度情况下,消费者执行的等待策略。
策略类名 | 描述 |
---|---|
BlockingWaitStrategy(常用) | 使用ReentrantLock,失败则进入等待队列等待唤醒重试。当吞吐量和低延迟不如CPU资源重要时使用。 |
YieldingWaitStrategy(常用) | 尝试100次,全失败后调用Thread.yield()让出CPU。该策略将使用100%的CPU,如果其他线程请求CPU资源,这种策略更容易让出CPU资源。 |
SleepingWaitStrategy(常用) | 尝试200次 。前100次直接重试,后100次每次失败后调用Thread.yield()让出CPU,全失败线程睡眠(默认100纳秒 )。 |
BusySpinWaitStrategy | 线程一直自旋等待,比较耗CPU。最好是将线程绑定到特定的CPU核心上使用。 |
LiteBlockingWaitStrategy | 与BlockingWaitStrategy类似,区别在增加了原子变量signalNeeded,如果两个线程同时分别访问waitFor()和signalAllWhenBlocking(),可以减少ReentrantLock加锁次数。 |
LiteTimeoutBlockingWaitStrategy | 与LiteBlockingWaitStrategy类似,区别在于设置了阻塞时间,超过时间后抛异常。 |
TimeoutBlockingWaitStrategy | 与BlockingWaitStrategy类似,区别在于设置了阻塞时间,超过时间后抛异常。 |
PhasedBackoffWaitStrategy | 根据时间参数和传入的等待策略来决定使用哪种等待策略。当吞吐量和低延迟不如CPU资源重要时,可以使用此策略。 |
消费者序列
所有消费者的消费序列(Sequence)都放在一个数组中,见com.lmax.disruptor.AbstractSequencer,通过SEQUENCE_UPDATER来更新对应的序列值。
调用更新的地方在com.lmax.disruptor.RingBuffer.addGatingSequences(Sequence... gatingSequences)。
消费太慢队列满了怎么办?
生产者线程被阻塞。生产者调用Sequenced.next()争夺写入权的时候需要判断最小的消费序列值进行比较。如果写入的位置还未消费则会进入循环不断获取最小消费序列值进行比较。
见包com.lmax.disruptor下SingleProducerSequencer或MultiProducerSequencer中next(int n)方法。
Disruptor开发步骤
- 创建Event、EventFactory、EventHandler和ExceptionHandler类
Event是环形队列(RingBuffer)中的元素,是生产者数据的载体;EventFactory是定义Event创建方式的工厂类;EventHandler则是Event的处理器,定义如何消费Event中的数据。
另外有必要定义一个消费异常处理器ExceptionHandler,它是和EventHandler绑定的。当EventHandler.onEvent()执行抛出异常时会执行对应的异常回调方法。
- 实例化Disruptor
创建Disruptor需要指定5个参数eventFactory、ringBufferSize、threadFactory、producerType、waitStrategy。
EventFactory是上面定义的Event工厂类;
ringBufferSize是环形队列的长度,这个值要是2的N次方;
threadFactory是定义消费者线程创建方式的工厂类;
producerType是指明生产者是一个(SINGLE)还是多个(MULTI)。默认是MULTI,会使用CAS(Compare And Swap/Set)保证线程安全。如果指定为SINGLE,则不使用没必要的CAS,使单线程处理更高效。
waitStrategy指明消费者等待生产时的策略。
- 设置消费者
指明EventHandler并绑定ExceptionHandler。指定多个EventHandler时,会为每个EventHandler分配一个线程,一个Event会被多个并行EventHandler处理。
也可以指明多个WorkHandler,每个WorkHandler分配一个线程并行消费队列中的Event,一个Event只会被一个WorkHandler处理。
- 创建/实例化EventTranslator
EventTranslator定义生产者数据转换为Event的方式,不同数量参数有不同的接口用来实现。
- 最后用Disruptor.publishEvent() 来发布元素指明EventTranslator和参数
例子程序
- 先引入Maven依赖
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.4.4</version>
</dependency>
- Event
/**
* 事件
*
* @param <T>发布的数据类型
*/
public class MyEvent<T>
private T data;
public T getData()
return data;
public MyEvent<T> setData(T data)
this.data = data;
return this;
- EventFactory
import com.lmax.disruptor.EventFactory;
/**
* 创建事件的工厂
*
* @param <T>发布的数据类型
*/
public class MyEventFactory<T> implements EventFactory<MyEvent<T>>
@Override
public MyEvent<T> newInstance()
return new MyEvent<>();
- EventHandler
import com.lmax.disruptor.EventHandler;
/**
* 事件消费方法
*
* @param <T>发布的数据类型
*/
public class MyEventHandler<T> implements EventHandler<MyEvent<T>>
@Override
public void onEvent(MyEvent<T> tMyEvent, long l, boolean b) throws Exception
System.out.println(Thread.currentThread().getName() + "MyEventHandler消费:" + tMyEvent.getData());
- ExceptionHandler
import com.lmax.disruptor.ExceptionHandler;
/**
* 消费者异常处理器
*
* @param <T>发布的数据类型
*/
public class MyExceptionHandler<T> implements ExceptionHandler<MyEvent<T>>
@Override
public void handleEventException(Throwable ex, long sequence, MyEvent<T> event)
System.out.println("handleEventException");
@Override
public void handleOnStartException(Throwable ex)
System.out.println("handleOnStartException");
@Override
public void handleOnShutdownException(Throwable ex)
System.out.println("handleOnShutdownException");
单消费者
import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
/**
* 单消费者
*/
public class SingleConsumerSample
public static void main(String[] args)
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建事件(Event)对象的工厂
MyEventFactory<String> eventFactory = new MyEventFactory<>();
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(eventFactory, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 指定一个处理器
MyEventHandler<String> eventHandler = new MyEventHandler<>();
disruptor.handleEventsWith(eventHandler);
// 处理器异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.setDefaultExceptionHandler(exceptionHandler);
disruptor.start();
// 通过事件转换器(EventTranslator)来指明如何将发布的数据转换到事件对象(Event)中
// 这里是一个参数的转换器,另外还有两个(EventTranslatorTwoArg)、三个(EventTranslatorThreeArg)
// 和多个(EventTranslatorVararg)参数的转换器可以使用,参数类型可以不一样
EventTranslatorOneArg<MyEvent<String>, String> eventTranslatorOneArg =
new EventTranslatorOneArg<MyEvent<String>, String>()
@Override
public void translateTo(MyEvent<String> event, long sequence, String arg0)
event.setData(arg0);
;
// 发布
for (int i = 0; i < 10; i++)
disruptor.publishEvent(eventTranslatorOneArg, "One arg " + i);
disruptor.shutdown();
单消费者Lambda写法
这种只是迎合Java8 Lambda语法特性,代码更简洁。
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.stream.Collectors;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
public class LambdaSample
public static void main(String[] args)
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(MyEvent::new, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 指定一个处理器
EventHandler<MyEvent<String>> eventHandler = (event, sequence, endOfBatch) ->
System.out.println(Thread.currentThread().getName() + "MyEventHandler消费:" + event.getData());
disruptor.handleEventsWith(eventHandler);
// 处理器异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.setDefaultExceptionHandler(exceptionHandler);
disruptor.start();
// 通过事件转换器(EventTranslator)来指明如何将发布的数据转换到事件对象(Event)中
// 一个参数的转换器
disruptor.publishEvent((event, sequence, param) -> event.setData(param), "One arg ");
// 两个参数的转换器
disruptor.publishEvent((event, sequence, pA, pB) -> event.setData(pA + pB), "Two arg ", 1);
// 三个参数的转换器
disruptor.publishEvent((event, sequence, pA, pB, pC) -> event.setData(pA + pB + pC)
, "Three arg ", 1, false);
// 多个参数的转换器
disruptor.getRingBuffer().publishEvent((event, sequence, params) ->
List<String> paramList = Arrays.stream(params).map(Object::toString).collect(Collectors.toList());
event.setData("Var arg " + String.join(",", paramList));
, "param1", "param2", "param3");
disruptor.shutdown();
多消费者重复消费元素
关键只在于指定多个EventHandler,并且EventHandler还可以分别绑定不同的ExceptionHandler。
每个EventHandler分配一个线程,一个Event会被每个EventHandler处理,适合两个不同的业务都需要处理同一个元素的情况,类似广播模式。
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
/**
* 一个元素多个消费者重复消费
*/
public class RepetitionConsumerSample
public static void main(String[] args)
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建事件(Event)对象的工厂
MyEventFactory<String> eventFactory = new MyEventFactory<>();
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(eventFactory, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 这里指定了2个消费者,那就会产生2个消费线程,一个事件会被消费2次
EventHandler<MyEvent<String>> eventHandler = (event, sequence, endOfBatch) ->
System.out.println(Thread.currentThread().getName() + "MyEventHandler消费:" + event.getData());
EventHandler<MyEvent<String>> eventHandler2 = (event, sequence, endOfBatch) ->
System.out.println(Thread.currentThread().getName() + "MyEventHandler——2消费:" + event.getData());
disruptor.handleEventsWith(eventHandler, eventHandler2);
// 分别指定异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.handleExceptionsFor(eventHandler).with(exceptionHandler);
disruptor.handleExceptionsFor(eventHandler2).with(exceptionHandler);
disruptor.start();
for (int i = 0; i < 10; i++)
disruptor.publishEvent((event, sequence, param) -> event.setData(param), "One arg " + i);
disruptor.shutdown();
多消费者
关键只在于定义WorkHandler,然后实例化多个来消费。
每个WorkHandler分配一个线程,一个元素只会被一个WorkHandler处理。
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.SleepingWaitStrategy;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import static com.lmax.disruptor.dsl.ProducerType.SINGLE;
public class MultiConsumerSample
public static void main(String[] args)
// 环形数组长度,必须是2的n次幂
int ringBufferSize = 1024;
// 创建事件(Event)对象的工厂
MyEventFactory<String> eventFactory = new MyEventFactory<>();
// 创建消费者线程工厂
ThreadFactory threadFactory = Executors.defaultThreadFactory();
// 等待策略
WaitStrategy waitStrategy = new SleepingWaitStrategy();
Disruptor<MyEvent<String>> disruptor =
new Disruptor<>(eventFactory, ringBufferSize, threadFactory, SINGLE, waitStrategy);
// 处理器异常处理器
ExceptionHandler<MyEvent<String>> exceptionHandler = new MyExceptionHandler<>();
disruptor.setDefaultExceptionHandler(exceptionHandler);
// 设置2个消费者,2个线程,一个Event只被一个消费者消费
WorkHandler<MyEvent<String>> workHandler = tMyEvent ->
System.out.println(Thread.currentThread().getName() + "WorkHandler消费:" + tMyEvent.getData());
disruptor.handleEventsWithWorkerPool(workHandler, workHandler2);
disruptor.start();
for (int i = 0; i < 10; i++)
disruptor.publishEvent((event, sequence, param) -> event.setData(param), "One arg " + i);
disruptor.shutdown();
参考链接
GitHub Disruptor Getting Started
disruptor使用(代码片段)
Disruptor作者,介绍Disruptor能每秒处理600万订单。这是一个可怕的数字。disruptor之所以那么快,是因为内部采用环形队列和无锁设计。使用cas来进行并发控制。通过获取可用下标来对事件发布和消费下标通过cas控制(Atomic)disruptor组... 查看详情
高性能并发队列disruptor(代码片段)
Disruptor基本概念Disruptor应用定义事件定义事件工厂定义事件处理实现定义事件处理的线程池指定等待策略启动Disruptor发布事件关闭DisruptorDisruptor原理核心概念RingBufferSequenceDisruptorSequencerSequenceBarrierWaitStrategyEventEventProcessorEventHandle... 查看详情
disruptor源码分析
本文将介绍Disruptor的工作机制,并分析Disruptor的主要源码基于的版本是3.3.7(发布于2017.09.28)水平有限,如有谬误请留言指正 0. 什么是Disruptor?Disruptor是一个开源的并发框架,提供了类似于Java中有界队列的功能,主要用... 查看详情
java无锁队列disruptor,内存队列的生产解决方案(代码片段)
背景Disruptor是英国外汇交易所LMAX开源的用于生产交易中的内存队列。为了实现高性能交易撮合队列时,现在普遍的交易撮合引擎都采用了内存队列的方式,这种方式减少了持久化过程中带来的磁盘IO延迟,可以提交整... 查看详情
springbootdisruptor高性能内存消息队列(代码片段)
一、背景工作中遇到项目使用Disruptor做消息队列,对你没看错,不是Kafka,也不是rabbitmq;Disruptor有个最大的优点就是快,还有一点它是开源的哦,下面做个简单的记录。二、Disruptor介绍Disruptor是英国外汇交易公司LMAX开发的一个高性能队... 查看详情
如何优雅地使用disruptor
参考技术A从功能上来看,Disruptor是实现了“队列”的功能,而且是一个有界队列。那么它的应用场景自然就是“生产者-消费者”模型的应用场合了。先从了解Disruptor的核心概念开始,来了解它是如何运作的。下面介绍的概念模... 查看详情
disruptor使用简介(代码片段)
【开发总结】Disruptor使用简介在极客时间看到王宝令老师关于Disruptor的一篇文章,觉得很有意思。看完之后又在网上找到一些其他关于Disruptor的资料看了一下。现在写篇文章总结一下。使用Disruptor百度翻译是干扰者,分裂器的意... 查看详情
高性能队列disruptor使用入门,原理和代码实现(代码片段)
网上有不少关于Disruptor的文章,这里聊聊自己的看法。我总结Disruptor高性能的实现在于如下几点:缓存行填充对齐无锁CAS操作环形缓冲区异步并发消费缓存行失效、伪共享一般我们知道在我们程序运行的时候,内存访... 查看详情
高性能并发队列disruptor(代码片段)
Disruptor基本概念Disruptor应用定义事件定义事件工厂定义事件处理实现定义事件处理的线程池指定等待策略启动Disruptor发布事件关闭DisruptorDisruptor原理核心概念RingBufferSequenceDisruptorSequencerSequenceBarrierWaitStrategyEventEventProcessorEventHandle... 查看详情
springbootdisruptor高性能内存消息队列(代码片段)
一、背景工作中遇到项目使用Disruptor做消息队列,对你没看错,不是Kafka,也不是rabbitmq;Disruptor有个最大的优点就是快,还有一点它是开源的哦,下面做个简单的记录。二、Disruptor介绍Disruptor是英国外汇交易公司LMAX开发的一个高性能队... 查看详情
高性能队列——disruptor(代码片段)
文章目录声明:高性能队列——Disruptor背景Java内置队列ArrayBlockingQueue的问题加锁伪共享Disruptor的设计方案一个生产者多个生产者读数据写数据总结性能等待策略生产者的等待策略消费者的等待策略Log4j2应用场景性能差异参考... 查看详情
高性能队列disruptor为什么这么快?(代码片段)
背景Disruptor是LMAX开发的一个高性能队列,研发的初衷是解决内存队列的延迟问题(在性能测试中发现竟然与I/O操作处于同样的数量级)。基于Disruptor开发的系统单线程能支撑每秒600万订单,2010年在QCon演讲后,获得了业界关注。... 查看详情
disruptor以及@contended注解(代码片段)
Disruptor感想很早之前阅读过Disruptor的使用,这里有篇美团团队的文章很详细的介绍了这种队列的相关原理以及为什么这么设计和这么设计为什么处理速度很快,https://tech.meituan.com/disruptor.html,不想再“拿来主义”... 查看详情
深入理解disruptor(代码片段)
Disruptor通过缓存行填充,利用CPU高速缓存,只是Disruptor“快”的一个因素,快的另一因素是“无锁”,尽可能发挥CPU本身的高速处理性能。1缓慢的锁Disruptor作为一个高性能的生产者-消费者队列系统,核心就是... 查看详情
disruptor——一种可替代有界队列完成并发线程间数据交换的高性能解决方案
本文翻译自LMAX关于Disruptor的论文,同时加上一些自己的理解和标注。Disruptor是一个高效的线程间交换数据的基础组件,它使用栅栏(barrier)+序号(Sequencing)机制协调生产者与消费者,从而避免使用锁和CAS,同时还组... 查看详情
disruptor(代码片段)
转载于:https://www.cnblogs.com/crazymakercircle/p/13909235.html1disruptor是什么?Disruptor是英国外汇交易公司LMAX开发的一个高性能队列,研发的初衷是解决内存队列的延迟问题(在性能测试中发现竟然与I/O操作处于同样的数量级)。基于Disru... 查看详情
springbootdisruptor构建高性能内存队列(代码片段)
...比较低,尽量使用无锁设计。接下来就我们来认识下Disruptor。Disruptor简单使用github地址:https://githu 查看详情
springboot+disruptor=王炸!!
...众号,Java干货及时送达01、背景工作中遇到项目使用Disruptor做消息队列,对你没看错,不是Kafka,也不是rabbitmq;Disruptor有个最大的优点就是快,还有一点它是开源的哦,下面做个简单的记录.02、Disruptor介绍Disruptor是英国外汇交易公司... 查看详情