Disruptor：高性能低延迟的内存有界队列框架

Disruptor是用于在多个线程之间通信的高性能低延时的内存消息队列。其实现了“队列”的功能，而且是一个有界队列。它适合“生产者-消费者”模型的应用场合。

 

Disruptor是一个高性能的异步处理框架，也可以认为是一个观察者模式实现，或者事件-监听模式的实现。

 

它类似于ArrayBlockingQueue队列类。

 

高性能原理：

       引入环形的数组结构：缓存基于数组，用位运算替代求模。缓存的长度总是2的n次方，这样可以用位运算 i & (length - 1) 替代 i % length。预分配缓存对象，通过更新缓存里对象的属性而不是删除对象来减少垃圾回收。

        无锁的设计：采用内存屏障机制和原子性的CAS操作实现无锁。用内存屏障加序列号的方式实现了无锁的并发机制。

        属性填充：通过添加额外的无用信息，避免伪共享问题。

        元素位置的定位：采用跟一致性哈希一样的方式，一个索引，进行自增。

 

伪共享：

       CPU的缓存一般是以缓存行为最小单位的，对应主存的一块相应大小的单元；当前的缓存行大小一般是64字节，每个缓存行一次只能被一个CPU核访问，如果一个缓存行被多个CPU核访问，就会造成竞争，导致某个核必须等其他核处理完了才能继续处理，响应性能。

 

适用场景：

        适用于大规模低延迟的并发场景。

        可用于读写操作分离、数据缓存，速度匹配（因为其实现了生产者-消费者模型）。

        基于内存的事件流处理机制的场景。

 

生产者-消费者模型使用场景：

        一个生产者 — 一个消费者的场景

        一个生产者 — 多个消费者的场景

        多个生产者 — 一个消费者的场景

        多个生产者 — 多个消费者的场景

        生产者 - 消费流，消费者之间既可以并行处理，也可以相互依赖形成处理的先后次序(形成一个依赖图)。

 

核心概念：

         Ring Buffer：环形的缓冲区。负责对通过 Disruptor 进行交换的数据（事件）进行存储和更新。

 

         Sequence：通过顺序递增的序号来编号管理通过其进行交换的数据(事件)，对数据(事件)的处理过程总是沿着序号逐个递增处理。一个Sequence用于跟踪标识某个特定的事件处理者(RingBuffer/Consumer)的处理进度。

 

         Sequencer：是Disruptor的真正核心。此接口有两个实现类 SingleProducerSequencer、MultiProducerSequencer，它们定义在生产者和消费者之间快速、正确地传递数据的并发算法。

 

         Sequence Barrier：用于保持对RingBuffer的main published Sequence和Consumer依赖其它Consumer的Sequence的引用。

 

          WaitStrategy：决定了一个消费者将如何等待生产者将Event置入Disruptor。

 

          Event：生产者和消费者之间进行交换的数据被称为事件(Event)。

 

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

 

          EventHandler：事件处理接口，由用户实现。

 

          Producer：生产者，由用户实现

 

常用的等待策略：用于抽象Consumer如何等待新事件

        BlockingWaitStrategy

               内部使用一个锁和条件变量来控制线程的执行和等待。

               是最慢的等待策略，也是CPU使用率最低和最稳定的选项。

 

        SleepingWaitStrategy

               循环等待并且在循环中间调用LockSupport.parkNanos(1)来睡眠。

               优点：生产线程只需要计数，而不执行任何指令，且没有条件变量的消耗。

               缺点：事件对象从生产者到消费者传递的延迟变大了。

               应用场景：用在不需要低延迟，且事件发布对于生产者的影响比较小的情况下。比如异步日志功能。

 

         YieldingWaitStrategy

                循环等待sequence增加到合适的值，循环中调用Thread.yield()允许其他准备好的线程执行。

                该策略在减低系统延迟的同时也会增加CPU运算量。

                应用场景：需要高性能而且事件消费者线程比逻辑内核少的场景。

 

         BusySpinWaitStrategy

                是性能最高的等待策略，也是对部署环境要求最高的策略。

                应用场景：用在事件处理线程比物理内核数目还要小的场景。

 

消息 事件类类：

public class MessageEvent<T>{
	private T message;

