Disruptor入门

Posted 2020-10-04 Ruthless

一、什么是 Disruptor
Disruptor是一个高性能的异步处理框架，或者可以认为是最快的消息框架(轻量的JMS)，也可以认为是一个观察者模式实现，或者事件-监听模式的实现，直接称disruptor模式。disruptor最大特点是高性能，其LMAX架构可以获得每秒6百万订单，用1微秒的延迟获得吞吐量为100K+。

可以理解为消费者-生产者的消息发布订阅模式。

二、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 的使用者定义并指定。

EventProcessor
EventProcessor 持有特定消费者(Consumer)的 Sequence，并提供用于调用事件处理实现的事件循环(Event Loop)。

EventHandler
Disruptor 定义的事件处理接口，由用户实现，用于处理事件，是 Consumer 的真正实现。

Producer
即生产者，只是泛指调用 Disruptor 发布事件的用户代码，Disruptor 没有定义特定接口或类型。

三、Disruptor入门示例代码，进行实验
jar版本：disruptor-3.2.1.jar

1、定义事件
事件(Event)就是通过 Disruptor 进行交换的数据类型。

package com.ljq.disruptor;

import java.io.Serializable;

/**
 * 定义事件数据，本质是个普通JavaBean
 * 
 * @author jqlin
 */
@SuppressWarnings("serial")
public class LongEvent implements Serializable {
	private long value;

	public LongEvent() {
		super();
	}

	public LongEvent(long value) {
		super();
		this.value = value;
	}

	public long getValue() {
		return value;
	}

	public void setValue(long value) {
		this.value = value;
	}

	@Override
	public String toString() {
		return "LongEvent [value=" + value + "]";
	}
	
	
}

　　

2、定义事件工厂
事件工厂(Event Factory)定义了如何实例化前面第1步中定义的事件(Event)，需要实现接口com.lmax.disruptor.EventFactory<T>。

Disruptor 通过 EventFactory 在 RingBuffer 中预创建 Event 的实例。一个 Event 实例实际上被用作一个“数据槽”，发布者发布前，先从 RingBuffer 获得一个 Event 的实例，然后往 Event 实例中填充数据，之后再发布到 RingBuffer 中，之后由 Consumer 获得该 Event 实例并从中读取数据。

package com.ljq.disruptor;

import com.lmax.disruptor.EventFactory;

/**
 * 定义事件工厂，实例化LongEvent事件
 * 
 * @author jqlin
 */
public class LongEventFactory implements EventFactory<LongEvent> {

	public LongEvent newInstance() {
		return new LongEvent();
	}

}

　　

3、LongEvent事件生产者
RingBuffer是消息存储结构，为环形存储结构，每个单元存储一条消息。类似于队列。当ringbuffer中数据填满后，环就会阻塞，等待消费者消费掉数据。当所有消费者消费掉环中一个数据，新的消息才可以加入环中。每个环插入数据后，都会分配下一个位置的编号，即sequence。

Disruptor的事件发布过程是一个两阶段提交的过程：
第一步：先从 RingBuffer 获取下一个可以写入的事件的序号；
第二步：获取对应的事件对象，将数据写入事件对象；
第三部：将事件提交到 RingBuffer;
事件只有在提交之后才会通知 EventProcessor 进行处理；

package com.ljq.disruptor;

import com.lmax.disruptor.RingBuffer;

/**
 * LongEvent事件生产者，生产LongEvent事件 
 * 
 * @author jqlin
 */
public class LongEventProducer {
	private final RingBuffer<LongEvent> ringBuffer;

	public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
		this.ringBuffer = ringBuffer;
	}

	public void produceData(long value) {
		long sequence = ringBuffer.next(); // 获得下一个Event槽的下标
		try {
			// 给Event填充数据
			LongEvent event = ringBuffer.get(sequence);
			event.setValue(value);
			
		} finally {
			// 发布Event，激活观察者去消费， 将sequence传递给该消费者
			// 注意，最后的 ringBuffer.publish() 方法必须包含在 finally 中以确保必须得到调用；如果某个请求的 sequence 未被提交，将会堵塞后续的发布操作或者其它的 producer。
			ringBuffer.publish(sequence);
		}
	}
}

　　

