架构师入门笔记七并发框架Disruptor快速入门

分手后的思念是犯贱 2022-06-10 06:39 341阅读 0赞

![aHR0cHM6Ly9pbWcyMDIwLmNuYmxvZ3MuY29tL2Jsb2cvODA2OTU2LzIwMjAwNy84MDY5NTYtMjAyMDA3MzEyMzQ1MDc3MjItMTExNjgwOTkyMi5naWY][]

架构师入门笔记七 并发框架Disruptor快速入门

# 1. 什么是Disruptor #

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

# 2. HelloWorld代码 #

在生产者-消费者设计模型的中，采用有界队列BlockingQueue作为存储任务的容器，生产者将任务分配给容器，再由容器分配给消费者。Disruptor框架和该模型很相似。区别在于它用一个环形的RingBuffer作为存储任务的容器。消费者是主动从RingBuffer中获取数据而不是等待分配。从某种角度来说，Disruptor框架是升级版的生产者-消费者模型。

这里用一段代码学习 Disruptor框架，业务逻辑是把10000以内的数据全部打印出来

1） 首先是创建传递数据的Event（Event是从生产者到消费者过程中所处理的数据单元）类，该类是由用户定义。

/**
     * 第一步：创建一个数据单元Event
     * Event：从生产者到消费者过程中所处理的数据单元
     *
     */
    public class MyDataEvent {
    
    	private Long value;
    
    	public Long getValue() {
    		return value;
    	}
    
    	public void setValue(Long value) {
    		this.value = value;
    	}
    
    }

2）创建一个实例化Event的工厂类

import com.lmax.disruptor.EventFactory;
    
    /**
     * 第二步创建工厂类实例化Event
     * EventFactory 工厂，用于实例化Event类
     */
    public class MyDataEventFactory implements EventFactory<MyDataEvent>{
    
    	@Override
    	public MyDataEvent newInstance() {
    		return new MyDataEvent();
    	}
    
    }

3）创建一个事件处理器，也就是消费者，这里只做数据打印的事件。

import com.lmax.disruptor.EventHandler;
    
    /**
     * 第三步：消费端
     * EventHandler：消费者，也可以理解为事件处理器
     */
    public class MyDataEventHandler implements EventHandler<MyDataEvent>{
    
    	@Override
    	public void onEvent(MyDataEvent myDataEvent, long arg1, boolean arg2)
    			throws Exception {
    		// 处理事件 ....
    		System.out.println("处理事件，打印数据： " + myDataEvent.getValue());
    	}
    
    }

4）生产者发布事件

import com.lmax.disruptor.RingBuffer;
    
    /**
     * 第四步：生产端
     * 生产者
     */
    public class MyDataEventProducer {
    	
    	private final RingBuffer<MyDataEvent> ringBuffer; // 敲黑板！ 很重要的知识点
    
    	public MyDataEventProducer(RingBuffer<MyDataEvent> ringBuffer) {
    		this.ringBuffer = ringBuffer;
    	}
    	
    	/**
    	 * 发布事件，每调用一次就发布一次事件
         * 它的参数会通过事件传递给消费者
    	 * @param byteBuffer 用 byteBuffer传参 是考虑到 Disruptor 是消息框架，而ByteBuffer又是读取时信道 (SocketChannel)最常用的缓冲区
    	 */
    	public void publishData(ByteBuffer byteBuffer){
    		// RingBuffer 是一个圆环，.next() 方法是获取下一个索引值
    		long sequence = ringBuffer.next();
    		try {
    			// 通过索引值获取其对象
    			MyDataEvent myDataEvent = ringBuffer.get(sequence);
    			// 给数据单元赋值
    			myDataEvent.setValue(byteBuffer.getLong(0)); // byteBuffer 的一个方法，文章中有链接
    		} catch (Exception e) {
    			e.printStackTrace();
    		} finally {
    			// 发布事件，其实就是发布索引 ，发布方法必须放在finally 中，避免出现阻塞情况。
    			ringBuffer.publish(sequence);
    		}
    	}
    
    }

注意：

发布事件是两个步骤，第一步：先要从RingBuffer获取下一个事件槽（可以理解为索引），第二步再是发送事件。需要注意的是：获取的事件槽，就要发布该事件槽对应的事件。不然会出现混乱的情况。所以发布事件的代码要放在finally中。 java8的写法，文章底部有链接。

5）执行的Main方法，打印10000以内的数据

