Disruptor并发框架，核心组件RingBuffer

1.1 Disruptor并发框架简介

Martin Fowler在自己网站上写了一篇LMAX架构的文章，在文章中他介绍了LMAX是一种新型零售金融交易平台，它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上，其核心是一个业务逻辑处理器，它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中，使用事件源驱动方式。业务逻辑处理器的核心是Disruptor。
Disruptor它是一个开源的并发框架，并获得2011 Duke’s 程序框架创新奖，能够在无锁的情况下实现网络的Queue并发操作。
Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量的JMS），也可以认为是一个观察者模式的实现，或者事件监听模式的实现。

1.2 Disruptor并发框架使用

目前我们使用disruptor已经更新到了3.x版本，比之前的2.x版本性能更加的优秀，提供更多的API使用方式。
下载disruptor-3.3.2.jar引入我们的项目既可以开始disruptor之旅。
在使用之前，首先说明disruptor主要功能加以说明，你可以理解为他是一种高效的"生产者-消费者"模型。也就性能远远高于传统的BlockingQueue容器。
官方学习网站：http://ifeve.com/disruptor-getting-started/

在Disruptor中，我们想实现hello world 需要如下几步骤：
第一：建立一个Event类 //一个具体的数据
第二：建立一个工厂Event类，用于创建Event类实例对象 //数据类型规定死
第三：需要有一个监听事件类，用于处理数据（Event类）
第四：我们需要进行测试代码编写。实例化Disruptor实例，配置一系列参数。然后我们对Disruptor实例绑定监听事件类，接受并处理数据。
第五：在Disruptor中，真正存储数据的核心叫做RingBuffer，我们通过Disruptor实例拿到它，然后把数据生产出来，把数据加入到RingBuffer的实例对象中即可。

我们一起来看下这个HelloWorld程序：com.bjsxt.base
HelloWorld1:

package bhz.base;//http://ifeve.com/disruptor-getting-started/
public class LongEvent { private long value;public long getValue() { return value; } public void setValue(long value) { this.value = value; }
}

package bhz.base;import com.lmax.disruptor.EventFactory;
// 需要让disruptor为我们创建事件，我们同时还声明了一个EventFactory来实例化Event对象。
public class LongEventFactory implements EventFactory { @Override public Object newInstance() { return new LongEvent(); }
}

package bhz.base;import com.lmax.disruptor.EventHandler;//我们还需要一个事件消费者，也就是一个事件处理器。这个事件处理器简单地把事件中存储的数据打印到终端：
public class LongEventHandler implements EventHandler<LongEvent>  {@Overridepublic void onEvent(LongEvent longEvent, long l, boolean b) throws Exception {System.out.println(longEvent.getValue());}}

