java Disruptor構(gòu)建高性能內(nèi)存隊列使用詳解

更新時間：2022年12月09日 10:37:27 作者：碼猿技術(shù)專欄

這篇文章主要為大家介紹了java Disruptor構(gòu)建高性能內(nèi)存隊列使用詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

Java中有哪些隊列

ArrayBlockingQueue 使用ReentrantLock
LinkedBlockingQueue 使用ReentrantLock
ConcurrentLinkedQueue 使用CAS

等等

我們清楚使用鎖的性能比較低，盡量使用無鎖設計。接下來就我們來認識下

Disruptor簡單使用

github地址：github.com/LMAX-Exchan…

先簡單介紹下：

Disruptor它是一個開源的并發(fā)框架，并獲得2011 Duke’s程序框架創(chuàng)新獎【Oracle】，能夠在無鎖的情況下實現(xiàn)網(wǎng)絡的Queue并發(fā)操作。英國外匯交易公司LMAX開發(fā)的一個高性能隊列，號稱單線程能支撐每秒600萬訂單~
日志框架Log4j2 異步模式采用了Disruptor來處理
局限呢，他就是個內(nèi)存隊列，也就是說無法支撐分布式場景。

簡單使用

數(shù)據(jù)傳輸對象

@Data
public class EventData {
    private Long value;
}

消費者

public class EventConsumer implements WorkHandler<EventData> {
    /**
     * 消費回調(diào)
     * @param eventData
     * @throws Exception
     */
    @Override
    public void onEvent(EventData eventData) throws Exception {
        Thread.sleep(5000);
        System.out.println(Thread.currentThread() + ", eventData:" + eventData.getValue());
    }
}

生產(chǎn)者

public class EventProducer {
    private final RingBuffer<EventData> ringBuffer;
    public EventProducer(RingBuffer<EventData> ringBuffer) {
        this.ringBuffer = ringBuffer;
    }
    public void sendData(Long v){
        // cas展位
        long next = ringBuffer.next();
        try {
            EventData eventData = ringBuffer.get(next);
            eventData.setValue(v);
        } finally {
            // 通知等待的消費者
            System.out.println("EventProducer send success, sequence:"+next);
            ringBuffer.publish(next);
        }
    }
}

測試類