import java.nio.ByteBuffer;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import com.lmax.disruptor.RingBuffer;
    import com.lmax.disruptor.dsl.Disruptor;
    
    public class MyDataEventMain {
    
    	public static void main(String[] args) {
    		// step1 : 创建缓冲池
    		ExecutorService executor = Executors.newCachedThreadPool();
    		// step2 : 创建工厂
    		MyDataEventFactory factory = new MyDataEventFactory();
    		// step3 : 创建bufferSize ,也就是RingBuffer大小，必须是2的N次方
    	 	int ringBufferSize = 1024 * 1024;
    		
    	 	// step4 : 创建disruptor
    	 	Disruptor<MyDataEvent> disruptor = new Disruptor<MyDataEvent>(factory, ringBufferSize, executor);
    	 	
    	 	// step5 : 连接消费事件方法<消费者>
    	 	disruptor.handleEventsWith(new MyDataEventHandler());
    	 	
    	 	// step6 : 启动
    	 	disruptor.start();
    	 	
    	 	RingBuffer<MyDataEvent> ringBuffer = disruptor.getRingBuffer(); // 获取 ringBuffer
    	 	
    	 	// step7 : 生产者发布事件
    	 	MyDataEventProducer producer = new MyDataEventProducer(ringBuffer);
    	 	
    	 	ByteBuffer byteBuffer = ByteBuffer.allocate(128); // 创建一个容量为128字节的ByteBuffer
    	 	
    	 	for (long data = 1; data <= 10000 ; data++) { // 不管是打印100，1000，10000，基本上都是一秒内输出。
    			byteBuffer.putLong(0, data); // 在下标为零的位置存储值
    			producer.publishData(byteBuffer); // 
    		}
    	 	
    	 	disruptor.shutdown(); // 关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；
    		executor.shutdown(); // 关闭 disruptor 使用的线程池；如果需要的话，必须手动关闭， disruptor 在 shutdown 时不会自动关闭；
    	}
     	
    }

# 3 组件说明 #

从生产者-消费者的整体：

RingBuffer：环形队列，是Disruptor最为重要的组件，其作用是存储和更新Disruptor中流通的数据。

Sequence：递增序号（AtomicLong），Disruptor使用Sequence标识一个特殊组件处理的序号。每个重要的组件基本都有一个Sequence。

Producer：生产者，泛指通过Disruptor发布事件的用户代码（实际业务代码，而并发框架代码）生成Event数据。

Event：事件，从生产者到消费者过程中的数据单元。由用户定义代码。

EventHandler：消费者，代表Disruptor框架中的一个消费者接口，由用户实现代码，负责处理Event数据，进度通过Sequence控制。

（打个比方：餐饮店买奶茶

你去餐饮店买奶茶，先要去柜台找服务员点一杯红豆抹茶，服务员会给你一个55号的排队号，等到服务员大喊：‘55号，55号’，于是你就屁颠屁颠的去拿红豆抹茶；

“你去买红豆抹茶” 就是 Producer

“红豆抹茶” 就是 Event

“柜台” 就是 RingBuffer

“55号” 就是 Sequence

“你去拿红豆抹茶” 就是 EventHandler）

从Disruptor框架如何处理Event的细节：

Sequecer：Disruptor框架真正的核心，在生产者和消费者直接进行高效准确快速的数据传输。通过复杂的算法去协调生存者和消费者之间的关系。

SequenceBarrier：Sequecer具体的实施者，字面理解是序号屏障，其目的是决定消费者 消费Evnet的逻辑。（生产者发布事件快于消费，生产者等待。消费速度大于生产者发布事件速度，消费者监听）

EventProcessor：可以理解为具体的消费线程，最后把结果返回给EventHandler。

WaitStrategy：当消费者等待在SequenceBarrier上时，有许多可选的等待策略

*  BusySpinWaitStrategy ： 自旋等待，类似Linux Kernel使用的自旋锁。低延迟但同时对CPU资源的占用也多。
 *  BlockingWaitStrategy ： 使用锁和条件变量。CPU资源的占用少，延迟大。
 *  SleepingWaitStrategy ： 在多次循环尝试不成功后，选择让出CPU，等待下次调度，多次调度后仍不成功，尝试前睡眠一个纳秒级别的时间再尝试。这种策略平衡了延迟和CPU资源占用，但延迟不均匀。
 *  YieldingWaitStrategy ： 在多次循环尝试不成功后，选择让出CPU，等待下次调。平衡了延迟和CPU资源占用，但延迟也比较均匀。
 *  PhasedBackoffWaitStrategy ： 上面多种策略的综合，CPU资源的占用少，延迟大。

（打个比方：

柜台的服务员通知某位厨师：“55号要一杯红豆抹茶”，然后厨师准备拿机器做奶茶，发现机器都在使用中，于是厨师就盯着机器看，当有空闲的机器就立马占用，做好后就端给客户。如果等了很久都没有空闲的机器，厨师会跟客服员说：“55号的红豆抹茶，可能要多等一会”，然后工作人员就和客户协调一下说明情况。

“服务员” 就是 Sequecer

“某位厨师” 就是 SequenceBarrier

“用机器做红豆抹茶” 就是 EventProcessor

“发现没有空闲机器，厨师监听” 就是 WaitStrategy

）

打的比方可能不是很形象。如果不理解的，可以反复的敲打代码，多问问为什么这样写，这样做有什么好处。慢慢的就理解了。

# 4 优质博客 #

[Disruptor入门][Disruptor]

以上边上Disruptor基础知识和相关的代码笔记，方便自己查阅，同时也希望能帮助到读者。下一章Disruptor场景应用。更多干货尽在 [ITDragon博客][ITDragon]

[aHR0cHM6Ly9pbWcyMDIwLmNuYmxvZ3MuY29tL2Jsb2cvODA2OTU2LzIwMjAwNy84MDY5NTYtMjAyMDA3MzEyMzQ1MDc3MjItMTExNjgwOTkyMi5naWY]: /images/20220610/02c3a794bf264ad88e2559bec6b40c05.png
[Disruptor]: http://ifeve.com/disruptor-getting-started/
[ITDragon]: http://www.itit123.cn/