当前位置： 首页 > news >正文

disruptor原理详解

news 2025/6/7 3:17:32

一、Disruptor
1、Disruptor基本原理

在现在的CPU中，每次从主内存读取数据到L3级缓存的大小是64k，也叫做cache line。
CPU找数据的过程：CPU->L1->L2->L3->主内存；CPU更新数据的过程也是如此。
CPU从内存加载数据的过程：L3->L2->L1->CPU
disruptor开源并发框架（RingBuffer:缓存一致性协议)
<dependency>
    <groupId>com.lmax</groupId>
    <artifactId>disruptor</artifactId>
    <version>3.4.2</version>
</dependency>

在多线程开发中，我们常常遇到这样一种场景：一些线程接受用户请求，另外一些线程处理这些请求。比如日志处理中的日志输入和告警。这种典型的生产者消费者场景十分常见，而生产者消费者模式的核心就是阻塞队列。由于阻塞队列会涉及大量的锁竞争和线程阻塞，都是非常耗费CPU的操作，因此阻塞队列的性能好坏能够在很大程度上决定上层应用的性能瓶颈。

JAVA中用BlockingQueue这个接口来描述阻塞队列，有数组实现的有界阻塞队列为 ArrayBlockingQueue，用链表实现的无界阻塞队列为LinkedBlockingQueue，除此之外还有优先级阻塞队列 PriorityBlockingQueue，这些阻塞队列除了自身特有逻辑外，都采用基于悲观锁的并发控制。这样的并发机制会有严重的锁冲突，大大影响并发性能。Disruptor满足了我们的要求。

锁是用来做并发最简单的方式，当然其代价也是最高的。内核态的锁的时候需要操作系统进行一次上下文切换，等待锁的线程会被挂起直至锁释放。

使用了Ringbuffer，内存屏障，乐观并发控制等众多优化手段后，Disrupter的阻塞队列与传统的阻塞队列相比有超过10倍的吞吐率。

Disruptor的主要设计思想是无锁的高并发，在设计上采用内存屏障的机制和CAS操作实现此思想。主流的并发程序 都离不开锁对资源的管控，或者尽量避开锁的使用。
其主要的实现原理总结有如下三点：

  • 采用消费者-生产者模型进行读写的分离。
  • 用循环缓存（实际是一个环形循环队列）实现了数据的暂存和读写速度的匹配。
  • 用内存屏障加序列号的方式实现了无锁的并发机制(Load-Store)。

为什么Disruptor的速度这么快？

我们知道Disruptor速度很快，但是为什么呢？
Disruptor没有使用很影响性能锁 。取而代之的是，在需要确保操作是线程安全的（特别是，在多生产者的环境下，更新下一个可用的序列号）地方，我们使用CAS（Compare And Swap/Set）操作。这是一个CPU级别的指令，它的工作方式有点像乐观锁——CPU去更新一个值，但如果想改的值不是原来的值，操作就失败，反之则去更新这个值。

说句题外话，Java中AtomicInteger也使用了CAS操作来保证原子性。在并发控制中，CAS操作是十分重要的。

CAS操作是CPU级别的指令，在Java中CAS操作在Unsafe类中（Unsafe，见名知意，这个类是不安全的，不建议在实际开发的时候使用）。关于CAS操作的原理网上有很多，在此不过多说明了。

另一个重要的因素是Disruptor消除了伪共享。 下面引用网上的一段话，来解释下什么是伪共享。

缓存系统中是以缓存行（cache line）为单位存储的。缓存行是 2 的整数幂个连续字节，一般为 32-256 个字节。最常见的缓存行大小是 64个字节。

当多线程修改互相独立的变量时，如果这些变量共享同一个缓存行，就会无意中影响彼此的性能，这就是伪共享，即缓存一致性协议。缓存行上的写竞争是运行在 SMP 系统中并行线程实现可伸缩性最重要的限制因素。有人将伪共享描述成无声的性能杀手，因为从代码中很难看清楚是否会出现伪共享。

第三个原因是Disruptor采用了RingBuffer。Ring Buffer 是一个数组，它比链表要快，而且有一个容易预测的访问模式，数据可以在硬件层面预加载到高速缓存，极大的提高了数据访问的速度。
RingBuffer可以预先分配内存，并且保持数组元素永远有效。这意味着内存垃圾收集（GC）在这种情况下几乎什么也不用做。

