并发框架Disruptor几个Demo - ING - ITeye技术网站
博客分类：
经过2个月的疯狂加班后终于迎来了五一节前的几天清闲日子，在这闲得蛋疼的日子里，本屌丝无意中从ifeve上看到“Disruptor”并发框架，它号称"能够在一个线程里每秒处理6百万订单" 当时就被它吸引住了，不管是不是在吹流逼吧，先研究研究在说....
要想了解Disruptor框架必需多花点时间研究下它的工作原理，不然代码是没法撸的！！！
关于Disruptor的详细资料及原理请细看!!!
/disruptor/
Disruptor版本： 3.2.1
消费者==事件处理器
一、现在你必须明白以下问题：
1、你必须明白Ringbuffer是什么，它的数据结构是怎么样的，有什么约定，为什么高效，它的职责是什么。
2、ConsumerBarrier （ifeve网上的译文版本比较早，这个类在2.0.0之后就一直被改名，3.2.1的版本中它是SequenceBarrier）它的职责是什么。
Disruptor框架在2.0版本之后不再采用生产者、消费者模型来编写API，而是使用事件模型，其实只是接口设计、类名和概念上的变化，内部原理其实还是一样的。
DEMO 一、使用原生API创建一个简单的生产者和消费者
//DEMO中使用的 消息全假定是一条交易
public class TradeTransaction {
private S//交易ID
//交易金额
public TradeTransaction() {
public TradeTransaction(String id, double price) {
this.price =
public String getId() {
public void setId(String id) {
public double getPrice() {
public void setPrice(double price) {
this.price =
public class TradeTransactionInDBHandler implements EventHandler&TradeTransaction&,WorkHandler&TradeTransaction& {
public void onEvent(TradeTransaction event, long sequence,
boolean endOfBatch) throws Exception {
this.onEvent(event);
public void onEvent(TradeTransaction event) throws Exception {
//这里做具体的消费逻辑
event.setId(UUID.randomUUID().toString());//简单生成下ID
System.out.println(event.getId());
public class Demo1 {
public static void main(String[] args) throws InterruptedException, ExecutionException {
int BUFFER_SIZE=1024;
int THREAD_NUMBERS=4;
* createSingleProducer创建一个单生产者的RingBuffer，
* 第一个参数叫EventFactory，从名字上理解就是“事件工厂”，其实它的职责就是产生数据填充RingBuffer的区块。
* 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
* 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
final RingBuffer&TradeTransaction& ringBuffer = RingBuffer.createSingleProducer(new EventFactory&TradeTransaction&() {
public TradeTransaction newInstance() {
return new TradeTransaction();
}, BUFFER_SIZE,new YieldingWaitStrategy());
//创建线程池
ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);
//创建SequenceBarrier
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
//创建消息处理器
BatchEventProcessor&TradeTransaction& transProcessor = new BatchEventProcessor&TradeTransaction&(
ringBuffer, sequenceBarrier, new TradeTransactionInDBHandler());
//这一部的目的是让RingBuffer根据消费者的状态 如果只有一个消费者的情况可以省略
ringBuffer.addGatingSequences(transProcessor.getSequence());
//把消息处理器提交到线程池
executors.submit(transProcessor);
//如果存大多个消费者 那重复执行上面3行代码 把TradeTransactionInDBHandler换成其它消费者类
Future&?& future=executors.submit(new Callable&Void&() {
public Void call() throws Exception {
for(int i=0;i&1000;i++){
seq=ringBuffer.next();//占个坑 --ringBuffer一个可用区块
ringBuffer.get(seq).setPrice(Math.random()*9999);//给这个区块放入 数据
如果此处不理解，想想RingBuffer的结构图
ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见
future.get();//等待生产者结束
Thread.sleep(1000);//等上1秒，等消费都处理完成
transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）
executors.shutdown();//终止线程
DEMO二、使用WorkerPool辅助创建消费者
public class Demo2 {
public static void main(String[] args) throws InterruptedException {
int BUFFER_SIZE=1024;
int THREAD_NUMBERS=4;
EventFactory&TradeTransaction& eventFactory=new EventFactory&TradeTransaction&() {
public TradeTransaction newInstance() {
return new TradeTransaction();
RingBuffer&TradeTransaction& ringBuffer=RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);
WorkHandler&TradeTransaction& workHandlers=new TradeTransactionInDBHandler();
* 这个类代码很简单的，亲自己看哈！~
WorkerPool&TradeTransaction& workerPool=new WorkerPool&TradeTransaction&(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), workHandlers);
workerPool.start(executor);
//下面这个生产8个数据，图简单就写到主线程算了
for(int i=0;i&8;i++){
long seq=ringBuffer.next();
ringBuffer.get(seq).setPrice(Math.random()*9999);
ringBuffer.publish(seq);
Thread.sleep(1000);
workerPool.halt();
executor.shutdown();
DEMO三、demo3写个流弊点的像下图这样。这次用Disruptor来完成整个构建工作.
从中图可以看出需求是介样子的：生产者生产数据经过C1,C2处理完成后再到C3。
假设如下场景：
1、交易网关收到交易(P1)把交易数据发到RingBuffer中，
2、负责处理增值业务的消费者C1和负责数据存储的消费者C2负责处理交易
3、负责发送JMS消息的消费者C3在C1和C2处理完成后再进行处理。
让代码说话：
public class TradeTransactionJMSNotifyHandler implements EventHandler&TradeTransaction& {
public void onEvent(TradeTransaction event, long sequence,
boolean endOfBatch) throws Exception {
//do send jms message
public class TradeTransactionPublisher implements Runnable{
Disruptor&TradeTransaction&
private CountDownL
private static int LOOP=;//模拟一千万次交易的发生
public TradeTransactionPublisher(CountDownLatch latch,Disruptor&TradeTransaction& disruptor) {
this.disruptor=
this.latch=
public void run() {
TradeTransactionEventTranslator tradeTransloator=new TradeTransactionEventTranslator();
for(int i=0;i&LOOP;i++){
disruptor.publishEvent(tradeTransloator);
latch.countDown();
class TradeTransactionEventTranslator implements EventTranslator&TradeTransaction&{
private Random random=new Random();
public void translateTo(TradeTransaction event, long sequence) {
this.generateTradeTransaction(event);
private TradeTransaction generateTradeTransaction(TradeTransaction trade){
trade.setPrice(random.nextDouble()*9999);
public class TradeTransactionVasConsumer implements EventHandler&TradeTransaction& {
public void onEvent(TradeTransaction event, long sequence,
boolean endOfBatch) throws Exception {
//do something....
public class Demo3 {
public static void main(String[] args) throws InterruptedException {
long beginTime=System.currentTimeMillis();
int bufferSize=1024;
ExecutorService executor=Executors.newFixedThreadPool(4);
//这个构造函数参数，相信你在了解上面2个demo之后就看下就明白了，不解释了~
Disruptor&TradeTransaction& disruptor=new Disruptor&TradeTransaction&(new EventFactory&TradeTransaction&() {
public TradeTransaction newInstance() {
return new TradeTransaction();
}, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
//使用disruptor创建消费者组C1,C2
EventHandlerGroup&TradeTransaction& handlerGroup=disruptor.handleEventsWith(new TradeTransactionVasConsumer(),new TradeTransactionInDBHandler());
TradeTransactionJMSNotifyHandler jmsConsumer=new TradeTransactionJMSNotifyHandler();
//声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3
handlerGroup.then(jmsConsumer);
disruptor.start();//启动
CountDownLatch latch=new CountDownLatch(1);
//生产者准备
executor.submit(new TradeTransactionPublisher(latch, disruptor));
latch.await();//等待生产者完事.
disruptor.shutdown();
executor.shutdown();
System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
浏览 11169
浏览: 25026 次
来自: 北京
讲解的非常详细，受益匪浅
还有一个问题，如果出现交叉了怎么解决？
楼主看到留言速回，谢谢，
如果要实现这个图片的流程，应该咋写代码呢，求楼主赐教，刚接触D ...有关于Disruptor框架（二）RingBuffer
好久没来总结disruptor内部结构了，借此来复习下disruptor框架中几个重要角色。之前写过篇博文介绍过，正是这些特性使得disruptor在执行并发的时候效率相当的高。
& &在之前一篇博文里面提到过，disruptor模式是
生产者-消费者模式，而介于两者时间的是RingBuffer环，生产者（例如请求事件）不断向RingBuffer中放入事件object，而消费者不断读取RingBuffer。
&首先来解决什么是RingBuffer这个问题。（声明：所有图片均转自）
上面这张图片就是一个RingBuffer，首先它是一个环，其次它是一个数组。我想你大概清楚了为什么要采用数组，不清楚的话可以参考我前面一片博文，我可以再稍微简单说下，数组的好处在于数组元素内存是连续的，CPU在访问数组其中某个元素时候会预加载其余另外几个元素，这样的效率要高于链表结构。
&4代表RingBuffer中其中某个可用元素（消费者会通过SequenceBarrier和RingBuffer产生对话）。大家会有个疑问，如果是环，那么应该有首尾两个指针，否则可能会产生环结构所带来的问题。
RingBuffer比较特殊，它只维护“指向后一个元素的指针”，所以它没有尾指针，生产者会源源不断向RingBuffer中添加数据，甚至覆盖之前的数据，这也是允许的！
这就是我刚才所说的 RingBuffer允许生产者覆盖之前的数据。那它是如何覆盖的呢？
long number = ringBuffer.next();//ringbuffer
获取指向下一个元素的号码
ringBuffer.get(number).set(XXXX);//ringbuffer 进行数据覆盖
ringBuffer.publish(number);//ringbuffer
通知消费者&
注：我在上一篇博文当中没和大家说明另外一个提升效率的地方---就是通过位运算向RingBuffer数组中添加数据。其实这个原理和向HashMap添加数据一模一样，RingBuffer的容量大多数情况下是
2的指数级别，那么向RingBuffer添加数据 只需要对 2^n-1 进行位运算就能得出放置的位置。
以上是对RingBuffer一个大概的了解，至于代码中RingBuffer怎么样，稍微给大家分析，因为具体情况还是得根据大家业务需求
------------------------------------分割线----------Disruptor模型---------------------------------
在应用分析前给大家看下整体的disruptor运行图（当然不可能在这么一篇博文里吧所有disruptor里的成员包括结构
类之间关系阐述的十分具体）
单个生产者vs多个消费者，这样的好处是，单个生产者不会产生资源竞争（“写”），而读对于RingBuffer来说根本不会产生资源竞争。&
&图中P1是生产者，通过ProduceBarrier想RingBuffer添加数据，一旦数据添加完成，更新RingBuffer中的数字，并且通知消费者（X号已经提交完成，各位消费者可以拿去消费了），C1、C2、C3是消费者，其中C1、C2直接从RingBuffer中通过ConsumerBarrier1（现在已经改成SequenceBarrier好像这样子更加符合现实中的样子），而C3则是通过ConsumerBarrier2从C1、C2获取数据，同时track
生产者的位置。不让消费者直接读取RingBuffer的好处在于松耦合，消费者只需要关注自身消费的内容事件，而至于监听生产者的通知，以及从RingBuffer中拿取数据则都交给ConsumerBarrier就行
---------------------------------------线程池
简介--------------------------------------------
&要理清disruptor各个概念，我想最基本的要求是必须懂线程池的概念。我认为disruptor的出现正是基于线程池的短处进行了重构，如果清楚理解线程池，那么对于disruptor框架的理解就会比较明了
& & 一个线程的创建，运行到结束 分别耗时为t1 t2
t3，而现实中多个短小的任务t2运行时间的综合都不及t1 t2频繁的开销，所以为了解决这一状况出现了线程池。
& &线程池中几个重要的角色：
& &线程数组 &
------------ &对应disruptor中的消费者
& &任务链表 &
------------ &对应disruptor中的RingBuffer
& &任务（Runnable） ------
&对应disruptor中消费者中的事件，即Event
& &向线程池提交任务 ------
&对应disruptor中的生产者
可以说Disruptor是线程池的升级版，对线程池的性能提升是较大的。就好比ConcurrentHashMap对HashTable性能提升也是非常大的。
已投稿到：
以上网友发言只代表其个人观点，不代表新浪网的观点或立场。LMAX在线交易出品的一个高效的无锁并发框架它高效核心在于其无锁队列RingBuffer的独特设计。它可以用来进行线程之间的数据交互。
老生重谈：锁
并发编程中，为了保证准确性，引入了锁的机制，包括乐观锁，悲观锁等。有锁就涉及到资源的竞争，竞争就可能出现死锁，这样的情况下，你只能重启你的机器了。
考虑一个简单自增的问题：
从1加到10亿，（测试机器 Mac Air）
单线程简单自增，耗时5S左右
单线程加锁自增，仅仅简单加锁，没有竞争，耗时40S左右，慢一个数量级
两个线程简单自增，耗时减少，但是结果无法保证准确性，比如存在脏读等问题。
两个线程加锁自增，按照理解，时间应该减半，但是因为引入了锁机制，导致竞争，实际时间更加长。
disruptor怎么做
disruptor在需要保证线程安全的地方，用到了CAS操作，这是一个CPU级别的指令，类似于乐观锁，即Campare and Set/Swap. JAVA 从1.5版本新引入了AtomicLong等支持CAS指令的数据结构。测试代码可以看出来，在引入了AtomicLong的情况下，单线程，耗时为18S左右，多线程，耗时26S左右。
说说数据结构：链表 or 数组
既然是数据交换，那就存在一个产生数据（producer），一个消费数据(consumer)，和数据存储(RingBuffer)，数据存储，我们可以理解，应该是一个队列，数据先到先处理，在java类库中，提供了例如LinkedBlockingQueue、ArrayBlockingQueue等，而disruptor之所以高效，是因为它没有直接使用java类库中提供的队列，而是自己写的RingBuffer。
它有什么特点呢？
顾名思义，它是一个环，准确说，是一个用数组实现的环形队列。
不像传统队列，维护对头，队尾，它只有Sequencer,指向下一个可用的数据缓存区
新产生的数据对原来数据进行写覆盖，不进行remove操作。
队列大小一定是2的N次方
它有什么好处呢？
相比于链表，它寻址更快，时间复杂度控制在O（n）
在初始化时，就已经分配好了内存，而且新产生的只覆盖，所以更少的GC
sequece 一直自增，进行位操作可以快速定位到实际slot , sequece & （array length－1） = array index，比如一共有8槽，9 &（8－1）= 1
对比例子代码
说说硬件：缓存
CPU和内存之间存在着多级缓存，我们都知道越靠近CPU的缓存越快，存储速度依次排列为L1,L2,L3,Memery,但是他们的存储空间大小依次排列为倒序，因此我们不能把所有数据都放在L1，我们需要把我们的数据从Memery中Load到cache里面，从而进行访问。下图有一些数据：
如果你想让你端到端的延时为10ms,那你mermery中load耗时80ns相对来说是一个比较重的操作。更为严重的问题是，我们的数据在缓存中，不是独立项存储的，你可以想象缓存为一个阵列，由多个缓存行组成，缓存行大小根据机器不同，有差异，常见为64个字节。每次LRU（或者其他算法）的时候，它会把你目标数据的相邻数据也load进来，放入一整行的缓存中。
现在假设我们要操作A.B两个数据，他们正好在内存中是紧挨着的，线程1想要对数据A进行写入操作，它把A从Memery中load到L1来，相应的B也被免费的load到L1中。
现在线程1需要对A进行写入，同时线程2需要对B进行写入，他们需要争夺对这个缓存行的所有权，加入线程1成功对A进行了写入，那线程2需要对自己的缓存置为失效。通过这样的一个方式，说明了两个不相关的线程，本来操作自己的数据，但是因为另外一个线程对自己数据的更改，导致自己的数据需要重新从Memery中load，这样会把自己的整体速度给拖慢。
这就是伪共享，因为每次你访问A的同时，你也会得到B,而且每次你访问B，同时你也会得到A。他们仿佛是一体的，但是实际没有任何关系。
disruptor怎么做
引入缓存行填充机制，在RingBuffer中，需要有一个指向当前数据区的序列号（Sequencer），在有生成者和消费者对RingBuffer进行，数据读写的时候，我们对这个序列号进行缓存行填充机制，保证一个序列号在内存中，占有一个缓冲行。
通过代码演示，我们也可以看到的确存在差异，而且随着读写的线程变多，这样的差距越大。
false share code官方给到的代码，没有进行完全填充（也就是没有沾满一个缓存行），我自己写的例子有进行改进。
在典型的消费者/生成者模型中，会存在等待现象，disruptor提供了以下的几种等待策略：
BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现
SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；
YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。经测试前两种效果差不多,延迟在微秒以内,可以忽略,cpu占用不高,YieldingWaitStrategy模式队列空闲时CPU达到100%,不适合
简单例子直接使用ringbuffer
public static void main(String[] args) {
int size = 1&&10;
ExecutorService executors = Executors.newCachedThreadPool();
//创建一个disruptor，指定ringbuffer的size和处理数据的factory
Disruptor&TestObject& disruptor = new Disruptor&TestObject&(new TestObjectFactory(), size, executors);
//disruptor里面设置一个处理方式
disruptor.handleEventsWith(new TestObjectHandler());
RingBuffer&TestObject& ringBuffer = disruptor.start();
for (long i = 0; i & 1000; i++) {
//下一个可以用的序列号
long seq = ringBuffer.next();
//这个序列号的slot 放入数据
TestObject valueEvent = ringBuffer.get(seq);
valueEvent.setValue(i);
} finally {
//发布通知，并且这一步一定要放在finally中，因为调用了ringBuffer.next(),就一定要发布，否则会导致disruptor状态的错乱
ringBuffer.publish(seq);
disruptor.shutdown();
executors.shutdown();
这个例子中，我们需要有不同的消费者，并且有些消费者之间存在依赖关系，有些消费者之间可以并行处理。
public static void main(String[] args) throws InterruptedException {
long beginTime=System.currentTimeMillis();
int bufferSize=4;
ExecutorService executor= Executors.newFixedThreadPool(10);//大于consumer的数量
Disruptor&TestObject& disruptor = new Disruptor&TestObject&(new TestObjectFactory(), bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());
//使用disruptor创建消费者AnalysisHandler,CalcHandler，两个可以并行执行
EventHandlerGroup&TestObject& handlerGroup=disruptor.handleEventsWith(new TestObjectAnalysisHandler(),new TestObjectCalcHandler());
//声明在AnalysisHandler,CalcHandler完事之后执行NotifyHandler
EventHandlerGroup&TestObject& then = handlerGroup.then(new TestObjectNotifyHandler());
//最终调用写入DB的handler,这里有启用多个线程，进行数据的写入
then.thenHandleEventsWithWorkerPool(new TestObjectDBHandler(),new TestObjectDBHandler());
//上面的也可以直接通过链式调用
//disruptor.handleEventsWith(new TestObjectAnalysisHandler(),new TestObjectCalcHandler()).then(new TestObjectNotifyHandler()).thenHandleEventsWithWorkerPool(new TestObjectDBHandler(),new TestObjectDBHandler());
disruptor.start();//启动
CountDownLatch latch=new CountDownLatch(1);
//生产者准备
executor.submit(new TestObjectPublisher(latch, disruptor));
latch.await();//等待生产者完事.
disruptor.shutdown();
executor.shutdown();
System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));
个人思考下来，它适合一切异步环境，但是对于并发量小的场景不一定需要。在log4j2中，已经使用了disruptor进行日志记录。同样是用异步，选择disruptor会更快。
在一些获取验证码，发短信的场景下，对实时性要求不够，如果收不到，用户可以再次要求重发。
对于一些奖品，卡券的发放，在高峰期，可以只入队，在之后用异步的方式慢慢发放。
对于比较复杂的逻辑可以进行并发操作
disruptor作为一个高并发框架，从CPU层面对整个代码进行优化。具有如下特点
队列使用数组结构，而不是使用传统的链表结构，寻址更快
新生产的对象采用覆盖的方式(不是传统阻塞队列，删除-&添加的逻辑)，减少GC回收的负担
从CPU层面优化，对Sequencer进行内存分配补齐，消除Java伪共享(cpu缓存行)
多个线程同时访问，由于他们都通过序号器Sequencer访问ringBuffer，通过CAS取代了加锁和同步块，这也是并发编程的一个指导性原则：把同步块最小化到一个变量上。
False Sharing
Getting-Started
Q1:通过Sequencer产生的序列号，一直处于自增状态，是否会爆掉
A1:不会，long的范围最大可以达到4775807，一年365
60 * 60 = 秒，每秒产生1W条数据，也可以使用292年
热点阅读：
小主，按键盘右方向键 → 翻页可以跳过片头呢
本文标题：
原文链接：
和本文相似的内容：
编辑推荐 &&
妖怪研究所系列其一：山童? 李家怪事绵延大青山，茫茫烟苍几百里，这里的树木相当繁茂，遮天蔽日，林中更是鹿鸣呦呦，空山绝响，充满了无限的生机和神秘。大青山下大青庄，有山民100余户。俗话说，坐山吃山，这里的人们多为猎户，平日里青壮男子外出打猎& & & 已经不记得最早接触到 Disruptor 是什么时候了，只记得发现它的时候它是以具有闪电般的速度被介绍的。于是在脑子里， Disruptor 和&闪电&一词关联了起来，然而却一直没有时间去探究一下。
& & & 最近正在进行一项对性能有很高要求的产品项目的研究，自然想起了闪电般的 Disruptor ，这必有它的用武之地，于是进行了一番探查，将成果和体会记录在案。
一、什么是 Disruptor&
从功能上来看，Disruptor 是实现了&队列&的功能，而且是一个有界队列。那么它的应用场景自然就是&生产者-消费者&模型的应用场合了。
可以拿 JDK 的 BlockingQueue 做一个简单对比，以便更好地认识 Disruptor 是什么。
我们知道 BlockingQueue 是一个 FIFO 队列，生产者(Producer)往队列里发布(publish)一项事件(或称之为&消息&也可以)时，消费者(Consumer)能获得通知；如果没有事件时，消费者被堵塞，直到生产者发布了新的事件。
这些都是 Disruptor 能做到的，与之不同的是，Disruptor 能做更多：
同一个&事件&可以有多个消费者，消费者之间既可以并行处理，也可以相互依赖形成处理的先后次序(形成一个依赖图)；
预分配用于存储事件内容的内存空间；
针对极高的性能目标而实现的极度优化和无锁的设计；
以上的描述虽然简单地指出了 Disruptor 是什么，但对于它&能做什么&还不是那么直截了当。一般性地来说，当你需要在两个独立的处理过程(两个线程)之间交换数据时，就可以使用 Disruptor 。当然使用队列（如上面提到的 BlockingQueue）也可以，只不过 Disruptor 做得更好。
拿队列来作比较的做法弱化了对 Disruptor 有多强大的认识，如果想要对此有更多的了解，可以仔细看看 Disruptor 在其东家 LMAX 交易平台(也是实现者) 是如何作为核心架构来使用的，这方面就不做详述了，问度娘或谷哥都能找到。
二、Disruptor 的核心概念
先从了解 Disruptor 的核心概念开始，来了解它是如何运作的。下面介绍的概念模型，既是领域对象，也是映射到代码实现上的核心对象。
Ring Buffer如其名，环形的缓冲区。曾经 RingBuffer 是 Disruptor 中的最主要的对象，但从3.0版本开始，其职责被简化为仅仅负责对通过 Disruptor 进行交换的数据（事件）进行存储和更新。在一些更高级的应用场景中，Ring Buffer 可以由用户的自定义实现来完全替代。
Sequence &Disruptor通过顺序递增的序号来编号管理通过其进行交换的数据（事件），对数据(事件)的处理过程总是沿着序号逐个递增处理。一个 Sequence 用于跟踪标识某个特定的事件处理者(&RingBuffer/Consumer&)的处理进度。虽然一个 AtomicLong 也可以用于标识进度，但定义 Sequence 来负责该问题还有另一个目的，那就是防止不同的 Sequence 之间的CPU缓存伪共享(Flase Sharing)问题。（注：这是 Disruptor 实现高性能的关键点之一，网上关于伪共享问题的介绍已经汗牛充栋，在此不再赘述）。
Sequencer&Sequencer 是 Disruptor 的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer ，它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。
Sequence Barrier用于保持对RingBuffer的 main published Sequence 和Consumer依赖的其它Consumer的 Sequence 的引用。&Sequence Barrier 还定义了决定 Consumer 是否还有可处理的事件的逻辑。
Wait Strategy定义 Consumer 如何进行等待下一个事件的策略。 （注：Disruptor 定义了多种不同的策略，针对不同的场景，提供了不一样的性能表现）
Event在 Disruptor 的语义中，生产者和消费者之间进行交换的数据被称为事件(Event)。它不是一个被 Disruptor 定义的特定类型，而是由 Disruptor 的使用者定义并指定。
EventProcessorEventProcessor 持有特定消费者(Consumer)的 Sequence，并提供用于调用事件处理实现的事件循环(Event Loop)。
EventHandlerDisruptor 定义的事件处理接口，由用户实现，用于处理事件，是 Consumer 的真正实现。
Producer即生产者，只是泛指调用 Disruptor 发布事件的用户代码，Disruptor 没有定义特定接口或类型。
三、如何使用 Disruptor&
Disruptor 的 API 十分简单，主要有以下几个步骤：
定义事件事件(Event)就是通过 Disruptor 进行交换的数据类型。
public class LongEvent
private long
public void set(long value)
this.value =
定义事件工厂事件工厂(Event Factory)定义了如何实例化前面第1步中定义的事件(Event)，需要实现接口&com.lmax.disruptor.EventFactory&T&。Disruptor 通过 EventFactory 在 RingBuffer 中预创建 Event 的实例。一个 Event 实例实际上被用作一个&数据槽&，发布者发布前，先从 RingBuffer 获得一个 Event 的实例，然后往 Event 实例中填充数据，之后再发布到 RingBuffer 中，之后由 Consumer 获得该 Event 实例并从中读取数据。
import com.lmax.disruptor.EventF
public class LongEventFactory implements EventFactory&LongEvent&
public LongEvent newInstance()
return new LongEvent();
定义事件处理的具体实现通过实现接口 com.lmax.disruptor.EventHandler&T&&定义事件处理的具体实现。
import com.lmax.disruptor.EventH
public class LongEventHandler implements EventHandler&LongEvent&
public void onEvent(LongEvent event, long sequence, boolean endOfBatch)
System.out.println("Event: " + event);
定义用于事件处理的线程池Disruptor 通过&java.util.concurrent.ExecutorService 提供的线程来触发 Consumer 的事件处理。例如：
ExecutorService executor = Executors.newCachedThreadPool();
指定等待策略Disruptor 定义了 com.lmax.disruptor.WaitStrategy 接口用于抽象&Consumer 如何等待新事件，这是策略模式的应用。Disruptor 提供了多个&WaitStrategy 的实现，每种策略都具有不同性能和优缺点，根据实际运行环境的 CPU 的硬件特点选择恰当的策略，并配合特定的 JVM 的配置参数，能够实现不同的性能提升。例如，BlockingWaitStrategy、SleepingWaitStrategy、YieldingWaitStrategy 等，其中，BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；SleepingWaitStrategy 的性能表现跟&BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。
WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();
WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();
WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
启动 Disruptor
EventFactory&LongEvent& eventFactory = new LongEventFactory();
ExecutorService executor = Executors.newSingleThreadExecutor();
int ringBufferSize = 1024 * 1024; // RingBuffer 大小，必须是 2 的 N 次方；
Disruptor&LongEvent& disruptor = new Disruptor&LongEvent&(eventFactory,
ringBufferSize, executor, ProducerType.SINGLE,
new YieldingWaitStrategy());
EventHandler&LongEvent& eventHandler = new LongEventHandler();
disruptor.handleEventsWith(eventHandler);
disruptor.start();
发布事件Disruptor 的事件发布过程是一个两阶段提交的过程：　　第一步：先从 RingBuffer 获取下一个可以写入的事件的序号；　　第二步：获取对应的事件对象，将数据写入事件对象；　　第三部：将事件提交到 RingB事件只有在提交之后才会通知 EventProcessor 进行处理；
// 发布事件；
RingBuffer&LongEvent& ringBuffer = disruptor.getRingBuffer();
long sequence = ringBuffer.next();//请求下一个事件序号；
LongEvent event = ringBuffer.get(sequence);//获取该序号对应的事件对象；
long data = getEventData();//获取要通过事件传递的业务数据；
event.set(data);
} finally{
ringBuffer.publish(sequence);//发布事件；
&注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
Disruptor 还提供另外一种形式的调用来简化以上操作，并确保 publish 总是得到调用。
static class Translator implements EventTranslatorOneArg&LongEvent, Long&{
public void translateTo(LongEvent event, long sequence, Long data) {
event.set(data);
public static Translator TRANSLATOR = new Translator();
public static void publishEvent2(Disruptor&LongEvent& disruptor) {
// 发布事件；
RingBuffer&LongEvent& ringBuffer = disruptor.getRingBuffer();
long data = getEventData();//获取要通过事件传递的业务数据；
ringBuffer.publishEvent(TRANSLATOR, data);
此外，Disruptor 要求 RingBuffer.publish 必须得到调用的潜台词就是，如果发生异常也一样要调用 publish ，那么，很显然这个时候需要调用者在事件处理的实现上来判断事件携带的数据是否是正确的或者完整的，这是实现者应该要注意的事情。
关闭 Disruptor
disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；
四、性能对比测试
　　为了直观地感受 Disruptor 有多快，设计了一个性能对比测试：Producer 发布 100 万次事件，从发布第一个事件开始计时，捕捉 Consumer 处理完所有事件的耗时。
　　测试用例在 Producer 如何将事件通知到 Consumer 的实现方式上，设计了三种不同的实现：
Producer 的事件发布和 Consumer 的事件处理都在同一个线程，Producer 发布事件后立即触发 Consumer 的事件处理；
Producer 的事件发布和 Consumer 的事件处理在不同的线程，通过 ArrayBlockingQueue 传递给 Consumer 进行处理；
Producer 的事件发布和 Consumer 的事件处理在不同的线程，通过 Disruptor 传递给 Consumer 进行处理；
此次测试用例仅做了只有一个 Producer 和一个 Consumer 的情形，测试用例的代码如下：
CounterTracer tracer = tracerFactory.newInstance(DATA_COUNT);//计数跟踪到达指定的数值；
TestHandler handler = new TestHandler(tracer);//Consumer 的事件处理；
EventPublisher publisher = publisherFactory.newInstance(new PublisherCreationArgs(DATA_COUNT, handler));//通过工厂对象创建不同的 Producer 的实现；
publisher.start();
tracer.start();
//发布事件；
for (int i = 0; i & DATA_COUNT; i++) {
publisher.publish(i);
//等待事件处理完成；
tracer.waitForReached();
publisher.stop();
//输出结果；
printResult(tracer);
事件处理的实现只是调用一个计数器(CounterTracer)加1，该计数器跟踪从开始到达到总的事件次数时所耗的时间。
public class TestHandler {
private CounterT
public TestHandler(CounterTracer tracer) {
this.tracer =
* 如果返回 true，则表示处理已经全部完成，不再处理后续事件；
* @param event
public boolean process(TestEvent event){
return tracer.count();
针对单一Producer 和单一 Consumer 的测试场景，CounterTracer 的实现如下：
* 测试结果跟踪器，计数器不是线程安全的，仅在单线程的 consumer 测试中使用；
* @author haiq
public class SimpleTracer implements CounterTracer {
private long startT
private long endT
private long count = 0;
private boolean end = false;
private final long expectedC
private CountDownLatch latch = new CountDownLatch(1);
public SimpleTracer(long expectedCount) {
this.expectedCount = expectedC
public void start() {
startTicks = System.currentTimeMillis();
end = false;
public long getMilliTimeSpan() {
return endTicks - startT
public boolean count() {
if (end) {
end = count &= expectedC
if (end) {
endTicks = System.currentTimeMillis();
latch.countDown();
public void waitForReached() throws InterruptedException {
latch.await();
第一种 Producer 的实现：直接触发事件处理；
public class DirectingPublisher implements EventPublisher {
private TestH
private TestEvent event = new TestEvent();
public DirectingPublisher(TestHandler handler) {
this.handler =
public void publish(int data) throws Exception {
event.setValue(data);
handler.process(event);
//省略其它代码；
第二种 Producer 的实现：通过 ArrayBlockinigQueue 实现；
public class BlockingQueuePublisher implements EventPublisher {
private ArrayBlockingQueue&TestEvent&
private TestH
public BlockingQueuePublisher(int maxEventSize, TestHandler handler) {
this.queue = new ArrayBlockingQueue&TestEvent&(maxEventSize);
this.handler =
public void start(){
Thread thrd = new Thread(new Runnable() {
public void run() {
thrd.start();
private void handle(){
while (true) {
evt = queue.take();
if (evt != null && handler.process(evt)) {
//完成后自动结束处理线程；
} catch (InterruptedException e) {
e.printStackTrace();
public void publish(int data) throws Exception {
TestEvent evt = new TestEvent();
evt.setValue(data);
queue.put(evt);
//省略其它代码；
第三种 Producer 的实现：通过 Disruptor 实现；
public class DisruptorPublisher implements EventPublisher {
private class TestEventHandler implements EventHandler&TestEvent& {
private TestH
public TestEventHandler(TestHandler handler) {
this.handler =
public void onEvent(TestEvent event, long sequence, boolean endOfBatch)
throws Exception {
handler.process(event);
private static final WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();
private Disruptor&TestEvent&
private TestEventH
private RingBuffer&TestEvent&
private ExecutorS
public DisruptorPublisher(int bufferSize, TestHandler handler) {
this.handler = new TestEventHandler(handler);
executor = Executors.newSingleThreadExecutor();
disruptor = new Disruptor&TestEvent&(EVENT_FACTORY, bufferSize,
executor, ProducerType.SINGLE,
YIELDING_WAIT);
@SuppressWarnings("unchecked")
public void start() {
disruptor.handleEventsWith(handler);
disruptor.start();
ringbuffer = disruptor.getRingBuffer();
public void publish(int data) throws Exception {
long seq = ringbuffer.next();
TestEvent evt = ringbuffer.get(seq);
evt.setValue(data);
} finally {
ringbuffer.publish(seq);
//省略其它代码；
Producer 第一种实现并没有线程间的交换，实际上就是直接调用计数器，因此以此种实现的测试结果作为基准，对比其它的两种实现的测试结果。
在我的CPU CORE i5 / 4G 内存 / Win7 64 位的笔记本上，数据量(DATA_COUNT)取值为 1024 * 1024 时的测试结果如下：
【基准测试】
[1]--每秒吞吐量：--；(1048576/0ms)
[2]--每秒吞吐量：--；(1048576/0ms)
[3]--每秒吞吐量：--；(1048576/0ms)
[4]--每秒吞吐量：；(1048576/15ms)
[5]--每秒吞吐量：--；(1048576/0ms)
【对比测试1: ArrayBlockingQueue 实现】
[1]--每秒吞吐量：4788018；(1048576/219ms)
[2]--每秒吞吐量：5165399；(1048576/203ms)
[3]--每秒吞吐量：4809981；(1048576/218ms)
[4]--每秒吞吐量：5165399；(1048576/203ms)
[5]--每秒吞吐量：5577531；(1048576/188ms)
【对比测试2: Disruptor实现】
[1]--每秒吞吐量：；(1048576/31ms)
[2]--每秒吞吐量：；(1048576/16ms)
[3]--每秒吞吐量：；(1048576/16ms)
[4]--每秒吞吐量：；(1048576/15ms)
[5]--每秒吞吐量：；(ms)
从测试结果看， Disruptor 的性能比 ArrayBlockingQueue 高出了几乎一个数量级，操作耗时也只有平均20毫秒左右。
由于篇幅有限，关于 Disruptor 实现高性能的原理，留待以后再做探讨。
六、参考资料
Diruptor 页面：/LMAX-Exchange/disruptor
阅读(...) 评论()