并发编程框架Disruptor之高性能设计

Posted 2022-04-09 公众号JavaEdge

架构 UML

1 单线程写

Disruptor的RingBuffer, 之所以可以做到完全无锁，也是因为"单线程写"，这是所有"前提的前提"，离了这个前提条件，没有任何技术可以做到完全无锁。Redis、Netty等等高性能技术框架的设计都是这个核心思想。

2 系统内存优化-内存屏障

实现无锁，还需一个关键技术：内存屏障。
对应到Java语言，就是valotile变量与happens before语义。

参阅： ​​内存屏障 - Linux的smp_wmb()/smp_ rmb()​​

系统内核：比如Linux的kfifo：smp_ wmb()，无论是底层的读写

都是使用了Linux的smp_ wmb

https://github.com/opennetworklinux/linux-3.8.13/blob/master/kernel/kfifo.c

3 系统缓存优化-消除伪共享

缓存系统中是以缓存行(cache line) 为单位存储的。缓存行是2的整数幂个连续字节，一般为32-256个字节。最常见的缓存行大小是64个字节。
当多线程修改互相独立的变量时，如果这些变量共享同一个缓存行，就会无意中影响彼此的性能，这就是伪共享。

核心：Sequence

可看成是个AtomicLong，用于标识进度。还有防止不同Sequence之间CPU缓存伪共享(Flase Sharing)。
如下设计保证保存的 value 永远在一个缓存行中。（8 个long，正好 64 字节），空间换时间。这些变量就是没有实际意义，只是帮助我们进行缓存行填充（Padding Cache Line），使得我们能够尽可能地用上CPU高速缓存（CPU Cache）

若访问内置在CPU的L1 Cache或L2 Cache，访问延时是内存的1/15乃至1/100。而内存访问速度远慢于CPU。想追求极限性能，需尽可能多从CPU Cache拿数据，而非从内存。
CPU Cache装载内存里的数据，不是一个个字段加载，而是加载整个缓存行。
如定义长度64的long类型数组，则数据从内存加载到CPU Cache，不是一个个数组元素加载，而是一次性加载固定长度的一个缓存行。
64位Intel CPU计算机的缓存行通常64个字节（Bytes）。一个long数据需8字节，所以一下会加载8个long数据。
即一次加载数组里面连续的8个数值。这样的加载使得遍历数组元素时，会很快。因为后面连续7次的数据访问都会命中缓存，无需重新从内存里读取数据。
但不使用数组，而使用单独变量时，这就出问题了。
Disruptor RingBuffer（环形缓冲区）定义了RingBufferFields类，里面有indexMask和其他几个变量存放RingBuffer的内部状态信息。

CPU在加载数据时，自然也会把这个数据从内存加载到高速缓存。
但这时，高速缓存除了这个数据，还会加载这个数据前后定义的其他变量。

这时，问题就来了，Disruptor是个多线程的服务器框架，在这个数据前后定义的其他变量，可能会被多个不同线程更新、读取数据。这些写入及读取的请求，会来自不同 CPU Core。于是，为保证数据的同步更新，不得不把CPU Cache里的数据，重新写回内存或重新从内存里加载数据。
这些CPU Cache的写回和加载，都不是以一个变量作为单位。这些都是以整个Cache Line作为单位。
所以，当INITIAL_CURSOR_VALUE 前后的那些变量被写回到内存时，这个字段自己也写回到了内存，这个常量的缓存也就失效了。
当要再次读取这个值时，要再重新从内存读取。这就意味着，读取速度大大变慢。

对此，Disruptor利用了缓存行填充，在 RingBufferFields里面定义的变量的前后，分别定义了7个long类型的变量：

• 前面7个来自继承的 RingBufferPad 类
• 后面7个直接定义在 RingBuffer 类

这14个变量无任何实际用途。我们既不读他们，也不写他们。
而RingBufferFields里面定义的这些变量都是final，第一次写入后就不会再修改。

所以，一旦它被加载到CPU Cache后，只要被频繁读取访问，就不会再被换出Cache。这意味着，对于该值的读取速度，会一直是CPU Cache的访问速度，而非内存的访问速度。

使用RingBuffer，利用缓存和分支预测

这利用CPU Cache的性能的思路，贯穿整个Disruptor。Disruptor整个框架，就是个高速的生产者-消费者模型（Producer-Consumer）下的队列。
生产者不停往队列生产新的待处理任务，而消费者不停从队列处理掉这些任务。

实现一个队列，最合适的是链表。只要维护好链表头和尾。生产者只要不断地往链尾插入新节点，
消费者只需不断从头部取出最老节点处理。LinkedBlockingQueue可直接用在生产者-消费者模式。

Disruptor并没用LinkedBlockingQueue，而使用个RingBuffer数据结构，这个RingBuffer底层是个固定长度的数组。比起链表，数组的数据在内存里会存在空间局部性。
数组的连续多个元素会一并加载到CPU Cache，所以访问遍历的速度会更快。而链表里面各个节点的数据，多半不会出现在相邻的内存空间，自然也就享受不到整个Cache Line加载后数据连续从高速缓存里面被访问到的优势。
数据的遍历访问还有一个很大的优势，就是CPU层面的分支预测会很准确。这可以使得我们更有效地利用了CPU里面的多级流水线，我们的程序就会跑得更快。这一部分的原理如果你已经不太记得了，可以回过头去复习一下第25讲关于分支预测的内容。

4 算法优化-序号栅栏机制

我们在生产者进行投递Event的时候，总是会使用:

long sequence = ringBuffer.next();

Disruptor3.0中，序号栅栏SequenceBarrier和序号Sequence搭配使用。协调和管理消费者与生产者的工作节奏，避免了锁和CAS的使用。

• 消费者序号数值必须小于生产者序号数值
• 消费者序号数值必须小于其前置(依赖关系)消费者的序号数值
• 生产者序号数值不能大于消费者中最小的序号数值
• 以避免生产者速度过快，将还未来得及消费的消息覆盖

SingleProducerSequencerPad#next

    /**
     * @see Sequencer#next(int)
     */
    @Override
    public long next(int n) // 1
    
        if (n < 1) // 初始值：sequence = -1
        
            throw new IllegalArgumentException("n must be > 0");
        
        // 语义级别的
        // nextValue为SingleProducerSequencer的变量
        long nextValue = this.nextValue;

        long nextSequence = nextValue + n;
        // 用于判断当前序号是否绕过整个 ringbuffer 容器
        long wrapPoint = nextSequence - bufferSize;
        // 用于缓存优化
        long cachedGatingSequence = this.cachedValue;

        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
        
            long minSequence;
            while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue)))
            
                LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin?
            

            this.cachedValue = minSequence;
        

        this.nextValue = nextSequence;

        return nextSequence;
    

参考

• https://zhuanlan.zhihu.com/p/21355046