2、Disuptor基本概念

  • RingBuffer：Ringbuffer是一个环形缓冲区，是Disruptor的核心。
  • Sequencer：序号管理器，使消费者和生产者之前快速正确地传输数据。
  • Sequence：RingBuffer中每个数据的序号，用于跟踪ringbuffer中任务的变化和消费者的消费情况。
  • SequenceBarrier：序号栅栏，管理和协调生产者的游标序号和各个消费者的序号，确保生产者不会覆盖消费者未来得及处理的消息，确保存在依赖的消费者之间能够按照正确的顺序处理。
  • Event：生产者和消费者之间进行交换的数据称为事件。
  • EventProcessor：事件处理器，监听RingBuffer的事件，并消费可用事件，从RingBuffer读取的事件会交由实际的生产者实现类来消费；它会一直侦听下一个可用的序号，直到该序号对应的事件已经准备好。
  • EventHandler：业务处理器，是实际消费者的接口，完成具体的业务逻辑实现，第三方实现该接口；代表着消费者。
  • Producer：生产者接口，第三方线程充当该角色，producer向RingBuffer写入事件

3、环形数组RingBuffer

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述 

二、Disruptor 与ArrayBlockingQueue的性能比对

1、测试结果
经测试发现：

1、容量大小 与 消费者的消费速度 与整个耗时 成正比。
2、Disruptor的性能在高并发、高数据规模（bufferSize 要大些）时表现更突出。
3、Disruptor与LinkedBlockingQueue（比ArrayBlockingQueue性能好些）比对而言，当bufferSize大些的时候，也有优势。


与等待策略说明进行比较：

 由上面的数据可以看出：

BlockingWaitStrategy和TimeoutBlockingWaitStrategy的效率差不多，都是应用在CPU资源紧缺，吞吐量和延迟并不重要的场景，其他几个等待策略略微处理事件略微有点大，是将近前两种的两倍。Disruptor在不配置策略的时候，默认使用BlockingWaitStrategy；在不配置单个生产者和多个生产者的时候，默认使用多个生产者。即使用ProducerType.MULTI。

2、在pom文件中引入disruptor

<dependency>
<groupId>com.lmax</groupId>
<artifactId>disruptor</artifactId>
<version>3.4.2</version>
</dependency>

3、测试入口

import java.time.Instant;
import java.time.format.DateTimeFormatter;/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：比对测试*/
public class CompareTest {public static int THREAD = 256;//2 << 5; // 线程数量 5 8  64 128 256public static int PER = 65536 ;//2 << 10; // 单个线程生产数量 65536public static int TOTAL_COUNT = THREAD * PER; // 数据总数量public static int SIZE =1048576; // 最大容量 32 128 512 1024 2048 1048576public static void main(String[] args) {println("线程数：" + THREAD + " 单线程生产量: " + PER + " 容量：" + SIZE + " 数据总量：" + TOTAL_COUNT);//分开执行//new Thread(() -> ArrayBlockingQueueTest.execute()).start();new Thread(() -> DisruptorTest.execute()).start();}public static void println(String msg) {System.out.println(DateTimeFormatter.ISO_INSTANT.format(Instant.now()) + "  " + msg);}
}


