Disruptor-源码解读 - IT评测·应用市场-qidao123.com

RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
// 获取下一个可用位置的下标(步骤1)
long sequence = ringBuffer.next();
try {
// 返回可用位置的元素
LongEvent event = ringBuffer.get(sequence);
// 设置该位置元素的值
event.set(l);
} finally {
// 发布
ringBuffer.publish(sequence);
}

复制代码

// 一般不会有以下写法，这里为了讲解源码才使用next(2)
// 向RingBuffer申请两个元素
long sequence = ringBuffer.next(2);
for (long i = sequence-1; i <= sequence; i++) {
try {
// 返回可用位置的元素
LongEvent event = ringBuffer.get(i);
// 设置该位置元素的值
event.set(1);
} finally {
ringBuffer.publish(i);
}
}

复制代码

abstract class SingleProducerSequencerPad extends AbstractSequencer
{
protected long p1, p2, p3, p4, p5, p6, p7;
SingleProducerSequencerPad(int bufferSize, WaitStrategy waitStrategy)
{
super(bufferSize, waitStrategy);
}
}
abstract class SingleProducerSequencerFields extends SingleProducerSequencerPad
{
SingleProducerSequencerFields(int bufferSize, WaitStrategy waitStrategy)
{
super(bufferSize, waitStrategy);
}
long nextValue = Sequence.INITIAL_VALUE;
long cachedValue = Sequence.INITIAL_VALUE;
}
public final class SingleProducerSequencer extends SingleProducerSequencerFields
{
protected long p1, p2, p3, p4, p5, p6, p7;
public SingleProducerSequencer(int bufferSize, WaitStrategy waitStrategy)
{
super(bufferSize, waitStrategy);
}
}

复制代码

// 调用路径
// RingBuffer#next()
// SingleProducerSequencer#next()
public long next(int n)
{
if (n < 1)
{
throw new IllegalArgumentException("n must be > 0");
}
long nextValue = this.nextValue;
//生产者当前序号值+期望获取的序号数量后达到的序号值
long nextSequence = nextValue + n;
//减掉RingBuffer的总的buffer值，用于判断是否出现‘覆盖’
long wrapPoint = nextSequence - bufferSize;
//从后面代码分析可得：cachedValue就是缓存的消费者中最小序号值，他不是当前最新的‘消费者中最小序号值’，而是上次程序进入到下面的if判定代码段时，被赋值的当时的‘消费者中最小序号值’
//这样做的好处在于：在判定是否出现覆盖的时候，不用每次都调用getMininumSequence计算‘消费者中的最小序号值’，从而节约开销。只要确保当生产者的节奏大于了缓存的cachedGateingSequence一个bufferSize时，从新获取一下 getMinimumSequence()即可。
long cachedGatingSequence = this.cachedValue;
//(wrapPoint > cachedGatingSequence) ：当生产者已经超过上一次缓存的‘消费者中最小序号值’（cachedGatingSequence）一个‘Ring’大小（bufferSize），需要重新获取cachedGatingSequence，避免当生产者一直在生产，但是消费者不再消费的情况下，出现‘覆盖’
//(cachedGatingSequence > nextValue) ： https://github.com/LMAX-Exchange/disruptor/issues/76
// 这里判断就是生产者生产的填满BingBUffer，需要等待消费者消费
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
{
cursor.setVolatile(nextValue); // StoreLoad fence
//gatingSequences就是消费者队列末尾的序列,也就是消费者消费到哪里了
//实际上就是获得处理的队尾,如果队尾是current的话,说明所有的消费者都执行完成任务在等待新的事件了
long minSequence;
while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue)))
{
// 等待1纳秒
LockSupport.parkNanos(1L); // TODO: Use waitStrategy to spin?
}
this.cachedValue = minSequence;
}
this.nextValue = nextSequence;
return nextSequence;
}
public void publish(long sequence)
{
// 更新序列号
cursor.set(sequence);
// 等待策略的唤醒
waitStrategy.signalAllWhenBlocking();
}

复制代码

public final class MultiProducerSequencer extends AbstractSequencer
{
// 缓存的消费者中最小序号值，相当于SingleProducerSequencerFields的cachedValue
private final Sequence gatingSequenceCache = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 标记元素是否可用
private final int[] availableBuffer;
public long next(int n)
{
if (n < 1)
{
throw new IllegalArgumentException("n must be > 0");
}
long current;
long next;
do
{
current = cursor.get();
next = current + n;
//减掉RingBuffer的总的buffer值，用于判断是否出现‘覆盖’
long wrapPoint = next - bufferSize;
//从后面代码分析可得：cachedValue就是缓存的消费者中最小序号值，他不是当前最新的‘消费者中最小序号值’，而是上次程序进入到下面的if判定代码段时，被赋值的当时的‘消费者中最小序号值’
//这样做的好处在于：在判定是否出现覆盖的时候，不用每次都调用getMininumSequence计算‘消费者中的最小序号值’，从而节约开销。只要确保当生产者的节奏大于了缓存的cachedGateingSequence一个bufferSize时，从新获取一下 getMinimumSequence()即可。
long cachedGatingSequence = gatingSequenceCache.get();
//(wrapPoint > cachedGatingSequence) ：当生产者已经超过上一次缓存的‘消费者中最小序号值’（cachedGatingSequence）一个‘Ring’大小（bufferSize），需要重新获取cachedGatingSequence，避免当生产者一直在生产，但是消费者不再消费的情况下，出现‘覆盖’
//(cachedGatingSequence > nextValue) ： https://github.com/LMAX-Exchange/disruptor/issues/76
// 这里判断就是生产者生产的填满BingBUffer，需要等待消费者消费
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
{
long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
if (wrapPoint > gatingSequence)
{
LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
continue;
}
gatingSequenceCache.set(gatingSequence);
}
// 使用cas保证只有一个生产者能拿到next
else if (cursor.compareAndSet(current, next))
{
break;
}
}
while (true);
return next;
}
......
}

复制代码

public void publish(final long sequence)
{
// 记录发布情况
setAvailable(sequence);
// 等待策略的唤醒
waitStrategy.signalAllWhenBlocking();
}
private void setAvailable(final long sequence)
{
// calculateIndex(sequence)：获取序号
// calculateAvailabilityFlag(sequence)：RingBuffer的圈数
setAvailableBufferValue(calculateIndex(sequence), calculateAvailabilityFlag(sequence));
}
private void setAvailableBufferValue(int index, int flag)
{
long bufferAddress = (index * SCALE) + BASE;
UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
// 上面相当于 availableBuffer[index] = flag 的高性能版
}

复制代码