	public T getMessage() {
		return message;
	}

	public void setMessage(T message) {
		this.message = message;
	}
}

 

消息 事件工厂类：

public class MessageEventFactory<T> implements EventFactory<MessageEvent<T>>{
	@Override
	public MessageEvent<T> newInstance() {
		return new MessageEvent<T>();
	}
}

 

异常处理类：

public class MessageExceptionHandler<T> implements ExceptionHandler<MessageEvent<T>> {
	@Override
	public void handleEventException(Throwable ex, long sequence, MessageEvent<T> event) {
		System.out.println("handleEventException: " + ex.toString());
	}
	
	@Override
	public void handleOnStartException(Throwable ex) {
		System.out.println("handleOnStartException: " + ex.toString());
	}
	
	@Override
	public void handleOnShutdownException(Throwable ex) {
		System.out.println("handleOnShutdownException: " + ex.toString());
	}
}

 

广播方式的消费者抽象类：

public abstract class AbstractBroadcastConsumer<T> implements EventHandler<MessageEvent<T>>{
	@Override
	public void onEvent(MessageEvent<T> event, long sequence, boolean endOfBatch) throws Exception {
		handleEvent(event, sequence, endOfBatch);
	}
	
	public abstract void handleEvent(MessageEvent<T> event, long sequence, boolean endOfBatch) throws Exception;
}

 

分组方式的消费者抽象类：

public abstract class AbstractGroupingConsumer<T> implements WorkHandler<MessageEvent<T>>{
	@Override
	public void onEvent(MessageEvent<T> event) throws Exception {
		handleEvent(event);
	}
	
	public abstract void handleEvent(MessageEvent<T> event) throws Exception;
}

 

消息生产和消费：

public class DisruptorTest {
	public static void main(String[] args) {
        int ringBufferSize = Double.valueOf(Math.pow(2, 15)).intValue(); //必须是2的N次方
        System.out.println("ringBufferSize=" + ringBufferSize);
        
		//ProducerType.SINGLE 单生产者
		Disruptor<MessageEvent<String>> disruptor
			= new Disruptor<>(new MessageEventFactory<String>(), 
					ringBufferSize, Executors.defaultThreadFactory(), 
					ProducerType.SINGLE, new YieldingWaitStrategy());
		
		//## 多消费者 ##
		//广播：同一个Event被所有消费者消费
		EventHandler<MessageEvent<String>> handler = new DefaultBroadcastConsumer();
		disruptor.handleEventsWith(handler, handler, handler); //无依赖关系
//		disruptor.handleEventsWith(handler, handler).then(handler); //有依赖关系
//		disruptor.after(handler).handleEventsWith(handler); //处理类有先后顺序
		
		//分组：一个Event只能被组内的一个消费者消费
		WorkHandler<MessageEvent<String>> handler2 = new DefaultGroupingConsumer();
		disruptor.handleEventsWithWorkerPool(handler2, handler2, handler2);
		
		disruptor.setDefaultExceptionHandler(new MessageExceptionHandler<String>());
		RingBuffer<MessageEvent<String>> ringBuffer = disruptor.start();
		
		while(true){
			publishEvent(ringBuffer, String.valueOf(++count));
		}
		disruptor.shutdown();
	}
	
	/**
	 * 发布消息
	 */
	private static void publishEvent(RingBuffer<MessageEvent<String>> ringBuffer, String message){
		ringBuffer.publishEvent((event, sequence, messageContent) -> {
			event.setMessage(messageContent);
		}, message);
	}
	
	/**
	 * 广播消费者
	 */
	public static class DefaultBroadcastConsumer extends AbstractBroadcastConsumer<String>{
		@Override
		public void handleEvent(MessageEvent<String> event, long sequence, boolean endOfBatch) throws Exception {
			System.out.println(Thread.currentThread().getName() + ": " + event.getMessage());
		}
	}

	/**
	 * 分组消费者
	 */
	public static class DefaultGroupingConsumer extends AbstractGroupingConsumer<String>{
		@Override
		public void handleEvent(MessageEvent<String> event) throws Exception {
			System.out.println(Thread.currentThread().getName() + ": " + event.getMessage());
		}
	}
}