disruptor学习笔记

Disruptor它是一个开源的并发框架，并获得2011 Duke’s 程序框架创新奖，能够在无锁的情况下实现网络的Queue并发操作。项目地址

Disruptor可以用来做什么？

当前业界开源组件使用Disruptor的包括Log4j2、Apache Storm等，它可以用来作为高性能的有界内存队列，基于生产者消费者模式，实现一个/多个生产者对应多个消费者。它也可以认为是观察者模式的一种实现，或者发布订阅模式。

Disruptor:为什么会这么快？

锁技术是慢的

如果有两个线程同时尝试修改同一个变量value，那么将会出现线程安全问题。如何解决？

悲观锁
只要线程２一获得Entry 的互斥锁，它就会阻击其它线程去改变它，然后它就可以随意做它要做的事情，设置值，然后做其它事情。
所以这是非常耗性能的，因为其它线程在系统各处徘徊着准备要获得锁然后又阻塞。线程越多，系统的响应性就会越慢．
乐观锁
乐观锁（ Optimistic Locking）其实是一种思想，相对悲观锁而言，乐观锁假设认为数据一般情况下不会造成冲突，所以在数据进行提交更新的时候，才会正式对数据的冲突与否进行检测。
线程１在线程２读完后到达并把值改为”blah”,线程２读到了这个新值，线程２不会把＂fluffy＂写到Entry里并把线程１所写的数据覆盖．线程２会重试（重新读新的值，与旧值比较，如果相等则在变量的值后面附上’y’）

很明确的一个问题：锁技术是慢的

锁定会带来各种各样的问题，比如死锁。另外在多线程竞争下，加锁、释放锁会导致比较多的上下文切换和调度延时，引起性能问题。
一个线程持有锁会导致其它所有需要此锁的线程挂起。
如果一个优先级高的线程等待一个优先级低的线程释放锁会导致优先级倒置，引起性能风险。

Disruptor如何解决这些问题

使用CAS（Compare And Swap/Set）操作。
这是一个CPU级别的指令，它的工作方式有点像乐观锁——CPU去更新一个值，但如果想改的值不再是原来的值，操作就失败，因为很明显，有其它操作先改变了这个值。
CAS操作比锁消耗资源少的多，因为它们不牵涉操作系统，它们直接在CPU上操作。但它们并非没有代价——在上面的试验中，单线程无锁耗时300ms，单线程有锁耗时10000ms，单线程使用CAS耗时5700ms。所以它比使用锁耗时少，但比不需要考虑竞争的单线程耗时多。
RingBuffer的特殊设计
传统队列在头、尾和大小变量上总是会有很多竞争，或者如果消费操作移除元素时需要使用一个写操作，那元素本身也包含竞争。RingBuffer只需要一个序列号来保存下一个可用的空间，减小竞争。
缓存行
意味着没有伪共享和非预期的竞争。
常用处理器的缓存行是64字节的，比较旧的处理器缓存行是32字节，并且它有效地引用主内存中的一块地址。

如上图在核心1上运行的线程想更新变量X，同时核心2上的线程想要更新变量Y。不幸的是，这两个变量在同一个缓存行中。每个线程都要去竞争缓存行的所有权来更新变量。如果核心1获得了所有权，缓存子系统将会使核心2中对应的缓存行失效。当核心2获得了所有权然后执行更新操作，核心1就要使自己对应的缓存行失效。这会来来回回的经过L3缓存，大大影响了性能。

Disruptor如何解决的？

class LhsPadding
{
    protected long p1, p2, p3, p4, p5, p6, p7;
}

class Value extends LhsPadding
{
    protected volatile long value;
}

一个Java的long类型是8字节，因此在一个缓存行中可以存8个long类型的变量。
所以如果你访问一个long数组，当数组中的一个值被加载到缓存中，它会额外加载另外7个。因此你能非常快地遍历这个数组
另外确保不会有两个线程往同一个缓存行中写，减少缓存行竞争

内存屏障

