如何使用Disruptor(二)如何从Ringbuffer读取
作者:Trisha 译者:古圣昌 校对:方腾飞
从上一篇文章中我们都了解了什么是Ring Buffer以及它是如何的特别。但遗憾的是,我还没有讲述如何使用Disruptor向Ring Buffer写数据和从Ring Buffer中读取数据。
ConsumerBarrier与消费者
(好,我开始后悔使用Paint/Gimp 了。尽管这是个购买绘图板的好借口,如果我继续写下去的话… UML界的权威们大概也在诅咒我的名字了。)
消费者(Consumer)是一个想从Ring Buffer里读取数据的线程,它可以访问ConsumerBarrier对象——这个对象由RingBuffer创建并且代表消费者与RingBuffer进行交互。就像Ring Buffer显然需要一个序号才能找到下一个可用节点一样,消费者也需要知道它将要处理的序号——每个消费者都需要找到下一个它要访问的序号。在上面的例子中,消费者处理完了Ring Buffer里序号8之前(包括8)的所有数据,那么它期待访问的下一个序号是9。
消费者可以调用ConsumerBarrier对象的waitFor()方法,传递它所需要的下一个序号.
[code lang=”java”] final long availableSeq = consumerBarrier.waitFor(nextSequence);[/code]
ConsumerBarrier返回RingBuffer的最大可访问序号——在上面的例子中是12。ConsumerBarrier有一个WaitStrategy方法来决定它如何等待这个序号,我现在不会去描述它的细节,代码的注释里已经概括了每一种WaitStrategy的优点和缺点 。
接下来怎么做?
接下来,消费者会一直原地停留,等待更多数据被写入Ring Buffer。并且,一旦数据写入后消费者会收到通知——节点9,10,11和12 已写入。现在序号12到了,消费者可以让ConsumerBarrier去拿这些序号节点里的数据了。
拿到了数据后,消费者(Consumer)会更新自己的标识(cursor)。
你应该已经感觉得到,这样做是怎样有助于平缓延迟的峰值了——以前需要逐个节点地询问“我可以拿下一个数据吗?现在可以了么?现在呢?”,消费者(Consumer)现在只需要简单的说“当你拿到的数字比我这个要大的时候请告诉我”,函数返回值会告诉它有多少个新的节点可以读取数据了。因为这些新的节点的确已经写入了数据(Ring Buffer本身的序号已经更新),而且消费者对这些节点的唯一操作是读而不是写,因此访问不用加锁。这太好了,不仅代码实现起来可以更加安全和简单,而且不用加锁使得速度更快。
另一个好处是——你可以用多个消费者(Consumer)去读同一个RingBuffer ,不需要加锁,也不需要用另外的队列来协调不同的线程(消费者)。这样你可以在Disruptor的协调下实现真正的并发数据处理。
BatchConsumer代码是一个消费者的例子。如果你实现了BatchHandler, 你可以用BatchConsumer来完成上面我提到的复杂工作。它很容易对付那些需要成批处理的节点(例如上文中要处理的9-12节点)而不用单独地去读取每一个节点。
更新:注意Disruptor 2.0版本使用了与本文不一样的命名。如果你对类名感到困惑,请阅读我的变更总结。
原创文章,转载请注明: 转载自并发编程网 – ifeve.com本文链接地址: 如何使用Disruptor(二)如何从Ringbuffer读取
Disruptor的这个设计真不错。消费者不能直接和ringBuffer进行通讯,而必须通过ConsumerBarrier。
怎么处理多个消费者不读到同样的数据?
为什么我感觉就是把一个窗口买票改成多个窗口买票?
应该是把争抢跑到多个窗口买票的行为,通过Barrier规范为:所有人先到Barrier处拿号,然后凭号去对应窗口买票。
拿号消耗的时间很短,一个Barrier就够;买票消耗时间很长,各个窗口独立服务。
RingBarrier是否可能成为性能瓶颈?它代理了所有其他线程对RingBuffer的访问句柄申请工作。
请教各位大神,SingleProducerSequencer的next(n)方法中有句if条件,第一个条件我看懂了,请问第二个条件cachedGatingSequence > nextValue是什么场景才出现的呢?
好吧。
如果仅从disruptor外部调用,是不会产生nextValue回溯的情况。disruptor内部有一个方法RingBuffer.resetTo()可以,不过这个方法已经被Deprecated了。
作者完全在自嗨