public class DisruptorTest {
    public static void main(String[] args) {
        // 2的n次方
        int bufferSize = 8;
        Disruptor<EventData> disruptor = new Disruptor<EventData>(
                () -> new EventData(), // 事件工廠
                bufferSize,            // 環(huán)形數(shù)組大小
                Executors.defaultThreadFactory(),       // 線程池工廠
                ProducerType.MULTI,    // 支持多事件發(fā)布者
                new BlockingWaitStrategy());    // 等待策略
        // 設置消費者
        disruptor.handleEventsWithWorkerPool(
                new EventConsumer(),
                new EventConsumer(),
                new EventConsumer(),
                new EventConsumer());
        disruptor.start();
        RingBuffer<EventData> ringBuffer = disruptor.getRingBuffer();
        EventProducer eventProducer = new EventProducer(ringBuffer);
        long i  = 0;
        for(;;){
            i++;
            eventProducer.sendData(i);
            try {
                Thread.sleep(1500);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

核心組件

基于上面簡單例子來看確實很簡單，Disruptor幫我們封裝好了生產(chǎn)消費模型的實現(xiàn)，接下來我們來看下他是基于哪些核心組件來支撐起一個高性能無鎖隊列呢？

RingBuffer： 環(huán)形數(shù)組，底層使用數(shù)組entries，在初始化時填充數(shù)組，避免不斷新建對象帶來的開銷。后續(xù)只會對entries做更新操作

Sequencer： 核心管家

定義生產(chǎn)同步的實現(xiàn)：SingleProducerSequencer單生產(chǎn)、MultiProducerSequencer多生產(chǎn)
當前寫的進度Sequence cursor
所有消費者進度的數(shù)組Sequence[] gatingSequences
MultiProducerSequencer可用區(qū)availableBuffer【利用空間換取查詢效率】

Sequence： 本身就是一個序號器用來標識處理進度，也可以當做是一個atomicInteger; 還有另外一個特點，為了解決偽共享問題而引入的：緩存行填充。這個在后面介紹。

workProcessor: 處理Event的循環(huán)，在循環(huán)中獲取Disruptor的事件，然后把事件分配給各個handler

EventHandler: 負責業(yè)務邏輯的handler,自己實現(xiàn)。

WaitStrategy： 消費者如何等待事件的策略，定義了如下策略

leepingWaitStrategy：自旋 + yield + sleep

BlockingWaitStrategy：加鎖，適合CPU資源緊張（不需要切換線程），系統(tǒng)吞吐量無要求的

YieldingWaitStrategy：自旋 + yield + 自旋

BusySpinWaitStrategy：自旋，減少線程之前切換

PhasedBackoffWaitStrategy：自旋 + yield + 自定義策略

帶著問題來解析代碼？

1、多生產(chǎn)者如何保證消息生產(chǎn)不會相互覆蓋。【如何達到互斥效果】

每個線程獲取不同的一段數(shù)組空間，然后通過CAS判斷這段空間是否已經(jīng)分配出去。

接下來我們看下多生產(chǎn)類MultiProducerSequencer中next方法【獲取生產(chǎn)序號】

// 消費者上一次消費的最小序號 // 后續(xù)第二點會講到
private final Sequence gatingSequenceCache = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 當前進度的序號
protected final Sequence cursor = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
// 所有消費者的序號 //后續(xù)第二點會講到
protected volatile Sequence[] gatingSequences = new Sequence[0];
 public long next(int n)
    {
        if (n < 1)
        {
            throw new IllegalArgumentException("n must be > 0");
        }
        long current;
        long next;
        do
        {
            // 當前進度的序號，Sequence的value具有可見性，保證多線程間線程之間能感知到可申請的最新值
            current = cursor.get();
            // 要申請的序號空間:最大序列號
            next = current + n;
            long wrapPoint = next - bufferSize;
            // 消費者最小序列號
            long cachedGatingSequence = gatingSequenceCache.get();
            // 大于一圈 || 最小消費序列號>當前進度
            if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current)
            {
                long gatingSequence = Util.getMinimumSequence(gatingSequences, current);
                // 說明大于1圈，并沒有多余空間可以申請
                if (wrapPoint > gatingSequence)
                {
                    LockSupport.parkNanos(1); // TODO, should we spin based on the wait strategy?
                    continue;
                }
                // 更新最小值到Sequence的value中
                gatingSequenceCache.set(gatingSequence);
            }
            // CAS成功后更新當前Sequence的value
            else if (cursor.compareAndSet(current, next))
            {
                break;
            }
        }
        while (true);
        return next;
    }

2、生產(chǎn)者向序號器申請寫的序號，如序號正在被消費，Sequencer是如何知道哪些序號是可以被寫入的呢？【未消費則被覆蓋如何處理】

從gatingSequences中取得最小的序號，生產(chǎn)者最多能寫到這個序號的后一位。通俗來講就是申請的序號不能大于最小消費者序號一圈【申請到最大序列號-buffersize 要小于/等于最小消費的序列號】的時候，才能申請到當前寫的序號

public final EventHandlerGroup<T> handleEventsWithWorkerPool(final WorkHandler<T>... workHandlers)
{
    return createWorkerPool(new Sequence[0], workHandlers);
}
EventHandlerGroup<T> createWorkerPool(
    final Sequence[] barrierSequences, final WorkHandler<? super T>[] workHandlers)
{
    final SequenceBarrier sequenceBarrier = ringBuffer.newBarrier(barrierSequences);
    final WorkerPool<T> workerPool = new WorkerPool<>(ringBuffer, sequenceBarrier, exceptionHandler, workHandlers);
    consumerRepository.add(workerPool, sequenceBarrier);
    final Sequence[] workerSequences = workerPool.getWorkerSequences();
    updateGatingSequencesForNextInChain(barrierSequences, workerSequences);
    return new EventHandlerGroup<>(this, consumerRepository, workerSequences);
}
    private void updateGatingSequencesForNextInChain(final Sequence[] barrierSequences, final Sequence[] processorSequences)
{
    if (processorSequences.length > 0)
    {
        // 消費者啟動后就會將所有消費者存放入AbstractSequencer中g(shù)atingSequences
        ringBuffer.addGatingSequences(processorSequences);
        for (final Sequence barrierSequence : barrierSequences)
        {
            ringBuffer.removeGatingSequence(barrierSequence);
        }
        consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
    }
}

3、在多生產(chǎn)者情況下，生產(chǎn)者是申請到一段可寫入的序號，然后再寫入這些序號中，那么消費者是如何感知哪些序號是可以被消費的呢？【借問提1圖說明】

這個前提是多生產(chǎn)者情況下，第一點我們說過每個線程獲取不同的一段數(shù)組空間，那么現(xiàn)在單單通過序號已經(jīng)不夠用了，MultiProducerSequencer使用了int 數(shù)組【availableBuffer】來標識當前序號是否可用。當生產(chǎn)者成功生產(chǎn)事件后會將availableBuffer中當前序列號置為1標識可以讀取。

如此消費者可以讀取的的最大序號就是我們availableBuffer中第一個不可用序號-1。

初始化availableBuffer流程

public MultiProducerSequencer(int bufferSize, final WaitStrategy waitStrategy)
{
    super(bufferSize, waitStrategy);
    // 初始化可用數(shù)組
    availableBuffer = new int[bufferSize];
    indexMask = bufferSize - 1;
    indexShift = Util.log2(bufferSize);
    initialiseAvailableBuffer();
}
// 初始化默認availableBuffer為-1
private void initialiseAvailableBuffer()
{
    for (int i = availableBuffer.length - 1; i != 0; i--)
    {
        setAvailableBufferValue(i, -1);
    }
    setAvailableBufferValue(0, -1);
}
// 生產(chǎn)者成功生產(chǎn)事件將可用區(qū)數(shù)組置為1
public void publish(final long sequence)
{
    setAvailable(sequence);
    waitStrategy.signalAllWhenBlocking();
}
private void setAvailableBufferValue(int index, int flag)
{
    long bufferAddress = (index * SCALE) + BASE;
    UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
}

消費者消費流程

WorkProcessor類中消費run方法
public void run()
    {
        boolean processedSequence = true;
        long cachedAvailableSequence = Long.MIN_VALUE;
        long nextSequence = sequence.get();
        T event = null;
        while (true)
        {
            try
            {
                // 先通過cas獲取消費事件的占有權(quán)
                if (processedSequence)
                {
                    processedSequence = false;
                    do
                    {
                        nextSequence = workSequence.get() + 1L;
                        sequence.set(nextSequence - 1L);
                    }
                    while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
                }
                // 數(shù)據(jù)就緒，可以消費
                if (cachedAvailableSequence >= nextSequence)
                {
                    event = ringBuffer.get(nextSequence);
                    // 觸發(fā)回調(diào)函數(shù)
                    workHandler.onEvent(event);
                    processedSequence = true;
                }
                else
                {
                    // 獲取可以被讀取的下標
                    cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
                }
            }
        // ....省略
        }
        notifyShutdown();
        running.set(false);
    }
    public long waitFor(final long sequence)
        throws AlertException, InterruptedException, TimeoutException
    {
        checkAlert();
        // 這個值獲取的current write 下標，可以認為全局消費下標。此處與每一段的write1和write2下標區(qū)分開
        long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);
        if (availableSequence < sequence)
        {
            return availableSequence;
        }
        // 通過availableBuffer篩選出第一個不可用序號 -1
        return sequencer.getHighestPublishedSequence(sequence, availableSequence);
    }
    public long getHighestPublishedSequence(long lowerBound, long availableSequence)
    {
        // 從current read下標開始， 循環(huán)至 current write,如果碰到availableBuffer 為-1 直接返回
        for (long sequence = lowerBound; sequence <= availableSequence; sequence++)
        {
            if (!isAvailable(sequence))
            {
                return sequence - 1;
            }
        }
        return availableSequence;
    }

解決偽共享問題

什么是偽共享問題呢？

為了提高CPU的速度，Cpu有高速緩存Cache,該緩存最小單位為緩存行CacheLine,他是從主內(nèi)存復制的Cache的最小單位，通常是64字節(jié)。一個Java的long類型是8字節(jié)，因此在一個緩存行中可以存8個long類型的變量。如果你訪問一個long數(shù)組，當數(shù)組中的一個值被加載到緩存中，它會額外加載另外7個。因此你能非?？斓乇闅v這個數(shù)組。

偽共享問題是指，當多個線程共享某份數(shù)據(jù)時，線程1可能拉到線程2的數(shù)據(jù)在其cache line中，此時線程1修改數(shù)據(jù)，線程2取其數(shù)據(jù)時就要重新從內(nèi)存中拉取，兩個線程互相影響，導致數(shù)據(jù)雖然在cache line中，每次卻要去內(nèi)存中拉取。