内存屏障是什么
它是一个CPU指令， a)确保一些特定操作执行的顺序； b)影响一些数据的可见性
编译器和CPU可以在保证输出结果一样的情况下对指令重排序，使性能得到优化。插入一个内存屏障，相当于告诉CPU和编译器先于这个命令的必须先执行，后于这个命令的必须后执行。
内存屏障另一个作用是强制更新一次不同CPU的缓存。
RingBuffer的使用场景
RingBuffer的指针(cursor)（指向队尾元素）属于一个volatile变量，同时也是我们能够不用锁操作就能实现Disruptor的原因之一。
（1）生产者

生产者对ring buffer调用commit方法来更新序列号（把cursor更新为该Entry的序列号）。对volatile字段(cursor)的写操作创建了一个内存屏障，这个屏障将刷新所有缓存里的值（或者至少相应地使得缓存失效）。
这时候，消费者们能获得最新的序列号码，并且因为内存屏障保证了它之前执行的指令的顺序
（2）消费者

消费者中的序列号是volatile类型的，为了防止下游的消费者和上游的消费者对同一个Entry竞争消费，导致在环形队列中互相覆盖数据，下游消费者要对上游消费者的消费情况进行跟踪
所以，如果你的下游消费者(C2)看见前一个消费者(C1)在消费号码为12的Entry，当C2的读取也到了12，它在更新序列号前将可以获得C1对该Entry的所作的更新。
基本来说就是，C1更新序列号前对ring buffer的所有操作必须先发生，待C2拿到C1更新过的序列号之后，C2才可以为所欲为
（3）对性能的影响
内存屏障作为另一个CPU级的指令，没有锁那样大的开销。但是内存屏障是有开销的——编译器/cpu不能重排序指令，导致不可以尽可能地高效利用CPU，另外刷新缓存亦会有开销。
所以Disruptor的实现对序列号的读写频率尽量降到最低。对volatile字段的每次读或写都是相对高成本的操作。因此在批量的情况下可以获得很好的性能。比如先获得一整批Entries,并在更新序列号前处理它们。
代码如下：

 long nextSequence = sequence + 1;
    while (running){
        try{
            final long availableSequence = consumerBarrier.waitFor(nextSequence);
            while (nextSequence <= availableSequence) {
                entry = consumerBarrier.getEntry(nextSequence);
                handler.onAvailable(entry);
                nextSequence++;
            }
            handler.onEndOfBatch();
            sequence = entry.getSequence();
        }
        …
        catch (final Exception ex) {
            exceptionHandler.handle(ex, entry);
            sequence = entry.getSequence();
            nextSequence = entry.getSequence() + 1;
        }
    }

在上面的代码中，我们在消费者处理entries的循环中用一个局部变量（nextSequence）来递增。这表明我们想尽可能地减少对volatile类型的序列号的进行读写。

Disruptor如何工作和使用

ringbuffer到底是什么

它是一个环（首尾相接的环），你可以把它用做在不同上下文（线程）间传递数据的buffer。

基本来说，ringbuffer拥有一个序号，这个序号指向数组中下一个可用的元素。（如下图右边的图片表示序号，这个序号指向数组的索引4的位置。）

随着你不停地填充这个buffer（可能也会有相应的读取），这个序号会一直增长，直到绕过这个环。

要找到数组中当前序号指向的元素，可以通过sequence & （array length－1） = array index，比如一共有8槽，3&（8－1）=3，HashMap就是用这个方式来定位数组元素的，这种方式比取模的速度更快。

常用的队列之间的区别
没有尾指针。只维护了一个指向下一个可用位置的序号。
不删除buffer中的数据，也就是说这些数据一直存放在buffer中，直到新的数据覆盖他们
ringbuffer采用这种数据结构原因

它在可靠消息传递方面有很好的性能
因为它是数组，所以要比链表快，数组内元素的内存地址的连续性存储的。这是对CPU缓存友好的—也就是说，在硬件级别，数组中的元素是会被预加载的，因此在ringbuffer当中，cpu无需时不时去主存加载数组中的下一个元素。因为只要一个元素被加载到缓存行，其他相邻的几个元素也会被加载进同一个缓存行。
其次，你可以为数组预先分配内存，使得数组对象一直存在（除非程序终止）。这就意味着不需要花大量的时间用于垃圾回收。此外，不像链表那样，需要为每一个添加到其上面的对象创造节点对象—对应的，当删除节点时，需要执行相应的内存清理操作。

如何从Ringbuffer读取

