LMAX Disruptor 原理

r囧r小猫 2022-08-05 19:59 166阅读 0赞

# [fxjwind][] #

## [LMAX Disruptor 原理][LMAX Disruptor] ##

[http://mechanitis.blogspot.com/search/label/disruptor][http_mechanitis.blogspot.com_search_label_disruptor]

[http://ifeve.com/disruptor/][http_ifeve.com_disruptor], 并发框架Disruptor译文

[http://blog.sina.com.cn/s/blog\_68ffc7a4010150yl.html][http_blog.sina.com.cn_s_blog_68ffc7a4010150yl.html], 论文译文

LMAX需要搭建high performance的交易平台, 所以需要基于并发编程模型 ([并发编程模型和访问控制][Link 1])   
当然他们也关注类似Actor或SEDA模型, 并进行了测试, 从而发现了性能瓶颈-- 对于队列的管理

如图这样比较简单的处理流程, 就需要4个queue和大量的message发送, disruptor设计了一种高效的替代方案

解决如下问题,

![Center][]

#### 1, 用何种数据机构来实现Queue ####

[如何使用Disruptor（一）Ringbuffer的特别之处][Disruptor_Ringbuffer]

实现queue首先想到链表, 但使用链表有下列问题,   
\- 节点分散, 不利于cache预读   
\- 节点每次需要分配和释放, 需要大量的垃圾回收, 低效   
\- 不利于批量读取   
\- 竞争点较多, head指针, tail指针, size   
   由于producer和consumer很难同步, 所以大部分queue都是满或空状态, 这样会导致大量的竞争, 比较低效   
\- 而且习惯的编程方式导致head指针, tail指针, size常常在一个cacheline中, 造成伪共享问题

![Center 1][]

那么用数组实现, 可以部分解决前3点问题, 但仍然无法解决竞争点问题, 以及由于数组的fix size, 带来扩展性问题

Disruptor采用特殊的ring buffer来作为queue实现的数据结构, 解决了上述的问题   
并且这种ring buffer只用了一个标志指针, 即标志下一个写入位置   
求余操作本身也是一种高耗费的操作, 所以ringbuffer的size设成2的n次方, 可以利用位操作来高效实现求余

![Center 2][]

#### 2, 减少竞争点, 分离关注 ####

对于传统的3个竞争点, Disruptor成功的通过ring buffer将其降低到1个, 提高了效率   
只有producer需要关注这个写入标志位, 如果只有一个producer的话, 那么完全就不需要lock, 当然如果有多个producer的时候, 就需要通过ProducerBarrier在写入标志位上做互斥   
对于consumer, 每个consumer各自记录读入标志位, 并且通过ConsumerBarrier不停的侦听当前最大可读标志位, 即写入标志位   
这样的设计成功的将关注点分离

#### 3, Lock-free ####

前面说了disruptor减少竞争点, 但是不可能完全消除竞争, 对于写入标志位, 当多个producer的时候仍然存在竞争, 竞争就需要加锁.

[剖析Disruptor:为什么会这么快？(一)锁的缺点][Disruptor]

锁是很低效的, 论文中的3.1讲的比较清晰, 并通过实验数据证明了这点, 使用锁会慢1000倍   
\- 系统态的锁会导致线程cache丢失. 锁竞争的时候需要进行仲裁. 这个仲裁会涉及到操作系统的内核切换, 并且在此过程中操作系统需要做一系列操作, 导致原有线程的指令缓存和数据缓很可能被丢掉   
\- 用户态的锁往往是通过自旋锁来实现(自旋即忙等), 而自旋在竞争激烈的时候开销是很大的(一直在消耗CPU资源)

那么disruptor的怎么做? lock-free, 不使用锁, 使用CAS（Compare And Swap/Set）   
严格意义上说仍然是使用锁, 因为CAS本质上也是一种乐观锁, 只不过是CPU级别指令, 不涉及到操作系统, 所以效率很高   
Java提供CAS操作的支持, [AtomicLong][]

![Center 3][]

> CAS依赖于处理器的支持, 当然大部分现代处理器都支持.   
> CAS相对于锁是非常高效的, 因为它不需要涉及内核上下文切换进行仲裁.   
> 但CAS并不是免费的, 它会涉及到对指令pipeline加锁, 并且会用到内存barrier（用来刷新内存状态，简单理解就是把缓存中，寄存器中的数据同步到内存中去）

CAS的问题就是更为复杂, 比使用lock更难于理解, 并且虽然相对于lock已经很高效, 但是由于上面提到的耗费, 仍然比不使用任何锁机制要慢的多   
所以对于disruptor, 如果能保证只有一个producer就可以完全不使用lock, 甚至CAS, 是很高效的方案   
当然在不得不使用多个producer的情况下, 只能使用CAS

#### 4, 解决伪共享(False Sharing) ####

[剖析Disruptor:为什么会这么快？(二)神奇的缓存行填充][Disruptor 1]

[剖析Disruptor:为什么会这么快？(三)伪共享][Disruptor 2]

前面提到, CPU cache的预读会大大提高执行效率, 这也是为什么选择数组来替代链表的很重要的原因, 因为数组集中存储可以通过预读大大提高效率   
上面谈到lock的耗费, 主要也是由于内核的切换导致cache的丢失

所以cache是优化的关键, cache越接近core就越快,也越小   
可以看出对于L1, L2级别的cache是每个core都独立的

![Center 4][]

cache-line(缓存行)  
  
  
缓存是由缓存行组成的, 通常是64字节, 一个Java的long类型是8字节，因此在一个缓存行中可以存8个long类型的变量.   
缓存行是缓存更新的基本单位, 就算你只读一个变量, 系统也会预读其余7个, 并cache这一行, 并且这行中的任一变量发生改变, 都需要重新加载整行, 而非仅仅重新加载一个变量.

![Center 5][]

这里谈的伪共享问题, 也是一种主要的cache丢失的case, 需要通过缓存行填充来解决

上面的提到的cache-line, 对于象数组这样连续存储的数据结构非常高效, 但是不能保证所有结构都是连续存储的, 比如对于链表, 就很容易出现伪共享问题, 即这种预读反而使效率降低.   
底下是典型伪共享的例子, 在链表中往往会连续定义head和tail指针, 所以对于cache-line的预读, 很有可能会导致head和tail在同一cache-line   
在实际使用中, 往往producer线程会持续更改tail指针, 而consumer线程会持续更改head指针   
当producer线程和consumer线程分别被分配到core2和core1, 就会出现以下状况,   
由于core1不断改变h, 导致该cache-line过期, 对于core2, 虽然他不需要读h, 或者t也没有改变, 但是由于cache-line的整行更新, 所以core2仍然需要不停的更新它的cache   
core2的缓存未命中被一个和它本身完全不相干的值h, 而被大大提高, 导致cache效率底下   
而实际情况下, core1会不断更新h, 而core2会不断更新t, 导致core1和core2都需要频繁的重新load cache, 这就是伪共享问题

![Center 6][]

那么如何解决这个问题?   
既然预读反而降低效率, 解决办法就是消除系统预读的影响   
简单的办法就是缓存行填充, 来保证这个cache-line只存储这一个数据, 从而避免其他数据的更改对该cache-line的影响

当然显而易见, 这种缓存行填充是非常浪费的, cache本身就是很昂贵的资源, 所以必须慎用   
在 [Disruptor][Disruptor 3]里我们对 [RingBuffer][]的cursor和 [BatchEventProcessor][]的序列进行了缓存行填充

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

以cursor为例, 本身是独立的变量, 和其他的数据没有关联关系, 并且cursor会频繁的被所有线程读取, 所以如果由于其他不相关的变量的更改而导致cursor所在的cache-line被频繁reload, 是非常低效的.   
所以, disruptor在cursor前后都pading了7个long, 从而避免cursor和任意其他的变量在同一个cache-line   
使用缓存行填充的准则,    
独立变量, 变量被大量线程touch, 会被频繁使用和读取

#### 5, 使用内存屏障 ####

[http://ifeve.com/linux-memory-barriers/][http_ifeve.com_linux-memory-barriers], 非常详细的介绍了内存屏障的原理

[剖析Disruptor:为什么会这么快？(四)揭秘内存屏障][Disruptor 4]

[聊聊并发（一）深入分析Volatile的实现原理][Volatile]

首先, 内存屏障本身不是一种优化方式, 而是你使用lock-free(CAS)的时候, 必须要配合使用内存屏障

因为CPU和memory之间有多级cache, CPU core只会更新cache-line, 而cache-line什么时候flush到memory, 这个是有一定延时的   
在这个延时当中, 其他CPU core是无法得知你的更新的, 因为只有把cache-line flush到memory后, 其他core中的相应的cache-line才会被置为过期数据

所以如果要保证使用CAS能保证线程间互斥, 即乐观锁, 必须当一个core发生更新后, 其他所有core立刻知道并把相应的cache-line设为过期, 否则在这些core上执行CAS读到的都是过期数据   
系统提供内存屏障就是做这个事的, 当设置内存屏障, 会立刻将cache-line flush到memory, 而没有延时

Java中用volatile来实现内存屏障

> Java语言规范第三版中对volatile的定义如下： java编程语言允许线程访问共享变量，为了确保共享变量能被准确和一致的更新，线程应该确保通过排他锁单独获得这个变量。Java语言提供了 volatile，在某些情况下比锁更加方便。如果一个字段被声明成volatile，java线程内存模型确保所有线程看到这个变量的值是一致的

volatile保证了线程间的可见性, 但是同样如果要实现互斥, 必须借助CAS, 以避免读取到更新之间的数据变更

> volatile的实现实质,
> 
> \- 将当前处理器缓存行的数据会写回到系统内存   
> \- 这个写回内存的操作会引起在其他CPU里缓存了该内存地址的数据无效

内存屏障另一种用途, CPU出于对执行指令和数据加载的优化会调整执行顺序, 所以在代码里面先写的指令不一定会被先执行, 当然是在保证逻辑一致性的前提下.   
但内存屏障, 可以限制这种调整, 屏障之前的命令必须先于屏障执行, 而屏障之后的必须后于屏障执行, 很形象.

所以可以看到内存屏障, 虽然和lock比是高效的, 但毕竟限制了CPU的优化并会强制flush cache-line, 所以仍然是比较昂贵的操作.

#### 6, 如何使用Disruptor替代Queue ####

[解析Disruptor关系组装][Disruptor 5]

我本来以为是用一个ringbuffer替代一个queue, 原来是用一个ringbuffer替代所有的queue, 怎么实现的?

![Center 7][] ![Center 8][]

如图, 所有consumer都是从RingBuffer里面读数据   
而C3, 依赖于C1和C2的执行结果, 那么通过设置ConsumerBarrier2来监控C1和C2的执行序号

那么有个问题是C3, 如何获得C1和C2的执行结果?   
答案是, C1和C2执行完后, 会把结果写回Ringbuffer中原来的entry中

![Center 9][]

如图, 当C3拿到Entry时, 里面有3个值, 本来的value, C1处理的结果, C2处理的结果, 并且不同的consumer写的字段不一样来避免冲突   
而Producer在监控consumer消费序号时, 只需要监控最后一层的, 即C3的, 因为只有C3处理完, 这个entry才能被覆盖.

看起来非常的复杂, 但是在使用时, 对用户很多机制其实是透明的, 比如上面的workflow的代码如下

ConsumerBarrier consumerBarrier1 =
        ringBuffer.createConsumerBarrier();
    BatchConsumer consumer1 =
        new BatchConsumer(consumerBarrier1, handler1);
    BatchConsumer consumer2 =
        new BatchConsumer(consumerBarrier1, handler2);
    ConsumerBarrier consumerBarrier2 =
        ringBuffer.createConsumerBarrier(consumer1, consumer2);
    BatchConsumer consumer3 =
        new BatchConsumer(consumerBarrier2, handler3);
    ProducerBarrier producerBarrier =
        ringBuffer.createProducerBarrier(consumer3);

对用户而言, 只需要知道ConsumerBarrier, Consumer, ProducerBarrier

#### 总结 ####

总体来说, disprutor从两个方面来对Actor模式的queue做了优化

最重要的是, Mechanical Sympathy(机械的共鸣), 了解硬件的工作方式来编写和硬件完美结合的软件, 很高的境界   
通过利用CAS+内存屏障实现lock-free, 并使用缓存行填充来解决伪共享, 可见虽然编程语言已经发展到很高级的地步, 但是如果要追求效率的机制, 必须要具有Mechanical Sympathy, 人剑合一

其次, 是通过ringbuffer来实现queue来替代链表的实现, 尤其当场景比较复杂需要很多queue的时候, 效率应该会得到很大的提高

其实, disruptor并没有实现queue的互斥consumer, 每个consumer都是自己保持序号, 各读各得, 但是对于普通queue, 被一个线程pop掉的数据, 其他线程是无法读到的

转帖：http://www.cnblogs.com/fxjwind/p/3180073.html

[fxjwind]: http://www.cnblogs.com/fxjwind/
[LMAX Disruptor]: http://www.cnblogs.com/fxjwind/p/3180073.html
[http_mechanitis.blogspot.com_search_label_disruptor]: http://mechanitis.blogspot.com/search/label/disruptor
[http_ifeve.com_disruptor]: http://ifeve.com/disruptor/
[http_blog.sina.com.cn_s_blog_68ffc7a4010150yl.html]: http://blog.sina.com.cn/s/blog_68ffc7a4010150yl.html
[Link 1]: http://www.cnblogs.com/fxjwind/p/3170032.html
[Center]: /images/20220806/8982ef0da27f454ba7c6bca074580aa1.png
[Disruptor_Ringbuffer]: http://ifeve.com/dissecting-disruptor-whats-so-special/
[Center 1]: /images/20220806/46e63b1b1c9e4d798ac7dd9932d5bf0c.png
[Center 2]: /images/20220806/682be12841ca459ab87d650082691381.png
[Disruptor]: http://ifeve.com/locks-are-bad/
[AtomicLong]: http://download.oracle.com/javase/6/docs/api/java/util/concurrent/atomic/AtomicLong.html
[Center 3]: /images/20220806/0f75e2e0127c418ab37e2f58fd95ee77.png
[Disruptor 1]: http://ifeve.com/disruptor-cacheline-padding/
[Disruptor 2]: http://ifeve.com/falsesharing/
[Center 4]: /images/20220806/43d6b437e8264020baecc24654a2e3ce.png
[Center 5]: /images/20220806/a01f2d7a915d4f92ada519a1a9c6fdf4.png
[Center 6]: /images/20220806/03d970a92019466caa9dc32aa77c0d91.png
[Disruptor 3]: http://code.google.com/p/disruptor/
[RingBuffer]: http://code.google.com/p/disruptor/source/browse/trunk/code/src/main/com/lmax/disruptor/RingBuffer.java
[BatchEventProcessor]: http://code.google.com/p/disruptor/source/browse/trunk/code/src/main/com/lmax/disruptor/BatchEventProcessor.java
[http_ifeve.com_linux-memory-barriers]: http://ifeve.com/linux-memory-barriers/
[Disruptor 4]: http://ifeve.com/disruptor-memory-barrier/
[Volatile]: http://ifeve.com/volatile/
[Disruptor 5]: http://ifeve.com/dissecting-disruptor-wiring-up-cn/
[Center 7]: https://img-blog.csdn.net/20141203214538921?watermark/2/text/aHR0cDovL2Jsb2cuY3Nkbi5uZXQvTWlyYWdlNTIw/font/5a6L5L2T/fontsize/400/fill/I0JBQkFCMA==/dissolve/70/gravity/Center
[Center 8]: /images/20220806/b9b2c7bfc7b241b4bdcbd45ad69e9c36.png
[Center 9]: /images/20220806/e94ae2c32ae74a48bb6cee1fe121678c.png