使用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; }
参考