秒级高并发框架-Disruptor
作者：技术趋势 查看原文
Disruptor介绍
Disruptor是一个高性能的并发框架，主要应用于创建具有高吞吐量、低延迟、无锁（lock-free）的数据结构和事件处理系统。它最初由LMAX公司开发的，已经成为了业界广泛使用的高性能并发框架。
Disruptor框架的特点和优势包括：
- 高性能：Disruptor框架能够通过无锁的方式提供非常高的并发性能和吞吐量，比如在大规模消息发布订阅场景下，能够每秒处理数百万个消息。
- 低延迟：与传统的基于共享内存的方式相比，Disruptor框架通过线程之间的缓存操作和快速消息传递实现低延迟。
- 易用性：Disruptor框架提供了简单的API，可以方便地实现生产者-消费者模式、消息队列、事件处理器等多种应用场景。
- 可扩展性：Disruptor框架支持多线程处理消息，可以根据实际需求设置线程数，以提高处理效率。
Disruptor解决了什么问题？
- Disruptor主要解决的是高性能应用中的并发问题，主要涉及数据缓存和线程通信这两个方面。在传统的并发编程中，由于共享状态和锁竞争等问题，很容易导致线程间的同步延迟，从而影响应用程序的性能和可扩展性。
- Disruptor采用了无锁（Lock-Free）的并发编程技术，将数据存储在一个环形缓冲区中，并通过CAS操作等方式实现数据的并发读写和线程间的通信。它在保证数据一致性的同时，最大限度地压缩了线程间的同步开销，从而能够实现高效的消息传递和事件处理。Disruptor 的高吞吐量、低延迟以及可扩展性好的特点，使得它成为许多高并发应用的首选方案之一。
Disruptor的基本使用
以下是用于disruptor这个框架的基本功能的学习demo。
引入jar包
<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.2.1</version>
</dependency>
生产者、消费者、事件代码
/**
*
* 功能描述: 定义事件event 通过Disruptor 进行交换的数据类型。
*
* @param:
* @return:
* @auther: csh
* @date: 2023/6/17 11:33 下午
*/
public class LogEvent {
private String message;
public String getMessage() {
return message;
}
public void setMessage(String message) {
this.message = message;
}
}
/**
*
* 功能描述: 生产者
*
* @param:
* @return:
* @auther: csh
* @date: 2023/6/18 12:01 上午
*/
public class Producer implements EventTranslator<LongEvent> {
@Override
public void translateTo(LongEvent event, long sequence) {
event.setNumber(sequence);
}
}
package com.hong.arithmetic.disruptor;
import com.lmax.disruptor.EventHandler;
/**
*
* 功能描述: 消费者
*
* @param:
* @return:
* @auther: csh
* @date: 2023/6/18 12:01 上午
*/
public class Consumer implements EventHandler<LongEvent> {
@Override
public void onEvent(LongEvent event, long sequence, boolean endOfBatch) throws Exception {
System.out.println("Consumer:" + event.getNumber());
}
}
package com.hong.arithmetic.disruptor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import java.util.Date;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
/**
* @author: csh
* @Date: 2023/6/17 23:37
* @Description:主测试方式
*/
public class DisruptorMain {
public static void main(String[] args) {
//开始时间
Date start = new Date();
//创建线程池
ExecutorService executor = Executors.newCachedThreadPool();
//初始化Disruptor 其中参数顺序如下：
// LongEvent新建
// ringBufferSize大小一定要是2的N次方
// executor为线程池
Disruptor<LongEvent> disruptor = new Disruptor<LongEvent>(LongEvent::new, 1024 * 1024, executor);
//连接消费者
disruptor.handleEventsWith(new Consumer());
//启动
RingBuffer<LongEvent> ringBuffer = disruptor.start();
//生产者
Producer producer = new Producer();
IntStream.range(0, 1000000)
.parallel()
.forEach(i -> {
ringBuffer.publishEvent(producer);
});
//关闭服务 关闭线程池
disruptor.shutdown();
executor.shutdown();
Date end = new Date();
System.out.println((end.getTime() - start.getTime()) / 1000 + "秒");
}
}
结果
...
Consumer:999847
Consumer:999848
Consumer:999849
Consumer:999850
Consumer:999851
Consumer:999852
Consumer:999853
Consumer:999854
Consumer:999855
Consumer:999856
Consumer:999857
Consumer:999858
Consumer:999859
Consumer:999860
Consumer:999861
Consumer:999862
Consumer:999863
Consumer:999864
Consumer:999865
Consumer:999866
Consumer:999867
Consumer:999868
Consumer:999869
Consumer:999870
Consumer:999871
Consumer:999872
Consumer:999873
Consumer:999874
Consumer:999875
Consumer:999876
Consumer:999877
Consumer:999878
Consumer:999879
Consumer:999880
Consumer:999881
Consumer:999882
Consumer:999883
Consumer:999884
Consumer:999885
Consumer:999886
Consumer:999887
Consumer:999888
Consumer:999889
Consumer:999890
Consumer:999891
Consumer:999892
Consumer:999893
Consumer:999894
Consumer:999895
Consumer:999896
Consumer:999897
Consumer:999898
Consumer:999899
Consumer:999900
Consumer:999901
Consumer:999902
Consumer:999903
Consumer:999904
Consumer:999905
Consumer:999906
Consumer:999907
Consumer:999908
Consumer:999909
Consumer:999910
Consumer:999911
Consumer:999912
Consumer:999913
Consumer:999914
Consumer:999915
Consumer:999916
Consumer:999917
Consumer:999918
Consumer:999919
Consumer:999920
Consumer:999921
Consumer:999922
Consumer:999923
Consumer:999924
Consumer:999925
Consumer:999926
Consumer:999927
Consumer:999928
3秒
测试了一下100万条是3秒，而300万是9秒，当然中间还有入队的时间，还没有业务逻辑，性能还过得去~。所以这个框架总体性能单体百万是不考虑具体业务逻辑的，当然我的电脑是M1，上面还跑了大量的软件，仅达到了33万/s,可能因为配置所限~，有兴趣同学可以自行测试~
Disruptor的核心设计原理
Disruptor是一款高性能内存队列框架，它采用了一些独特的设计原理，通过利用Java虚拟机的内存模型和CPU缓存等硬件特性来实现高效的数据传输和处理。下面是Disruptor的核心设计原理：
- 环形缓冲区
Disruptor中最基本的数据结构是一个环形缓冲区，所有的生产者和消费者都通过这个缓冲区进行数据交换。缓冲区的大小是预先指定的，可以根据实际业务需求进行调整。在Disruptor中，每个元素都是一个事件对象（Event），用于封装需要传递和处理的数据。
- 内存屏障
Disruptor利用了Java虚拟机的内存模型和CPU缓存等硬件特性来实现高效的数据传输和处理。为了保证数据在不同线程之间的可见性和有序性，Disruptor采用了内存屏障（Memory Barrier）的技术，在不同的操作之间加入特殊的指令以实现对内存的障碍保护和重排序优化。
- RingBuffer填充
为了避免生产者和消费者之间的竞争，Disruptor采用了预填充（Pre-Fill）的技术，在RingBuffer中提前创建一定数量的Event对象，并通过游标指针标记已经填充的位置。这样，在生产者向RingBuffer发布事件时，就可以直接将数据写入预分配好的Event中，避免了动态创建Event和竞争锁的过程。
- CAS原子操作(无锁)
Disruptor利用了CAS（Compare And Swap）原子操作来实现对RingBuffer的读写操作，以避免传统的锁竞争问题。在Disruptor中，每个消费者都维护一个自己的Sequence（序号），用于表示其处理的事件在RingBuffer中的位置，消费者通过CAS操作不断地更新自己的Sequence来实现对事件的读取和处理。
- 分离发布和提交应用
Disruptor引入了“两阶段提交”（Two-Phase Commit）的模式，在生产者端将事件发布到RingBuffer前，先将事件封装为BatchEventProcessor，并通过Barrier来进行事件的校验和等待。然后再将批量的事件提交到RingBuffer中，由消费者进行消费。这种设计模式能够有效提高事件的处理效率，减少线程之间的竞争。
Disruptor的数据结构
Disruptor是一个高性能、低延迟的事件处理框架，内部使用多种数据结构协同工作，保证了其在多线程并发环境下高效的数据传输和处理。主要的数据结构包括：
- RingBuffer：Disruptor的核心数据结构，被用作队列的数据结构，通过Claim Strategy将事件发布到RingBuffer中，然后由Wait Strategy等待消费者获取并处理事件。RingBuffer采用预分配的方式，即在初始化时预先为每个slot分配了内存空间，避免了动态分配内存带来的开销和竞争。
- Sequence：Sequence是序管理器是Disruptor中的一个序列号(自增)，代表了生产者或消费者已经处理到哪个位置。每个生产者和每个消费者都维护自己的Sequence值，RingBuffer就是通过Sequence值来标识每个槽位的。Disruptor使用Sequence来实现流水线的形式，不同的处理阶段之间会通过Sequence进行衔接。
- SequenceBarrier：SequenceBarrier简称序栅栏是用来保证消费者和生产者之间的协作。当消费者读取RingBuffer中的事件时，需要等待生产者提供的事件可用。而生产者发布事件时，也需要等待消费者已经处理完之前的事件。SequenceBarrier就可以提供这种等待机制，它可以阻塞消费者线程直到生产者发布了足够的事件，也可以阻塞生产者线程直到消费者处理完之前的事件。
- BatchEventProcessor：BatchEventProcessor称为事件批理处理器是Disruptor的核心处理器，用来处理RingBuffer中的事件。它使用一个Sequence来表示消费者已经处理完哪些事件，当新的事件可用时，它会批量处理一定数量的事件并更新自己维护的Sequence值。
- EventHandler：EventHandler是Disruptor中的事件处理器，用于处理从RingBuffer中读取的事件。每个EventHandler都需要实现onEvent方法，在其中编写业务逻辑。
Disruptor的等待策略主要有以下几种具体实现类：
- BusySpinWaitStrategy：忙等待策略，使用循环检查的方式等待新事件，能够实现最低的延迟，但会消耗大量的CPU资源。
- SleepingWaitStrategy：休眠等待策略，当没有新事件到来时，消费者线程会进入睡眠状态，在指定的时间后醒来继续检查RingBuffer，相对于Busy Spin可以更加节约CPU资源。
- YieldingWaitStrategy：让步等待策略，当没有新事件到来时，消费者线程会暂停执行，将CPU资源让给其他线程，适用于中等延迟场景。
- BlockingWaitStrategy：阻塞等待策略，它会使消费者线程进入阻塞状态，直到新的事件可用或者超时。适用于更高延迟的场景，但是会对系统吞吐量产生影响。
- LiteBlockingWaitStrategy：是一种非重入锁的阻塞等待策略，它在实现上相对于BlockingWaitStrategy更加轻量级，同时也能够实现阻塞等待。
- TimeoutBlockingWaitStrategy：超时等待策略，它会使消费者线程进入阻塞状态，在指定的时间内等待新的事件，如果等待超时则退出。相较于BlockingWaitStrategy，它具有更好的响应性能和可控的阻塞时间。
这些具体实现类可以根据不同的需求和场景进行选择，以达到更加高效和稳定的数据传输和处理效果。
Disruptor的源码学习
Disruptor类关系图
代码位置：com.lmax.disruptor.dsl.Disruptor
public class Disruptor<T>
{
private final RingBuffer<T> ringBuffer; // RingBuffer对象
private final Executor executor; // 线程池对象
private final ConsumerRepository<T> consumerRepository = new ConsumerRepository<T>(); // 消费者仓库对象
private final AtomicBoolean started = new AtomicBoolean(false);//多线程环境中并发访问时的安全性
private ExceptionHandler exceptionHandler;//异常处理策略
/**
* 创建Disruptor对象
*
* @param eventFactory 事件工厂
* @param bufferSize RingBuffer缓冲区大小
* @param executor 线程池
*/
public Disruptor(EventFactory<T> eventFactory, int bufferSize, Executor executor) {
this.executor = executor;
ringBuffer = new RingBuffer<T>(eventFactory, bufferSize);
// 创建SequenceBarrier对象
SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
// 创建WorkerPool对象，并启动消费者线程
WorkerPool<T> workerPool = new WorkerPool<T>(ringBuffer, sequenceBarrier, new FatalExceptionHandler(), consumerRepository.getAll());
ringBuffer.setGatingSequences(workerPool.getWorkerSequences());
workerPool.start(executor);
}
/**
* 注册事件处理器
*
* @param handlers 事件处理器
* @return EventHandlerGroup对象
*/
public EventHandlerGroup<T> handleEventsWith(EventHandler<? super T>... handlers) {
return consumerRepository.addHandlers(ringBuffer, handlers);
}
/**
* 注册关联事件处理器
*
* @param barrierSequences SequenceBarrier序列数组
* @param handlers 事件处理器
* @return EventHandlerGroup对象
*/
public EventHandlerGroup<T> handleEventsWithWorkerPool(SequenceBarrier... barrierSequences, WorkHandler<? super T>... handlers) {
return consumerRepository.addWorkerPool(ringBuffer, barrierSequences, handlers);
}
/**
* 注册事件处理器，每n个事件触发一次处理
*
* @param n 处理的事件数
* @param handlers 事件处理器
* @return EventHandlerGroup对象
*/
public EventHandlerGroup<T> handleEventsWithBatchSize(int n, EventHandler<? super T>... handlers) {
return consumerRepository.addBatchEventHandlers(ringBuffer, n, handlers);
}
/**
* 注册事件处理器，每n个事件触发一次处理
*
* @param n 处理的事件数
* @param handlers 事件处理器
* @return EventHandlerGroup对象
*/
public EventHandlerGroup<T> handleEventsWithBatchSize(int n, BatchEventHandler<? super T>... handlers) {
return consumerRepository.addBatchEventHandlers(ringBuffer, n, handlers);
}
/**
* 注册异常处理器
*
* @param exceptionHandler 异常处理器
*/
public void handleExceptionsWith(ExceptionHandler<? super T> exceptionHandler) {
consumerRepository.setExceptionHandler(exceptionHandler);
}
/**
* 获取RingBuffer对象
*
* @return RingBuffer对象
*/
public RingBuffer<T> getRingBuffer() {
return ringBuffer;
}
/**
* 关闭Disruptor对象
*/
public void shutdown() {
consumerRepository.removeAll();
ringBuffer.setGatingSequences(new Sequence[0]);
}
}
代码位置：com.lmax.disruptor.RingBuffer
//环形缓冲的核心实现类
public final class RingBuffer<T>
{
//RingBuffer中entry数组的大小掩码，被定义为 bufferSize - 1L，用于执行快速的modulo运算。
private final long indexMask;
// Object数组，用于存储Event对象。
private final Object[] entries;
//环形缓冲区的大小，即entry数组的长度。
private final int bufferSize;
//实现了RingBuffer序列化的Sequencer对象。
private final Sequencer sequencer;
/**
* 构建一个RingBuffer实例
*
* @param factory 工厂对象，用于创建RingBuffer对象
* @param sequencer Sequencer对象，用于对RingBuffer进行序列化
*/
public RingBuffer(EventFactory<T> factory, Sequencer sequencer) {
this.sequencer = sequencer;
this.bufferSize = sequencer.getBufferSize();
this.indexMask = bufferSize - 1L;
this.entries = new Object[sequencer.getBufferSize()];
// 循环调用EventFactory创建初始的事件对象
for (int i = 0; i < bufferSize; i++)
{
entries[i] = factory.newInstance();
}
}
/**
* 获取RingBuffer的大小
*
* @return bufferSize
*/
public int getBufferSize() {
return bufferSize;
}
/**
* 将指定的sequence位置上的entry设置为null
*
* @param sequence 序列号
*/
public void resetTo(long sequence) {
this.entries[(int) (sequence & indexMask)] = null;
}
/**
* 获取下一个可用的序列号
*
* @return 下一个可用的序列号
*/
public long next() {
return sequencer.next();
}
/**
* 获取指定的可用的sequence编号
*
* @param n 获取的sequence的数量
* @return 获取的sequence编号集合
*/
public long next(int n) {
return sequencer.next(n);
}
/**
* 发布指定序列号的事件，通知消费者
*
* @param sequence 索引值
*/
public void publish(long sequence) {
sequencer.publish(sequence);
}
/**
* 发布指定区间的事件，通知消费者
*
* @param lo 序列号起始值
* @param hi 序列号结束值
*/
public void publish(long lo, long hi) {
sequencer.publish(lo, hi);
}
/**
* 根据序列号获取对应的entry
*
* @param sequence 序列号
* @return 对应的entry
*/
@SuppressWarnings("unchecked")
public T get(long sequence) {
return (T) entries[(int) (sequence & indexMask)];
}
...
}
RingBuffer类实现了Disruptor中的环形缓冲区，用于存储事件(Event)对象并将其提供给消费者进行处理。下面是该类的主要属性及方法：
Object[] entries: Object数组，用于存储Event对象。
long indexMask: RingBuffer中entry数组的大小掩码，被定义为bufferSize - 1L，用于执行快速的modulo运算。
int bufferSize: 环形缓冲区的大小，即entry数组的长度。
Sequencer sequencer: 实现了RingBuffer序列化的Sequencer对象。
RingBuffer(EventFactoryfactory, Sequencer sequencer): 构造函数，用于创建RingBuffer实例并初始化相应属性。
getBufferSize(): 返回RingBuffer的大小。
resetTo(long sequence): 将指定的序列号上的entry设置为null。
next(): 返回下一个可用的序列号。
next(int n): 获取指定数量的可用序列号。
publish(long sequence): 发布指定序列号的事件。
publish(long lo, long hi): 发布指定区间内的事件。
get(long sequence): 根据序列号获取对应的entry。
在Disruptor中，生产者会首先请求一个序列号（通过调用next()或next(n)方法），然后使用该序列号将事件放入RingBuffer中。消费者会根据序列号从RingBuffer中取出相应的事件，并进行处理。此过程由RingBuffer和Sequencer负责协同完成。
源码就简单看到这里，有兴趣同学可以自行了解~
最后
Disruptor框架其实很早之前就出来了~，大致是12年左右那段时间比较火，后续被很多公司借鉴了这种设计理念，所以国内部分开源的并发框架或自研的都是基于这个或参考这个框架实现的，只是我们很多时候不了解而以，较老一点的程序人应该都有所了解~，当然本文主要是用于了解Disruptor这个框架实现源理和基本应用以便后续一些高并发场景设计方案可以借鉴，希望能与各位共同前进~，最后可以参考下面文献还是挺不错的借鉴~
参考文献：
Disruptor详解：
https://www.jianshu.com/p/bad7b4b44e48
美团-高性能队列——Disruptor：
https://tech.meituan.com/2016/11/18/disruptor.html
高性能无锁并发框架 Disruptor，太强了！：