4、ArrayBlockingQueue 测试用例

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：ArrayBlockingQueue Test*/
public class ArrayBlockingQueueTest {public static void execute() {ArrayBlockingQueue<String> queue = new ArrayBlockingQueue<String>(CompareTest.SIZE);AtomicBoolean endP = new AtomicBoolean(false);AtomicBoolean endC = new AtomicBoolean(false);long startTime = System.currentTimeMillis();AtomicLong count = new AtomicLong(0);for (int i = 0; i < CompareTest.THREAD; i++) {final int m = i;new Thread(() -> {for (int j = 0; j < CompareTest.PER; j++) {try {queue.put("i" + m + "j" + j); // 队列不够，等待生产} catch (InterruptedException e) {e.printStackTrace();}if (count.incrementAndGet() == CompareTest.TOTAL_COUNT) {CompareTest.println("ArrayBlockingQueue 生产耗时：" + (System.currentTimeMillis() - startTime));endP.set(true);}}}).start();}new Thread(() -> {AtomicLong consumerCount = new AtomicLong(0);while (true) {try {queue.take(); // 直到消费完所有信息} catch (InterruptedException e) {e.printStackTrace();}if (consumerCount.incrementAndGet() == CompareTest.TOTAL_COUNT) {break;}}CompareTest.println("处理count：" + consumerCount.get() + "  ArrayBlockingQueue 消费耗时："+ (System.currentTimeMillis() - startTime));endC.set(true);}).start();while (!(endC.get() && endP.get())) {}CompareTest.println("ArrayBlockingQueue 总耗时：" + (System.currentTimeMillis() - startTime));}
}


5、Disruptor 测试用例

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：disruptor测试线程*/
public class DisruptorTest {public static void execute() {Disruptor<DataEvent> disruptor = new Disruptor<DataEvent>(new DataEventFactory(),     //创建事件（数据）工厂CompareTest.SIZE,           //环形数组大小，即创建ringBuffer大小，一定是2的N次new ThreadFactory() {       //创建一个线程工厂 提供线程来触发Consumer的事件处理@Overridepublic Thread newThread(Runnable eventProcessor) {// 对事件处理总线的封装CompareTest.println("EventProcessor wrapper");Thread thread = new Thread(eventProcessor);thread.setName("EventProcessorWrapper");return thread;}},/*** 1、ProducerType.SINGLE 表示单个生产者* 2、ProducerType.MULTI 表示多个生产者*/ProducerType.MULTI,/***  消费者的等待策略*          名称                    措施                     适用场景* BlockingWaitStrategy            加锁                     CPU资源紧缺，吞吐量和延迟 并不重要的场景* BusySpinWaitStrategy            自旋                     通过不断重试，减少切换线程导致的系统调用，而降低延迟。推荐在线程绑定到固定的CPU的场景下使用* PhasedBackoffWaitStrategy    自旋 yield 自定义策略     CPU资源紧缺，吞吐量和延迟 并不重要的场景* SleepingWaitStrategy            自旋 yield sleep         性能和CPU资源之间有很好的折中。延迟不均匀* TimeoutBlockingWaitStrategy 加锁，有超时限制           CPU资源紧缺，吞吐量和延迟 并不重要的场景* YieldingWaitStrategy            自旋 yield 自旋            性能和CPU资源之间有很好的折中。延迟比较均匀* 这里需要降低延迟，所以选择 BusySpinWaitStrategy*/new BlockingWaitStrategy());//new TimeoutBlockingWaitStrategy(2000L, TimeUnit.SECONDS));//new PhasedBackoffWaitStrategy(1000L , 1000L , TimeUnit.SECONDS , new SleepingWaitStrategy()));CompareTest.println("测试策略：BusySpinWaitStrategy");/*** 创建EventProcessors<Runnable>.* 子过程Disruptor.checkNotStarted()事件处理handler必须在启动之前绑定.*/disruptor.handleEventsWith(new DataEventHandler());disruptor.start();CompareTest.println("disruptor start success！");RingBuffer<DataEvent> ringBuffer = disruptor.getRingBuffer();DataProducer producer = new DataProducer(ringBuffer);//DataEventProducerWithTranslator translator = new DataEventProducerWithTranslator(ringBuffer);long start = System.currentTimeMillis();for (int l = 0; l < CompareTest.THREAD; l++) {new Thread(() -> {for (int m = 0; m < CompareTest.PER; m++) {producer.onData(start);// translator.onData(start);}}).start();}/*** 关闭 disruptor，方法会堵塞，直至所有的事件都得到处理；并不会自动关闭外部指定的executor,需要主动关闭*///disruptor.shutdown();//CompareTest.println("disruptor shutdown success！");//executor.shutdown();}
}


事件

/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：事件实例封装 业务数据传递对象*/
public class DataEvent {private long startTime;public long getStartTime() {return startTime;}public void setStartTime(long startTime) {this.startTime = startTime;}}

生产者

import com.lmax.disruptor.RingBuffer;
/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：当前还是生产线程*          onData用来发布事件，每调用一次就发布一次事件事件 它的参数会通过事件传递给消费者*/
public class DataProducer {private final RingBuffer<DataEvent> ringBuffer;public DataProducer(RingBuffer<DataEvent> ringBuffer) {this.ringBuffer = ringBuffer;}public void onData(long data) {//获取下一个事件槽并发布事件要使用try/finally保证事件一定会被发布, 所以最好直接使用 ringBuffer.publishEvent方式将数据交由Translator来处理填充DataEvent，最后finally发布// 可以把ringBuffer看做一个事件队列，那么next就是得到下面一个事件槽, 若没有空闲的时间槽则阻塞long sequence = ringBuffer.next();// CompareTest.println("生产置入sequence:" + sequence);try {// 用上面的索引取出一个空的事件用于填充DataEvent event = ringBuffer.get(sequence);event.setStartTime(data);} finally {// 发布事件ringBuffer.publish(sequence);}}
}

import com.lmax.disruptor.EventTranslatorOneArg;
import com.lmax.disruptor.RingBuffer;/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：获取下一个事件槽并发布事件（发布事件的时候要使用try/finally保证事件一定会被发布）。* 如果我们使用RingBuffer.next()获取一个事件槽，那么一定要发布对应的事件。* 如果不能发布事件，那么就会引起Disruptor状态的混乱。* 尤其是在多个事件生产者的情况下会导致事件消费者失速，从而不得不重启应用才能会恢复。*/
public class DataEventProducerWithTranslator {private final RingBuffer<DataEvent> ringBuffer;/*** 一个translator可以看做一个事件初始化器，publicEvent方法会调用它* 填充Event*/private static final EventTranslatorOneArg<DataEvent, Long> TRANSLATOR = new EventTranslatorOneArg<DataEvent, Long>() {public void translateTo(DataEvent event, long sequence, Long startTime) {event.setStartTime(startTime);}};public DataEventProducerWithTranslator(RingBuffer<DataEvent> ringBuffer) {this.ringBuffer = ringBuffer;}/*public void onData(Long bb) {ringBuffer.publishEvent(TRANSLATOR, bb);// 当前还是生产者线程// CompareTest.println(Thread.currentThread().getName() + " pulishEvent end!");}*/
}

消费者

import com.lmax.disruptor.EventHandler;
import java.util.concurrent.atomic.AtomicLong;/*** 创建时间：2021年5月19日 13:38:07* 创建作者：admin* 当前版本：v1.0* 基本功能：对指定事件的处理过程*/
public class DataEventHandler implements EventHandler<DataEvent> {public AtomicLong count = new AtomicLong(0);/*** 消费者消费的入口 可以进行业务处理* @param event* @param sequence* @param endOfBatch* @throws Exception* @author admin* @date 2021年5月19日 13:38:07*/@Overridepublic void onEvent(DataEvent event, long sequence, boolean endOfBatch) throws Exception {/*** 消费者线程由初始化Disruptor时指定的threadFactory创建的*/if (count.incrementAndGet() == CompareTest.TOTAL_COUNT) {CompareTest.println("处理的sequence：" + sequence + " count：" + count.get() + "  Disruptor 总耗时："+ (System.currentTimeMillis() - event.getStartTime()));}}}



 

http://www.xdnf.cn/news/823771.html

相关文章：

  • 网安学途—SQL SERVER 2008安装教程
  • Apache Log4j2 详解 （一）
  • C语言——动态内存函数（malloc、calloc、realloc、free）
  • 【揭秘】ScheduledExecutorService全面解析
  • postfix（邮件服务器）说明与postconfig命令详解
  • 2025年5月TIOBE 指数头条：Python 统治世界！多家权威机构____编程语言排行榜__薪酬状况
  • User-Agent（用户代理）是什么？
  • CentOS7保姆级安装教程
  • Linux--Shell基础
  • 图床的千层套路
  • 【全面讲解 C 语言的结构体(struct)，一网打尽】【转载】
  • Grok-1 ：目前参数最大的开源大模型
  • Android反编译apk并重新编译
  • 句柄详解，什么是句柄？句柄有什么用？
  • jQuery-Ajax（详解）
  • LINQ详解(查询表达式)
  • fgets()函数的详解-使用技巧-C语言基础
  • 几种常用的排序方法——选择排序
  • Spring 中的 AOP 详解
  • 如何破解压缩包密码，CTF压缩包处理
  • Unity中Shader旋转矩阵（四维旋转矩阵）
  • Matlab下载安装详细教程
  • sin函数对照表_三角函数表值对照表格
  • 一文详解| 通俗易懂讲解MOS管
  • 计算机组成原理（4）-----Cache的原理及相关知识点
  • 一句话木马该怎么实现？现在就带你了解
  • 什么是Serv-U,什么是servu,Serv-U,servu
  • SQL DISTINCT关键字详解
  • 【嵌入式开发】529
  • 五分钟带你了解前端开发，以及前端开发的发展机会