消费者(Consumer)是一个想从Ring Buffer里读取数据的线程，它可以访问ConsumerBarrier对象——这个对象由RingBuffer创建并且代表消费者与RingBuffer进行交互。就像Ring Buffer显然需要一个序号才能找到下一个可用节点一样，消费者也需要知道它将要处理的序号——每个消费者都需要找到下一个它要访问的序号。在上面的例子中，消费者处理完了Ring Buffer里序号8之前（包括8）的所有数据，那么它期待访问的下一个序号是9。

消费者可以调用ConsumerBarrier对象的waitFor()方法，传递它所需要的下一个序号.

final long availableSeq = consumerBarrier.waitFor(nextSequence);

ConsumerBarrier返回RingBuffer的最大可访问序号——在上面的例子中是12。ConsumerBarrier有一个WaitStrategy方法来决定它如何等待这个序号.

接下来

消费者会一直原地停留，等待更多数据被写入Ring Buffer。并且，一旦数据写入后消费者会收到通知——节点9，10，11和12 已写入。现在序号12到了，消费者可以让ConsumerBarrier去拿这些序号节点里的数据了。

拿到了数据后，消费者(Consumer)会更新自己的标识(cursor)。

好处

这样做有助于平缓延迟的峰值了——以前需要逐个节点地询问“我可以拿下一个数据吗？现在可以了么？现在呢？”，消费者(Consumer)现在只需要简单的说“当你拿到的数字比我这个要大的时候请告诉我”，函数返回值会告诉它有多少个新的节点可以读取数据了。因为这些新的节点的确已经写入了数据（Ring Buffer本身的序号已经更新），而且消费者对这些节点的唯一操作是读而不是写，因此访问不用加锁。这太好了，不仅代码实现起来可以更加安全和简单，而且不用加锁使得速度更快。

另一个好处是——你可以用多个消费者(Consumer)【实现BatchHandler】去读同一个RingBuffer ，不需要加锁，也不需要用另外的队列来协调不同的线程(消费者)。这样你可以在Disruptor的协调下实现真正的并发数据处理。

问题：怎么处理多个消费者不读到同样的数据?

写入 Ringbuffer

问题：如何不要让 Ring 重叠；如何通知消费者；生产者一端的批处理；以及多个生产者如何协同工作？

ProducerBarriers

Ring Buffer 提供了一个 ProducerBarrier 对象，让生产者通过它来写入 Ring Buffer。

写入 Ring Buffer 的过程涉及到两阶段提交 (two-phase commit)。首先，你的生产者需要申请 buffer 里的下一个节点。然后，当生产者向节点写完数据，它将会调用 ProducerBarrier 的 commit 方法。

那么让我们首先来看看第一步。 “给我 Ring Buffer 里的下一个节点”，这句话听起来很简单。的确，从生产者角度来看它很简单：简单地调用 ProducerBarrier 的 nextEntry() 方法，这样会返回给你一个 Entry 对象，这个对象就是 Ring Buffer 的下一个节点。

ProducerBarrier 如何防止 Ring Buffer 重叠

在后台，由 ProducerBarrier 负责从 Ring Buffer 中找到下一个节点，然后才允许生产者向它写入数据。

在这幅图中，我们假设只有一个生产者写入 Ring Buffer。

ConsumerTrackingProducerBarrier 对象拥有所有正在访问 Ring Buffer 的 消费者列表。Disruptor 由消费者负责通知它们处理到了哪个序列号，而不是 Ring Buffer。所以，如果我们想确定我们没有让 Ring Buffer 重叠，需要检查所有的消费者们都读到了哪里。

在上图中，有一个 消费者 顺利的读到了最大序号 12（用红色/粉色高亮）。第二个消费者 有点儿落后——可能它在做 I/O 操作之类的——它停在序号 3。因此消费者 2 在赶上消费者 1 之前要跑完整个 Ring Buffer 一圈的距离。

现在生产者想要写入 Ring Buffer 中序号 3 占据的节点，因为它是 Ring Buffer 当前游标的下一个节点。但是 ProducerBarrier 明白现在不能写入，因为有一个消费者正在占用它。所以，ProducerBarrier 停下来自旋 (spins)，等待，直到那个消费者离开。

申请下一个节点