package bhz.base;import java.nio.ByteBuffer;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;public class LongEventMain {public static void main(String[] args) throws Exception {//创建缓冲池ExecutorService  executor = Executors.newCachedThreadPool();//创建工厂LongEventFactory factory = new LongEventFactory();//创建bufferSize ,也就是RingBuffer大小，必须是2的N次方int ringBufferSize = 1024 * 1024; // /**//BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现WaitStrategy BLOCKING_WAIT = new BlockingWaitStrategy();//SleepingWaitStrategy 的性能表现跟BlockingWaitStrategy差不多，对CPU的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景WaitStrategy SLEEPING_WAIT = new SleepingWaitStrategy();//YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于CPU逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性WaitStrategy YIELDING_WAIT = new YieldingWaitStrategy();*///创建disruptorDisruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, ringBufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());// 连接消费事件方法disruptor.handleEventsWith(new LongEventHandler());// 启动disruptor.start();//Disruptor 的事件发布过程是一个两阶段提交的过程：//发布事件RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();LongEventProducer producer = new LongEventProducer(ringBuffer); //LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);ByteBuffer byteBuffer = ByteBuffer.allocate(8);for(long l = 0; l<100; l++){byteBuffer.putLong(0, l);producer.onData(byteBuffer);//Thread.sleep(1000);}disruptor.shutdown();//关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；executor.shutdown();//关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；     }
}

package bhz.base;import java.nio.ByteBuffer;import com.lmax.disruptor.RingBuffer;
/*** 很明显的是：当用一个简单队列来发布事件的时候会牵涉更多的细节，这是因为事件对象还需要预先创建。* 发布事件最少需要两步：获取下一个事件槽并发布事件（发布事件的时候要使用try/finnally保证事件一定会被发布）。* 如果我们使用RingBuffer.next()获取一个事件槽，那么一定要发布对应的事件。* 如果不能发布事件，那么就会引起Disruptor状态的混乱。* 尤其是在多个事件生产者的情况下会导致事件消费者失速，从而不得不重启应用才能会恢复。*/
public class LongEventProducer {private final RingBuffer<LongEvent> ringBuffer;public LongEventProducer(RingBuffer<LongEvent> ringBuffer){this.ringBuffer = ringBuffer;}/*** onData用来发布事件，每调用一次就发布一次事件* 它的参数会用过事件传递给消费者*/public void onData(ByteBuffer bb){//1.可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽long sequence = ringBuffer.next();try {//2.用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）LongEvent event = ringBuffer.get(sequence);//3.获取要通过事件传递的业务数据event.setValue(bb.getLong(0));} finally {//4.发布事件//注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。ringBuffer.publish(sequence);}}
}

package bhz.base;import java.nio.ByteBuffer;import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;/*** Disruptor 3.0提供了lambda式的API。这样可以把一些复杂的操作放在Ring Buffer，* 所以在Disruptor3.0以后的版本最好使用Event Publisher或者Event Translator来发布事件* <B>系统名称：</B><BR>* <B>模块名称：</B><BR>* <B>中文类名：</B><BR>* <B>概要说明：</B><BR>* @author 北京尚学堂（alienware）* @since 2015年11月23日*/
public class LongEventProducerWithTranslator {//一个translator可以看做一个事件初始化器，publicEvent方法会调用它//填充Eventprivate static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR = new EventTranslatorOneArg<LongEvent, ByteBuffer>() {@Overridepublic void translateTo(LongEvent event, long sequeue, ByteBuffer buffer) {event.setValue(buffer.getLong(0));}};private final RingBuffer<LongEvent> ringBuffer;public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer) {this.ringBuffer = ringBuffer;}public void onData(ByteBuffer buffer){ringBuffer.publishEvent(TRANSLATOR, buffer);}
}

HelloWorld2:

package bhz.multi;import java.util.concurrent.atomic.AtomicInteger;import com.lmax.disruptor.WorkHandler;public class Consumer implements WorkHandler<Order>{private String consumerId;private static AtomicInteger count = new AtomicInteger(0);public Consumer(String consumerId){this.consumerId = consumerId;}@Overridepublic void onEvent(Order order) throws Exception {System.out.println("当前消费者: " + this.consumerId + "，消费信息：" + order.getId());count.incrementAndGet();}public int getCount(){return count.get();}}

package bhz.multi;import java.nio.ByteBuffer;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.ProducerType;public class Main {public static void main(String[] args) throws Exception {//创建ringBufferRingBuffer<Order> ringBuffer = RingBuffer.create(ProducerType.MULTI, new EventFactory<Order>() {  @Override  public Order newInstance() {  return new Order();  }  }, 1024 * 1024, new YieldingWaitStrategy());SequenceBarrier barriers = ringBuffer.newBarrier();Consumer[] consumers = new Consumer[3];for(int i = 0; i < consumers.length; i++){consumers[i] = new Consumer("c" + i);}WorkerPool<Order> workerPool = new WorkerPool<Order>(ringBuffer, barriers, new IntEventExceptionHandler(),consumers);ringBuffer.addGatingSequences(workerPool.getWorkerSequences());  workerPool.start(Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors()));  final CountDownLatch latch = new CountDownLatch(1);for (int i = 0; i < 100; i++) {  final Producer p = new Producer(ringBuffer);new Thread(new Runnable() {@Overridepublic void run() {try {latch.await();} catch (InterruptedException e) {e.printStackTrace();}for(int j = 0; j < 100; j ++){p.onData(UUID.randomUUID().toString());}}}).start();} Thread.sleep(2000);System.out.println("---------------开始生产-----------------");latch.countDown();Thread.sleep(5000);System.out.println("总数:" + consumers[0].getCount() );}static class IntEventExceptionHandler implements ExceptionHandler {  public void handleEventException(Throwable ex, long sequence, Object event) {}  public void handleOnStartException(Throwable ex) {}  public void handleOnShutdownException(Throwable ex) {}  }
}

package bhz.multi;public class Order {  private String id;//ID  private String name;private double price;//金额  public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public double getPrice() {return price;}public void setPrice(double price) {this.price = price;}}

package bhz.multi;import java.nio.ByteBuffer;
import java.util.UUID;import bhz.base.LongEvent;import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;/*** <B>系统名称：</B><BR>* <B>模块名称：</B><BR>* <B>中文类名：</B><BR>* <B>概要说明：</B><BR>* @author 北京尚学堂（alienware）* @since 2015年11月23日*/
public class Producer {private final RingBuffer<Order> ringBuffer;public Producer(RingBuffer<Order> ringBuffer){this.ringBuffer = ringBuffer;}/*** onData用来发布事件，每调用一次就发布一次事件* 它的参数会用过事件传递给消费者*/public void onData(String data){//可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽long sequence = ringBuffer.next();try {//用上面的索引取出一个空的事件用于填充（获取该序号对应的事件对象）Order order = ringBuffer.get(sequence);//获取要通过事件传递的业务数据order.setId(data);} finally {//发布事件//注意，最后的 ringBuffer.publish 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。ringBuffer.publish(sequence);}}}

helloWorld3:

package bhz.generate1;import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;import com.lmax.disruptor.BatchEventProcessor;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventProcessor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.YieldingWaitStrategy;public class Main1 {  public static void main(String[] args) throws Exception {  int BUFFER_SIZE=1024;  int THREAD_NUMBERS=4;  /* * createSingleProducer创建一个单生产者的RingBuffer， * 第一个参数叫EventFactory，从名字上理解就是"事件工厂"，其实它的职责就是产生数据填充RingBuffer的区块。 * 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率 * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略 */  final RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(new EventFactory<Trade>() {  @Override  public Trade newInstance() {  return new Trade();  }  }, BUFFER_SIZE, new YieldingWaitStrategy());  //创建线程池  ExecutorService executors = Executors.newFixedThreadPool(THREAD_NUMBERS);  //创建SequenceBarrier  SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();  //创建消息处理器  BatchEventProcessor<Trade> transProcessor = new BatchEventProcessor<Trade>(  ringBuffer, sequenceBarrier, new TradeHandler());  //这一步的目的就是把消费者的位置信息引用注入到生产者    如果只有一个消费者的情况可以省略 ringBuffer.addGatingSequences(transProcessor.getSequence());  //把消息处理器提交到线程池  executors.submit(transProcessor);  //如果存在多个消费者 那重复执行上面3行代码 把TradeHandler换成其它消费者类  Future<?> future= executors.submit(new Callable<Void>() {  @Override  public Void call() throws Exception {  long seq;  for(int i=0;i<10;i++){  seq = ringBuffer.next();//占个坑 --ringBuffer一个可用区块  ringBuffer.get(seq).setPrice(Math.random()*9999);//给这个区块放入 数据 ringBuffer.publish(seq);//发布这个区块的数据使handler(consumer)可见  }  return null;  }  }); future.get();//等待生产者结束  Thread.sleep(1000);//等上1秒，等消费都处理完成  transProcessor.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）  executors.shutdown();//终止线程  }
}

package bhz.generate1;import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.IgnoreExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.WorkHandler;
import com.lmax.disruptor.WorkerPool;public class Main2 {  public static void main(String[] args) throws InterruptedException {  int BUFFER_SIZE=1024;  int THREAD_NUMBERS=4;  EventFactory<Trade> eventFactory = new EventFactory<Trade>() {  public Trade newInstance() {  return new Trade();  }  };  RingBuffer<Trade> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE);  SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();  ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);  WorkHandler<Trade> handler = new TradeHandler();  WorkerPool<Trade> workerPool = new WorkerPool<Trade>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), handler);  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();  }
}

package bhz.generate1;import java.util.concurrent.atomic.AtomicInteger;public class Trade {  private String id;//ID  private String name;private double price;//金额  private AtomicInteger count = new AtomicInteger(0);public String getId() {return id;}public void setId(String id) {this.id = id;}public String getName() {return name;}public void setName(String name) {this.name = name;}public double getPrice() {return price;}public void setPrice(double price) {this.price = price;}public AtomicInteger getCount() {return count;}public void setCount(AtomicInteger count) {this.count = count;} }

package bhz.generate1;import java.util.UUID;import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;public class TradeHandler implements EventHandler<Trade>, WorkHandler<Trade> {  @Override  public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {  this.onEvent(event);  }  @Override  public void onEvent(Trade event) throws Exception {  //这里做具体的消费逻辑  event.setId(UUID.randomUUID().toString());//简单生成下ID  System.out.println(event.getId());  }
}

helloWorld4:

package bhz.generate2;import java.util.UUID;import bhz.generate1.Trade;import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;public class Handler1 implements EventHandler<Trade>,WorkHandler<Trade> {  @Override  public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {  this.onEvent(event);  }  @Override  public void onEvent(Trade event) throws Exception {  System.out.println("handler1: set name");event.setName("h1");Thread.sleep(1000);}
}

package bhz.generate2;import bhz.generate1.Trade;import com.lmax.disruptor.EventHandler;public class Handler2 implements EventHandler<Trade> {  @Override  public void onEvent(Trade event, long sequence,  boolean endOfBatch) throws Exception {  System.out.println("handler2: set price");event.setPrice(17.0);Thread.sleep(1000);}  }

package bhz.generate2;import bhz.generate1.Trade;import com.lmax.disruptor.EventHandler;public class Handler3 implements EventHandler<Trade> {@Override  public void onEvent(Trade event, long sequence,  boolean endOfBatch) throws Exception {  System.out.println("handler3: name: " + event.getName() + " , price: " + event.getPrice() + ";  instance: " + event.toString());}
}

package bhz.generate2;import java.util.UUID;import bhz.generate1.Trade;import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;public class Handler4 implements EventHandler<Trade>,WorkHandler<Trade> {  @Override  public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {  this.onEvent(event);  }  @Override  public void onEvent(Trade event) throws Exception {  System.out.println("handler4: get name : " + event.getName());event.setName(event.getName() + "h4");}
}

package bhz.generate2;import java.util.UUID;import bhz.generate1.Trade;import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;public class Handler5 implements EventHandler<Trade>,WorkHandler<Trade> {  @Override  public void onEvent(Trade event, long sequence, boolean endOfBatch) throws Exception {  this.onEvent(event);  }  @Override  public void onEvent(Trade event) throws Exception {  System.out.println("handler5: get price : " + event.getPrice());event.setPrice(event.getPrice() + 3.0);}
}

package bhz.generate2;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;import bhz.generate1.Trade;
import bhz.generate1.TradeHandler;import com.lmax.disruptor.BusySpinWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.EventHandlerGroup;
import com.lmax.disruptor.dsl.ProducerType;public class Main {  public static void main(String[] args) throws InterruptedException {  long beginTime=System.currentTimeMillis();  int bufferSize=1024;  ExecutorService executor=Executors.newFixedThreadPool(8);  Disruptor<Trade> disruptor = new Disruptor<Trade>(new EventFactory<Trade>() {  @Override  public Trade newInstance() {  return new Trade();  }  }, bufferSize, executor, ProducerType.SINGLE, new BusySpinWaitStrategy());  //菱形操作/**//使用disruptor创建消费者组C1,C2  EventHandlerGroup<Trade> handlerGroup = disruptor.handleEventsWith(new Handler1(), new Handler2());//声明在C1,C2完事之后执行JMS消息发送操作 也就是流程走到C3 handlerGroup.then(new Handler3());*///顺序操作/**disruptor.handleEventsWith(new Handler1()).handleEventsWith(new Handler2()).handleEventsWith(new Handler3());*///六边形操作. /**Handler1 h1 = new Handler1();Handler2 h2 = new Handler2();Handler3 h3 = new Handler3();Handler4 h4 = new Handler4();Handler5 h5 = new Handler5();disruptor.handleEventsWith(h1, h2);disruptor.after(h1).handleEventsWith(h4);disruptor.after(h2).handleEventsWith(h5);disruptor.after(h4, h5).handleEventsWith(h3);*/disruptor.start();//启动  CountDownLatch latch=new CountDownLatch(1);  //生产者准备  executor.submit(new TradePublisher(latch, disruptor));latch.await();//等待生产者完事. disruptor.shutdown();  executor.shutdown();  System.out.println("总耗时:"+(System.currentTimeMillis()-beginTime));  }
}

package bhz.generate2;import java.util.Random;
import java.util.concurrent.CountDownLatch;import bhz.generate1.Trade;import com.lmax.disruptor.EventTranslator;
import com.lmax.disruptor.dsl.Disruptor;public class TradePublisher implements Runnable {  Disruptor<Trade> disruptor;  private CountDownLatch latch;  private static int LOOP=10;//模拟百万次交易的发生  public TradePublisher(CountDownLatch latch,Disruptor<Trade> disruptor) {  this.disruptor=disruptor;  this.latch=latch;  }  @Override  public void run() {  TradeEventTranslator tradeTransloator = new TradeEventTranslator();  for(int i=0;i<LOOP;i++){  disruptor.publishEvent(tradeTransloator);}  latch.countDown();}
}  class TradeEventTranslator implements EventTranslator<Trade>{  private Random random=new Random();  @Override  public void translateTo(Trade event, long sequence) {  this.generateTrade(event);  }  private Trade generateTrade(Trade trade){  trade.setPrice(random.nextDouble()*9999);  return trade;  }  }