【雜談】Disruptor——RingBuffer問題整理（一）

  • 2020 年 5 月 10 日
  • 筆記

純CAS為啥比加鎖要快？

同樣是修改數據，一個採用加鎖的方式保證原子性，一個採用CAS的方式保證原子性。

都是能夠達到目的的，但是常用的鎖（例如顯式的Lock和隱式的synchonized），都會把獲取不到鎖的執行緒掛起，相對於CAS的不掛起，多了掛起和喚醒的開銷。

尾指針是如何管理的，如何防止覆蓋舊數據？

別的帖子都說RingBuffer中不維護尾指針，尾指針由消費者維護（所謂維護指針，就是修改、移動指針）其實這一句話有點誤導性，如果RingBuffer不知道尾部在哪裡，那它的數據存儲肯定就會出問題，例如把還沒消費過的數據給覆蓋了。

確實，消費者會自行維護自己的消費指針，RingBuffer也不會去干涉消費者指針的維護，但是它會引用所有消費者的指針，讀取他們的值，以此作為「尾部」的判斷依據。實際上就是最慢的那個消費者為準。

註：消費者指針是消費者消費過的最後一條數據的序號

我們直接來看程式碼，這個是RingBuffer的publishEvent方法，我們看到，它首先取得一個可用的序列號，然後再將數據放入該序列號的對應位置中。我們來看看這個序列號是如何取得的。

@Override
    public void publishEvent(EventTranslator<E> translator)
    {
        final long sequence = sequencer.next();
        translateAndPublish(translator, sequence);
    }

我們先看Sequencer的SingleProducerSequencer實現。這裡就是判斷如果生產者新指針的位置是否會超過尾部，如果超過尾部就掛起等待。注意這裡的等待方式也是自旋方式，只不過，每次失敗後都會自行掛起片刻。

這裡附上幾個圖可能更好理解：（畫圖工具不太好，無法通過單元格上色的方式體現空閑情況）

情況1：隊列已滿，生產者嘗試使用新序號14，但由於（14 - 8 = 6），由於最慢的消費者目前消費的最後一條數據的序號是5，5號之後的數據還沒被消費，6 > 5，所以序號14還不能用。生產者執行緒掛起，下次再次嘗試。

 

 

 情況2：消費者1消費了序號6的數據。（14 - 8 = 6） 不大於 6，這時序號14可用，生產者得到可用的序號。

 

 

    @Override
    public long next()
    {
        return next(1);
    }

    /**
     * @see Sequencer#next(int)
     */
    @Override
    public long next(int n)
    {
        if (n < 1 || n > bufferSize)
        {
            throw new IllegalArgumentException("n must be > 0 and < bufferSize");
        }

        long nextValue = this.nextValue; //當前RingBuffer的游標，即生產者的位置指針

        long nextSequence = nextValue + n; 
        long wrapPoint = nextSequence - bufferSize; //減掉一圈
        long cachedGatingSequence = this.cachedValue; //上一次快取的最小的消費者指針

        //條件1：生產者指針的位置超過當前消費最小的指針
        //條件2：為特殊情況，這裡先不考慮，詳見：
        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue)
        {
            cursor.setVolatile(nextValue);  // StoreLoad fence

            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;
    }

另外對於多生產者的情況，在不會越界的情況下，需要通過CAS來保證獲取序號的原子性。具體可以查看MultiProducerSequencer的next方法。

 消費者指針是如何讀取的？

RingBuffer如何知道有哪些消費者？哪些gatingSequense是從哪裡來的？

在構建RingBuffer註冊處理類的時候，就將消費者Sequense註冊到RingBuffer中了。

看程式碼的話，定位到gatingSequences在AbastractSequencer，對應的有個addGatingSequenses方法用於注入gatingSequence

public abstract class AbstractSequencer implements Sequencer {
    //...
    protected volatile Sequence[] gatingSequences = new Sequence[0];

    @Override
    public final void addGatingSequences(Sequence... gatingSequences)
    {
        SequenceGroups.addSequences(this, SEQUENCE_UPDATER, this, gatingSequences);
    }

    //...
}

再查看addGatingSequences被調用的地方，即通過RingBuffer的方法，設置到Sequencer中，這個Sequence是生產者使用的序號管理器

public final class RingBuffer<E> extends RingBufferFields<E> implements Cursored, EventSequencer<E>, EventSink<E> {
    //...
    protected final Sequencer sequencer;
    
    public void addGatingSequences(Sequence... gatingSequences) {
        sequencer.addGatingSequences(gatingSequences);
    }
    //...
}

而RingBuffer的addGatingSequence則在Disruptor配置處理器的時候被調用

public class Disruptor<T> {
    //...
    private final RingBuffer<T> ringBuffer;
    private final ConsumerRepository<T> consumerRepository = new ConsumerRepository<>();
    
    
    public EventHandlerGroup<T> handleEventsWith(final EventProcessor... processors)
    {
        for (final EventProcessor processor : processors)
        {
            consumerRepository.add(processor);
        }

        final Sequence[] sequences = new Sequence[processors.length];
        for (int i = 0; i < processors.length; i++)
        {
            sequences[i] = processors[i].getSequence();
        }

        ringBuffer.addGatingSequences(sequences);

        return new EventHandlerGroup<>(this, consumerRepository, Util.getSequencesFor(processors));
    }
    //...
}

快取的意義是什麼？

我們看到在SiingleProducerSequencer的next方法中，會快取上一次的消費者最小序列號，這有什麼用呢？

用途就是不需要每次都讀取各消費者的序號，只要沒超過上一次的最小值的地方都可以直接分配，如果超過了，則進行再次判斷

為啥讀取最小值不需要保證原子性？

看了這個獲取最小消費序號的，可能會奇怪，為啥這個操作不需要上鎖，這個不是會獲取到舊值嗎？

確實，這個最小值獲取到的時候，實際上數值已經變更。但是由於我們的目的是為了防止指針越位，所以用舊值是沒有問題的。（舊值<=實際上的最小值）

public static long getMinimumSequence(final Sequence[] sequences, long minimum)
    {
        for (int i = 0, n = sequences.length; i < n; i++)
        {
            long value = sequences[i].get();
            minimum = Math.min(minimum, value);
        }

        return minimum;
    }

 

Tags:

VirMach 便宜 VPS