现在可以想像消费者 2 已经处理完了一批节点，并且向前移动了它的序号。可能它挪到了序号 9

上图显示了当消费者 2 挪动到序号 9 时发生的情况。
ProducerBarier 会看到下一个节点——序号 3 那个已经可以用了。它会抢占这个节点上的 Entry（理解为一个放写入到某个序号的 Ring Buffer 数据的桶），把下一个序号（13）更新成 Entry 的序号，然后把 Entry 返回给生产者。生产者可以接着往 Entry 里写入数据。

提交新的数据

绿色表示最近写入的 Entry，序号是 13
当生产者结束向 Entry 写入数据后，它会要求 ProducerBarrier 提交。

ProducerBarrier等待 Ring Buffer 游标到达准备提交的节点的前一个节点时它才会提交（对于单生产者这毫无意义－比如，我们已经知道游标到了 12 ，而且没有其他人正在写入 Ring Buffer）。然后 ProducerBarrier 更新 Ring Buffer 的游标到刚才写入的 Entry 序号－在我们这儿是 13。接下来，ProducerBarrier 会让消费者知道 buffer 中有新东西了。它戳一下 ConsumerBarrier 上的 WaitStrategy 对象说－“喂，醒醒！有事情发生了！”（注意－不同的 WaitStrategy 实现以不同的方式来实现提醒，取决于它是否采用阻塞模式。）

现在消费者 1 可以读 Entry 13 的数据，消费者 2 可以读 Entry 13 以及前面的所有数据。

ProducerBarrier 上的批处理

Disruptor 可以同时在生产者和消费者两端实现批处理。还记得伴随着程序运行，消费者 2 最后达到了序号 9 吗？ProducerBarrier 可以在这里做一件很狡猾的事－它知道 Ring Buffer 的大小，也知道最慢的消费者位置。因此它能够发现当前有哪些节点是可用的。

如果 ProducerBarrier 知道 Ring Buffer 的游标指向 12，而最慢的消费者在 9 的位置，它就可以让生产者写入节点 3，4，5，6，7 和 8，中间不需要再次检查消费者的位置。

多个生产者的场景

多个生产者的场景下，同时在向 Ring Buffer 写入，就有可能出现某些 Entry 正在被生产者写入但还没有提交的情况。

每个生产者都向 ClaimStrategy 申请下一个可用的节点。生产者 1 拿到序号 13，这和上面单个生产者的情况一样。生产者 2 拿到序号 14，尽管 Ring Buffer的当前游标仅仅指向 12。这是因为 ClaimSequence 不但负责分发序号，而且负责跟踪哪些序号已经被分配。

现在每个生产者都拥有自己的写入节点和一个崭新的序号。

生产者 1 和它的写入节点是绿色，生产者 2 和它的写入节点是紫色

假设生产者 1 因为某些原因没有来得及提交数据。生产者 2 已经准备好提交了，并且向 ProducerBarrier 发出了请求。

就像先前在 commit 示意图中看到的一样，ProducerBarrier 只有在 Ring Buffer 游标到达准备提交的节点的前一个节点时它才会提交。在当前情况下，游标必须先到达序号 13 我们才能提交节点 14 的数据。但是我们不能这样做，因为生产者 1 某些原因，还没来得及提交。因此 ClaimStrategy 就停在那儿自旋 (spins)，直到 Ring Buffer 游标到达它应该在的位置。

现在生产者 1 从清醒过来并且申请提交节点 13 的数据（生产者 1 发出的绿色箭头代表这个请求）。ProducerBarrier 让 ClaimStrategy 先等待 Ring Buffer 的游标到达序号 12，当然现在已经到了。因此 Ring Buffer 移动游标到 13，让 ProducerBarrier 戳一下 WaitStrategy 告诉所有人都知道 Ring Buffer 有更新了。现在 ProducerBarrier 可以完成生产者 2 的请求，让 Ring Buffer 移动游标到 14，并且通知所有人都知道。

你会看到，尽管生产者在不同的时间完成数据写入，但是 Ring Buffer 的内容顺序总是会遵循 nextEntry() 的初始调用顺序。也就是说，如果一个生产者在写入 Ring Buffer 的时候暂停了，只有当它解除暂停后，其他等待中的提交才会立即执行。

参考链接

Disruptor