Disruptor还提供另外一种形式的调用来简化以上操作，并确保 publish 总是得到调用。

package com.ljq.disruptor;

import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;

/**
 * LongEvent事件生产者，Disruptor提供另外一种形式的调用来简化事件生产者的操作，并确保 publish 总是得到调用。
 * 
 * @author jqlin
 */
public class LongEventProducerWithTranslator {
	private final RingBuffer<LongEvent> ringBuffer;

	public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer) {
		this.ringBuffer = ringBuffer;
	}

	public void produceData(long value) {
		ringBuffer.publishEvent(TRANSLATOR, value);
	}

	// 使用EventTranslator, 封装获取Event的过程
	private static final EventTranslatorOneArg<LongEvent, Long> TRANSLATOR = new EventTranslatorOneArg<LongEvent, Long>() {
		@Override
		public void translateTo(LongEvent event, long sequeue, Long value) {
			event.setValue(value);
		}

	};
	
}

　　

translateTo方法将RingBuffer中的消息，转换成java对象格式。示例为LongEvent对象，后续消费者LongEventHandler处理器直接操作LongEvent对象，获取消息各属性信息，本示例 为value属性。

produceData()方法，将生产者生产的消息放入RingBuffer中。

4、LongEvent事件消息者
定义事件处理的具体实现，通过实现接口 com.lmax.disruptor.EventHandler<T> 定义事件处理的具体实现。

定义如何处理消息的地方，此处执行速度要足够快。否则，会影响RingBuffer后续没空间加入新的数据。因此，不能做业务耗时操作。建议另外开始 java 线程池处理消息。

package com.ljq.disruptor;

import com.lmax.disruptor.EventHandler;

/**
 * LongEvent事件消息者，消息LongEvent事件
 * 
 * @author Administrator
 *
 */
public class LongEventHandler implements EventHandler<LongEvent> {

	@Override
	public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {
		System.out.println("consumer:" + Thread.currentThread().getName() + " Event: value=" + event.getValue() + ",sequence=" + sequence + ",endOfBatch=" + endOfBatch);
	}

}

　　

5、LongEventMain
消费者-生产者启动类，其依靠构造Disruptor对象，调用start()方法完成启动线程。

指定等待策略
Disruptor 定义了 com.lmax.disruptor.WaitStrategy 接口用于抽象 Consumer 如何等待新事件，这是策略模式的应用。
Disruptor 提供了多个 WaitStrategy 的实现，每种策略都具有不同性能和优缺点，根据实际运行环境的 CPU 的硬件特点选择恰当的策略，并配合特定的 JVM 的配置参数，能够实现不同的性能提升。

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

package com.ljq.disruptor;

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 InterruptedException {
		long beginTime = System.currentTimeMillis();
		
		// 定义用于事件处理的线程池，Disruptor 通过 java.util.concurrent.ExecutorService 提供的线程来触发 Consumer 的事件处理
		ExecutorService executor = Executors.newCachedThreadPool();
		
		// 指定事件工厂
		LongEventFactory factory = new LongEventFactory();

		// 指定 ring buffer字节大小，必需为2的N次方(能将求模运算转为位运算提高效率 )，否则影响性能
		int bufferSize = 1024 * 1024;

		// 单线程模式，获取额外的性能
		Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(factory, bufferSize, executor, 
				ProducerType.SINGLE, new YieldingWaitStrategy());
		// 设置事件业务处理器---消费者
		disruptor.handleEventsWith(new LongEventHandler());
		// 启动disruptor线程
		disruptor.start();

		// 获取 ring buffer环，用于接取生产者生产的事件
		RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
		// 为 ring buffer指定事件生产者
		LongEventProducer producer = new LongEventProducer(ringBuffer);
		//LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);

		for (int i = 0; i<100000; i++) {
			producer.produceData(i);// 生产者生产数据
		}
		
		disruptor.shutdown(); //关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
		executor. shutdown(); //关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；

		System.out.println(String.format("总共耗时%s毫秒", (System.currentTimeMillis() - beginTime)));
	}
}