Disruptor(四)RingBuffer多生产者写入

上一章主要介绍了单个生产者如何向RingBuffer数据写入数据,如何不要让Ring重叠,写入后通知消费者,生产者一端的批处理,以及多个生产者如何协同工作,本章主要介绍多生产者向RingBuffer数据写入数据。

1、多生产者MultiProducerSequencer申请下一个节点


和单生产者不同的是在next方法中会直接通过cursor.compareAndSet(current,
next)
设置生产者的游标cursorsequence。大家很可能会问设置了生产者的游标后,没有提交数据之前,多生产者场景中消费者是否就能够获取到数据,答案是否定的,在MultiProducerSequencer实现的getHighestPublishedSequence的方法和单生产者有所区别,后面会详细讲解。

2多生产者MultiProducerSequencer提交数据

和单生产者的区别是使用setAvailable将数据设置成可用状态。

在多个生产者的场景下,还需要其他东西来追踪序号。这个序号是指当前可写入的序号。注意这和“向RingBuffer的游标加1”不一样,如果你有一个以上的生产者同时在向RingBuffer写入,就有可能出现某些Entry正在被生产者写入但还没有提交的情况。

生产者1拿到序号14,生产者2拿到序号15。现在假设生产者1因为某些原因没有来得及提交数据。

生产者2通过setAvailable(15)请求完成提交数据,如图所示。

当这个时候消费者通过waitFor(14),返回的结果会为13,不错任何事件处理。

当生产者1通过setAvailable(14)请求完成提交数据,如图所示。

BatchEventProcessorrun实现会处理1415位置上的数据,在下一次通过waitFor(16)获取可用的数据。

3、MutiProducerSequencer生产者类图。

MutiProducerSequencer继承AbstractSequencer,实现了Sequencer接口。

Sequencer提供增加删除消费者序列,创建SequenceBarrier,获取最小序号,和最大发布的序号。

Cursored获取当前的游标。

Sequenced获取当前ringbuffer大小,获取想一个序号,以及提交数据接口。

消费者和生产者之间的关联和单生产者一样,不做重复介绍。

4、多生产者通过next获取下一个可用的序号

public long next(int n) {
    if (n < 1) {
        throw new IllegalArgumentException("n must be > 0");
    }
    long current;
    long next;
    do {
        // ringbuffer当前生产者cursor
        current = cursor.get();
        // 下一个可用的序号
        next = current + n;
        // 重叠点位置
        long wrapPoint = next - bufferSize;
        // 缓存的消费者处理的序号
        long cachedGatingSequence = gatingSequenceCache.get();
        // wrapPoint > cachedGatingSequence,
        // 重叠位置大于缓存的消费者处理的序号,说明有消费者没有处理完成,不能够防止数据
        // cachedGatingSequence > nextValue
        // 只会在https://github.com/LMAX-Exchange/disruptor/issues/76情况下存在
        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current) {
            // 获取消费者和生产者最小的序号
            long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
            // 仍然重叠
            if (wrapPoint > gatingSequence) {
                // 通知消费者处理事件
                waitStrategy.signalAllWhenBlocking();
                // 生产者等待的时候后自旋,后续需要使用策略
                LockSupport.parkNanos(1);
                continue;
            }
            // 没有重叠的话,设置消费者缓存
            gatingSequenceCache.set(gatingSequence);
        }
        // 没有重叠,直接将RingBuffer的序号设置成next
        else if (cursor.compareAndSet(current, next)) {
            break;
        }
    }
    while (true);
    // 返回可用的序号
    return next;
}

5、多生产者通过publish提交数据

public void publish(final long sequence) {
    // 将sequence设置为可用状态
    setAvailable(sequence);
    // 通知消费者处理事件
    waitStrategy.signalAllWhenBlocking();
}

多生产者在获取序号next方法中就已经设置了cusor,提交数据的时候是将该sequence设置成可用状态,才能够被消费者使用。

6、消费者消费数据

再回忆下ProcessingSequenceBarrierwaitFor函数,其中调用到了sequencer.getHighestPublishedSequence(sequence,availableSequence);

public long waitFor(final long sequence)
        throws AlertException, InterruptedException, TimeoutException {
    // 检查clert异常
    checkAlert();
    // 通过waitStrategy策略获取可用的序号,cursorSequence为当前的Sequence,dependentSequence为依赖的Sequence[]
    long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
    // 产生比预期的sequence小,可能序号被重置回老的的oldSequence值
    //可参考https://github.com/LMAX-Exchange/disruptor/issues/76
    if (availableSequence < sequence) {
        return availableSequence;
    }
    // 获取最大的可用的已经发布的sequence,可能比sequence小
    // 会在多生产者中出现,当生产者1获取到序号13,生产者2获取到14;生产者1没发布,生产者2发布,会导致获取的可用序号为12,而sequence为13
    return sequencer.getHighestPublishedSequence(sequence, availableSequence);
}

获取最大的可用的已经发布的sequence

public long getHighestPublishedSequence(long lowerBound, long availableSequence) {
    for (long sequence = lowerBound; sequence <= availableSequence; sequence++) {
        // 判断是否可用
        if (!isAvailable(sequence)) {
            return sequence - 1;
        }
    }
    return availableSequence;
}

其中判断isAvailable通过availableBuffer进行判断

public boolean isAvailable(long sequence) {
    // 计算((int) sequence) & indexMask的索引index
    int index = calculateIndex(sequence);
    // 计算(int) (sequence >>> indexShift) ringbuffer的slot的设置次数
    int flag = calculateAvailabilityFlag(sequence);
    // index在数组中的偏移量
    long bufferAddress = (index * SCALE) + BASE;
    // 如果和flag相等,说明可用
    return UNSAFE.getIntVolatile(availableBuffer, bufferAddress) == flag;
}

内部使用的变量如下。

// availableBuffer跟踪每个ringbuffer的slot槽的状态,是否可用
private final int[] availableBuffer = new int[bufferSize]; // 初始值为-1
private final int indexMask = bufferSize - 1;
private final int indexShift = Util.log2(bufferSize);

通过以上方式就能够判断当前的sequence是否可用了。

通过在MutiProducerSequencergetHighestPublishedSequence方法中直接返回可用的availableSequence,通知消费者消费数据,生产者和消费者就协同起来了。