Java并发编程---Disruptor体验

本文主要是介绍Java并发编程---Disruptor体验，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

最近在学习中接触到Disruptor这个框架，虽然目前没有能实际运用到项目中，但是做个了解，在面试吹牛逼?的时候还能避免尴尬！学的不深，仅限于简单的使用和特性的认识。

什么是Disruptor

---

• Martin Fowler在自己网站上写了一篇LMAX架构的文章，在文章中他介绍了LMAX是一种新型零售金融交易平台，它能够以很低的延迟产生大量交易。这个系统是建立在JVM平台上，其核心是一个业务逻辑处理器，它能够在一个线程里每秒处理6百万订单。业务逻辑处理器完全是运行在内存中，使`用事件源驱动方式。业务逻辑处理器的核心是Disruptor。

• Disruptor它是一个开源的并发框架，并获得2011 Duke’s 程序框架创新奖，能够在无锁的情况下实现网络的Queue并发操作。Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架（轻量的JMS），也可以认为是一个观察者模式的实现，或者事件监听模式的实现。

从功能上来看，Disruptor 是实现了“队列”的功能，而且是一个有界队列。那么它的应用场景自然就是“生产者-消费者”模型的应用场合了。
可以拿 JDK 的 BlockingQueue 做一个简单对比，以便更好地认识 Disruptor 是什么。

我们知道 BlockingQueue 是一个 FIFO 队列，生产者(Producer)往队列里发布(publish)一项事件(或称之为“消息”也可以)时，消费者(Consumer)能获得通知；如果没有事件时，消费者被堵塞，直到生产者发布了新的事件。

BlockingQueue是基于锁实现的, 而锁的效率通常较低. 没有使用CAS机制实现。而Disruptor使用观察者模式, 主动将消息发送给消费者, 而不是等消费者从队列中取; 在无锁的情况下, 实现queue(环形, RingBuffer)的并发操作, 性能远高于BlockingQueue

Disruptor的设计方案

Disruptor通过以下设计来解决队列速度慢的问题：

• 环形数组结构
  为了避免垃圾回收，采用数组而非链表。同时，数组对处理器的缓存机制更加友好。
• 元素位置定位
  数组长度2^n，通过位运算，加快定位的速度。下标采取递增的形式。不用担心index溢出的问题。index是long类型，即使100万QPS的处理速度，也需要30万年才能用完。
• 无锁设计
  每个生产者或者消费者线程，会先申请可以操作的元素在数组中的位置，申请到之后，直接在该位置写入或者读取数据。

Disruptor核心

• Ring Buffer
  Disruptor底层数据结构实现，核心类，是线程间交换数据的中转地
• 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 的使用者定义并指定。
• EventProcessor
  EventProcessor 持有特定消费者(Consumer)的 Sequence，并提供用于调用事件处理实现的事件循环(Event Loop)。
• EventHandler
  Disruptor 定义的事件处理接口，由用户实现，用于处理事件，是 Consumer 的真正实现。
• Producer
  即生产者，只是泛指调用 Disruptor 发布事件的用户代码，Disruptor 没有定义特定接口或类型。

什么是RingBuffer

它是一个环（首尾相接的环），你可以把它用做在不同上下文（线程）间传递数据的buffer。

随着不停地填充这个buffer（可能也会有相应的读取），这个序号会一直增长，直到超过这个环。

Disruptor要求数组大小设置为2的N次方。这样可以通过Seq & (QueueSize - 1) 直接获取，其效率要比取模快得多。这是因为（Queue - 1）的二进制为全1等形式。例如，上图中QueueSize大小为8，Seq为10，则只需要计算二进制1010 & 0111 = 2，可直接得到index=2位置的元素。

在RingBuffer中，生产者向数组中写入数据，生产者写入数据时，使用CAS操作。消费者从中读取数据时，为防止多个消费者同时处理一个数据，也使用CAS操作进行数据保护。

Disruptor简单使用

1. 引入maven依赖或导入jar包

	<dependencies><dependency><groupId>com.lmax</groupId><artifactId>disruptor</artifactId><version>3.2.1</version></dependency></dependencies>

2. 定义事件

// 定义事件event  通过  Disruptor 进行交换的数据类型。
public class LongEvent {private long value;public long getValue() {return value;}public void setValue(long value) {this.value = value;}}

3. 定义事件工厂
   事件工厂(Event Factory)定义了如何实例化前面第1步中定义的事件(Event)，需要实现接口 com.lmax.disruptor.EventFactory。
   Disruptor 通过 EventFactory 在 RingBuffer 中预创建 Event 的实例。
   一个 Event 实例实际上被用作一个“数据槽”，发布者发布前，先从 RingBuffer 获得一个 Event 的实例，然后往 Event 实例中填充数据，之后再发布到 RingBuffer 中，之后由 Consumer 获得该 Event 实例并从中读取数据。

//EventFactory来实例化Event对象。
public class LongEventFactory implements EventFactory<LongEvent> {public LongEvent newInstance() {// TODO Auto-generated method stubreturn new LongEvent();}}

4. 定义事件处理的具体实现
   通过实现接口 com.lmax.disruptor.EventHandler 定义事件处理的具体实现。

// 事件消费者 这个事件处理器简单地把事件中存储的数据打印到终端：
public class LongEventHandler  implements EventHandler<LongEvent>{public void onEvent(LongEvent arg0, long arg1, boolean arg2) throws Exception {System.out.println("消费者:"+arg0.getValue());}}

5.定义生产者发送数据

//定义生产这发送事件
public class LongEventProducer {public final RingBuffer<LongEvent> ringBuffer;public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {this.ringBuffer = ringBuffer;}public void onData(ByteBuffer byteBuffer) {// 1.ringBuffer 事件队列 下一个槽long sequence = ringBuffer.next();Long data = null;try {// 2.取出空的事件队列LongEvent longEvent = ringBuffer.get(sequence);data = byteBuffer.getLong(0);// 3.获取事件队列传递的数据longEvent.setValue(data);try {Thread.sleep(10);} catch (InterruptedException e) {// TODO Auto-generated catch blocke.printStackTrace();}} finally {System.out.println("生产这准备发送数据");// 4.发布事件ringBuffer.publish(sequence);}}}

6.启动 Disruptor

public class DisruptorMain {public static void main(String[] args) {// 1.创建一个可缓存的线程 提供线程来给Consumer 的事件处理ExecutorService executor = Executors.newCachedThreadPool();// 2.创建工厂EventFactory<LongEvent> eventFactory = new LongEventFactory();// 3.创建ringBuffer 大小int ringBufferSize = 1024 * 1024; // ringBufferSize大小一定要是2的N次方// 4.创建Disruptor@SuppressWarnings("deprecation")Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(eventFactory, ringBufferSize, executor,ProducerType.SINGLE, new YieldingWaitStrategy());// 5.连接消费端方法disruptor.handleEventsWith(new LongEventHandler());// 多个消费者使用是重复取值并不是均摊// 6.启动disruptor.start();// 7.创建RingBuffer容器RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();// 8.创建生产者LongEventProducer producer = new LongEventProducer(ringBuffer);// 9.指定缓冲区大小ByteBuffer byteBuffer = ByteBuffer.allocate(8);for (int i = 1; i <= 100; i++) {byteBuffer.putLong(0, i);producer.onData(byteBuffer);}// 10.关闭disruptor和executordisruptor.shutdown();executor.shutdown();}}

特殊说明：

• BlockingWaitStrategy 是最低效的策略，但其对CPU的消耗最小并且在各种不同部署环境中能提供更加一致的性能表现；
• SleepingWaitStrategy 的性能表现跟 BlockingWaitStrategy 差不多，对 CPU 的消耗也类似，但其对生产者线程的影响最小，适合用于异步日志类似的场景；
• YieldingWaitStrategy 的性能是最好的，适合用于低延迟的系统。在要求极高性能且事件处理线数小于 CPU 逻辑核心数的场景中，推荐使用此策略；例如，CPU开启超线程的特性。

总结

---

理解不深，有待完善！

附录：

Disruptor GitHub地址: https://github.com/LMAX-Exchange/disruptor

这篇关于Java并发编程---Disruptor体验的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！

---