【优化技术专题】「线程间的高性能消息框架」再次细节领略 Disruptor 的底层原理和优势分析

Disruptor 原理

首先 Disruptor 是为了解决高并发缓存的队列，为线程间通讯提供高效的性能，它是如何做到无阻塞、多生产、多消费的？

上图简单的画了一下构建 Disruptor 的各个参数以及 ringBuffer 的构造，下面简单的说一下。

生产者需要组件

生产者，产生消息，并将消息发布到 RingBuffer 内存队列中。

• Event 模型：从生产者传递给消费者的数据单位，完全由用户定义其类型。

@Datapublic class SampleEvent {    private Long id。    private String sampleDataStr。}

• EventFactory：创建事件（任务）的工厂类。（这里任务会创建好，保存在内存中，可以看做是一个空任务）。

public class SampleEventFactory implements EventFactory<SampleEvent> {    @Override    public SampleEvent newInstance() {        // 实例化数据（建好空数据，等待后面初始化）        return new SampleEvent()。    }}

• RingBuffer：环形缓冲区通常被认为是 Disruptor 的主要实现，当前版本即 3.0 版本之后，RingBuffer 仅负责存储和更新通过 Disruptor 的数据（Event）。

• ringBufferSize：容器的长度。（ Disruptor 的核心容器是 ringBuffer，环转数组，有限长度）。

• ProductType：生产者类型：单生产者、多生产者。

• Sequencer：Sequencer 是 Disruptor 的核心 API。该接口的 2 个实现类（SingleProducer，MultiProducer）实现了所有并发算法，用于在生产者和消费者之间快速，正确地传递数据。

• WaitStrategy：等待策略。（当队列里的数据都被消费完之后，消费者和生产者之间的等待策略），等待策略确定消费者如何等待生产者将事件放入 Disruptor。

• RingBuffer：存放数据的容器。

@Data@AllArgsConstructorpublic class SampleEventProducer {    private RingBuffer<OrderEvent> ringBuffer。    public void sendData(long id) {        //获取下一个可用序号        long sequence = ringBuffer.next()。        try {            //获取一个空对象（没有填充值）            SampleEvent sampleEent = ringBuffer.get(sequence)。        }finally {            //提交            ringBuffer.publish(sequence)。        }    }}

消费者需要组件

• Executor：消费者线程池，执行任务的线程。（每一个消费者都需要从线程池里获得线程去消费任务）。

• EventProcessor：用于处理来自 Disruptor 的事件的主事件循环，并具有消费者序列的所有权。有一个名为 BatchEventProcessor 表示，它包含事件循环的有效实现，并将回调到使用的提供的 EventHandler 接口实现。

• EventHandler：事件处理器，由用户实现并代表 Disruptor 的使用者的接口,用户客户端实现消息的处理机制，由客户端具体实现。

public class SampleEventHandler implements EventHandler<SampleEvent> {     /**     * 事件驱动监听--消费者消费的主体     */    @Override    public void onEvent(SampleEvent  event, long sequence, boolean endOfBatch) throws Exception {        System.out.println(event.getSampleDataStr() + "  " +Thread.currentThread().getName())。    }}

算法核心 Sequence 序号

• Sequence：Disruptor 使用 Sequences 作为识别特定组件所在位置的方法。

• 每个消费者（EventProcessor）都像 Disruptor 本身一样维护一个 Sequence。大多数并发代码依赖于这些 Sequence 值的变化或者叫移动，因此 Sequence 支持 AtomicLong 的许多当前功能。

• 事实上，唯一真正的区别是 Sequence 包含额外的功能，以防止序列和其他值之间的错误共享。

• Sequence Barrier：序列屏障由 Sequencer 产生，包含对 Sequencer 中主要发布的 sequence 和任何依赖性消费者的序列的引用。它包含确定是否有任何可供消费者处理的事件的逻辑。

Disruptor 的优点：

1. 多线程之间没有竞争即没有锁。

2. 所有访问者都记录自己的序号的实现方式，允许多个生产者与多个消费者共享相同的数据结构。

3. 每个对象中都能跟踪序列号（ring buffer, claim strategy,生产者和消费者），加上神奇的缓存行填充，就意味着没有伪共享和非预期的竞争。

下面再简单介绍下 RingBuffer 核心实现，来看看队列的实现细节。

其为环形队列，有点像一致性 Hash 算法中的闭环，但完全不一样。

底层的话是一个固定大小的数组结构，相比于队列来说，其只有一个下标指针 cursor，如果槽的个数是 2 的 N 次方更有利于基于二进制的计算机进行计算。如果看过 HashMap 源码应该知道，HashMap 定位元素槽时使用了一种巧妙的方式，hash&(length-1)。

RingBuffer 同样是相同的计算方式，sequence&(length-1),当然你可以进行取模操作。

• 取模操作在寄存器中的计算，需要多次的迭代加操作进行的，所以相对于计算速度来说，对于计算机进行位运算效率绝对是高于取模操作的，尤其是对于高并发状况下的计算，能够节省很多单位 cpu 开销。

一般实现线性存储有两种实现方式：

• 一种是基于连续内存分配的 HashTable

• 一种是基于随机内存分配的迭代指针。

为什么 RingBuffer 选用数组作为存储结构，而不选用链表存储?

缓存或者程序的局部性原理

• （Good）数组内存属是连续分配内存的预读策略，也就是内存加载时，会将部分连续内存地址预先加载到高速缓存中，即认为你可能会使用，上面我们分析了操作系统中的 cpu 操作数据的流程，可以看出这种设计是为了不用反复从内存中加载。

• （Bad）链表的内存分配是碎片化的所以其存储地址不是连续的，导致每次都会 cpu 都会重新计算下一个链表位置的地址，并从内存中加载相关的数据，数据量小的情况下并不能看出性能的优劣，但是当数据量大的情况下，这种极小的消耗，会对整体的运行效率产生影响。

因为 RingBuffer 不会涉及到存储地址的修改和维护，就因为选用数组就对性能产生了有利和积极的影响。

伪共享

内存以高速缓存行的形式存储在高速缓存系统中。高速缓存行是 2 的 N 次方个连续字节，其大小通常为 32-256，最常见的缓存行大小为 64 字节。

伪共享是一个术语，适用于线程在修改共享同一缓存行的独立变量时无意中影响彼此的性能。在高速缓存行上写入争用是实现 SMP 系统中并行执行线程的可伸缩性的最大限制因素。(出自百度定义！)

• 首先我们知道对于锁来说是关中断实现，锁定 bus 消息总线实现，而对于共享内存，计算机使用的是缓存行，共享变量的多个线程，共享相同的缓存行。

• 实现线程数量的线性可伸缩性，我们必须确保没有两个线程写入同一个变量或缓存行。而当使用 volatile 的时候，我们读取直接共享变量从主内存或者叫共享内存中读取变量的值，其本质是使计算机缓存行失效。

在 CPU 核心 A 运行的线程想要更新变量 X，而 CPU 核心 B 上的线程想要更新变量 Y。

这两个热变量位于同一缓存行中。每个线程都将竞争缓存行的所有权，以便他们可以更新它。如果核心 A 获得所有权，那么 MESI/MOSI 缓存子系统将需要使核心 B 的相应缓存行无效。反之也是一样，极大地影响性能。如果竞争核心在不同的套接字上并且还必须跨越套接字互连，则缓存行问题将进一步加剧。

总结一下：如果多个线程操作不同的成员变量， 但是这些变量存储在同一个缓存行，如果有处理器更新了缓存行的数据并刷新到主存之后，根据缓存一致性原则，其他处理器将失效该缓存行（I 状态）导致缓存未命中，需要重新去内存中读取最新数据，这就是伪共享问题。

• 特别是不同的线程操作同一个缓存行，需要发出 RFO（Request for Owner）信号锁定缓存行，保证写操作的原子性，此时其他线程不能操作这个缓存行，这将对效率有极大的影响。

为了避免避免经常执行写操作的变量因为在同一个缓存行而导致的伪共享问题，常用的解决方式就是缓存行填充，或者称为缓存行对齐。

缓存行填充的概念

当多个线程同时对共享的缓存行进行写操作的时候，因为缓存系统自身的缓存一致性原则，会引发伪共享问题，解决的常用办法是将共享变量根据缓存行大小进行补充对齐，使其加载到缓存时能够独享缓存行，避免与其他共享变量存储在同一个缓存行。

下面是缓存行实现，另外缓存行填充有一个前提同时分配的对象往往位于同一位置。

public long p1, p2, p3, p4, p5, p6, p7; // cache line paddingprivate volatile long cursor = INITIAL_CURSOR_VALUE;public long p8, p9, p10, p11, p12, p13, p14; // cache line padding

如果有不同的消费者往不同的字段写入，你需要确保各个字段间不会出现伪共享。

/** * 数组保存了VolatileLongPadding，其中数组中一个long类型保存数组长度，算上 * 自身long类型value，需要再填充6个long类型，就能将数组中的对象填充满一个缓存行。 * 注意：这里使用继承的方式实现缓存行对齐，因为Java编译器会优化无效的字段。 */class CacheLinePadding {    // 如果不需要填充，只需要注释掉这段代码即可    public volatile long p1, p2, p3, p4, p5, p6;}class CacheLinePaddingObject extends CacheLinePadding {    //实际操作的值    public volatile long value = 0L;}

RingBuffer 实现上同样也是使用了缓存行填充，保证了数组中的数据没有伪共享的存在，RingBuffer 除了一个 long 类型的 cursor 索引指针，p1->p7 都为缓存行填充，一般来讲 8 个 Long 类型的字段，正好是 64Byte，会填充一个缓存行，当然前提是你需要，因为毕竟还有你自己的数据信息字段。