大家好，我是君哥。

Disruptor 是一個很受歡迎的內存消息隊列，它源于 LMAX 對并發(fā)、性能和非阻塞算法的研究。今天一起來學習一下這個消息隊列。

簡介

對于主流的分布式消息隊列來說，一般會包含 Producer、Broker、Consumer、注冊中心等模塊。比如 RocketMQ 架構如下：

圖片

Disruptor 并不是分布式消息隊列，它是一款內存消息隊列，因此架構上跟分布式消息隊列有很大差別。下面是一張 LMAX 使用 Disruptor 的案例圖：

圖片

我們介紹一下 Disruptor 架構中的核心概念。

1.1 Ring Buffer

Ring Buffer 通常被認為是 Disruptor 的最主要的設計，但是從 3.0 版本開始，Ring Buffer 只負責存儲和更新經過 Disruptor 的數據。在一些高級的使用場景，它甚至完全可以被用戶替換。

1.2 Sequence

Disruptor 使用 Sequence 來識別特定組件的位置。每個 Consumer（也就是事件處理器）都像 Disruptor 一樣持有一個 Sequence。并發(fā)相關的核心代碼依賴 Sequence 的自增值，因此 Sequence 具有跟 AtomicLong 相似的特性，事實上唯一的不同就是不同的 Sequence 之間不存在偽共享問題。

偽共享：CPU 緩存是以緩存行為單位進行加載和存儲，CPU 每次從主存中拉取數據時，會把相鄰的數據也存入同一個緩存行。即使多個線程操作的是同一緩存行中不同的變量，只要有一個線程修改了緩存行中的某一個變量值，該緩存行就會被標記為無效，需要重新從主從中加載。在多線程環(huán)境下，頻繁地重新加載緩存行，會嚴重影響了程序執(zhí)行效率。

1.3 Sequencer

Sequencer 是 Disrupter 的真正核心，有單個生產者和多個生產者兩種實現（SingleProducerSequencer 和 MultiProducerSequencer）。為了讓數據在生產者和消費者之間快速、準確地傳輸，它們都實現了所有并發(fā)算法。

1.4 Sequence Barrier

Sequencer 生成一個 Sequence Barrier，它包含由 Sequencer 生成的 Sequence 和消費者擁有的 Sequence 的引用。Sequence Barrier 決定是否有事件給消費者處理。

1.5 Wait Strategy

消費者怎樣等待事件的到來。

1.6 Event Processor

主要負責循環(huán)處理來自 Disruptor 事件，它擁有消費者 Sequence 的所有權。有一個單獨的實現類 BatchEventProcessor，這個類擁有高效的事件循環(huán)處理能力并且處理完成后可以回調實現 EventHandler 接口的用戶。

1.7 Event Handler

由用戶來實現并且代表 Disruptor 消費者的接口。

2 Disruptor 特性

2.1 多播事件

多播事件是 Disruptor 區(qū)別于其他隊列的最大差異。其他隊列都是一個事件消息只能被單個消費者消費，而 Disruptor 如果有多個消費者監(jiān)聽，則可以將所有事件消息發(fā)送給所有消費者。

在前面 LMAX 使用 Disruptor 的案例圖中，有 JournalConsumer、ReplicationConsumer 和 ApplicationConsumer 三個消費者監(jiān)聽了 Disruptor，這三個消費者將收到來了 Disruptor 的所有消息。

2.2 消費者依賴關系圖

為了支持并發(fā)處理在實際業(yè)務場景中的需要，有時消費者直接需要做協(xié)調。再回到前面 LMAX 使用 Disruptor 的案例，在 journalling 和 replication 這兩個消費者處理完成之前，有必要阻止業(yè)務邏輯消費者開始處理。我們稱這個特征為“gating”（或者更準確地說，該特征是“gating”的一種形式）。

首先，確保生產者數量不會超過消費者。這通過調用 RingBuffer.addGatingConsumers（）來將相關消費者添加到 Disruptor。其次，消費者依賴關系的實現是通過構建一個 SequenceBarrier，SequenceBarrier 擁有需要在它前面完成處理邏輯的消費者的 Sequence。

就拿前面 LMAX 使用 Disruptor 的案例來說，ApplicationConsumer 的 SequenceBarrier 擁有 JournalConsumer 和 ReplicationConsumer 這 2 個消費者的 Sequence，所以 ApplicationConsumer 對 JournalConsumer 和 ReplicationConsumer 的依賴關系可以從 SequenceBarrier 到 Sequence 的連接中看到。

Sequencer 和下游消費者的關系也需要注意。Sequencer 的一個角色就是發(fā)布的事件消息不能超出 Ring Buffer。這就要求下游消費者的 Sequence 不能小于 Ring Buffer 的 Sequence，也不能小于 Ring Buffer 的大小。

上面圖中，因為 ApplicationConsumer 的 Sequence 必須要保證小于等于 JournalConsumer 和 ReplicationConsumer 的 Sequence，因此 Sequencer 只需要關心 ApplicationConsumer 的 Sequence。

2.3 內存預分配

Disruptor 的目標是低延遲，因此減少或者去除內存分配是必要的。在基于 Java 的系統(tǒng)中，目標是減少 STW 次數。

為了支持這一點，用戶可以在 Disruptor 中預分配事件所需的內存。在預分配內存時，用戶提供的 EventFactory 將對 Ring Buffer 的所有元素進行調用。當生產者向 Disruptor 發(fā)送新的事件消息時，Disruptor 的 API 允許用戶使用構造好的對象，他們可以調用對象的方法或者更新對象的字段。Disruptor 需要確保并發(fā)安全。

2.4 無鎖并發(fā)

Disruptor 實現低延遲的另一個關鍵方法時使用無鎖算法，通過使用內存屏障和 CAS 來實現內存可見性和正確性。Disruptor 唯一使用鎖的地方就是在 BlockingWaitStrategy。

3 調優(yōu)選項

雖然大多數場景下 Disruptor 可以表現出優(yōu)秀的性能，但是仍然有一些調優(yōu)參數可以改進 Disruptor 的性能。

3.1 單個/多個生產者

Disruptor<LongEvent> disruptor = new Disruptor(
 factory,
 bufferSize,
 DaemonThreadFactory.INSTANCE,
 ProducerType.SINGLE, 
 new BlockingWaitStrategy() 
);

上面是 disruptor 的構造函數，ProducerType.SINGLE 表示創(chuàng)建單生產者的 Sequencer，ProducerType.MULTI  表示創(chuàng)建多生產者的 Sequencer。

在并發(fā)系統(tǒng)中提高系統(tǒng)性能的最好方式是遵循單寫原則。下面是官方的一個 disruptor 吞吐量測試結果，測試環(huán)境是 i7 Sandy Bridge MacBook Air。

單生產者：

圖片

多生產者：

圖片

3.2 等待策略

• BlockingWaitStrategy

disruptor 的默認等待策略是 BlockingWaitStrategy，這種策略使用鎖和喚醒鎖的 Condition 變量。

• SleepingWaitStrategy

跟 BlockingWaitStrategy 策略類似，他是通過 LockSupport.parkNanos(1) 方法來實現等待，不需要給 Condition 變量發(fā)送信號來喚醒等待。

主要適用于對延時要求不高的場景，比如異步打印日志。

• YieldingWaitStrategy

YieldingWaitStrategy 策略使用 Busy spin（不釋放 CPU 資源，通過循環(huán)檢查條件直到條件滿足為止）技術來等待 sequence 增長到一個合適的值。在循環(huán)內部會調用 Thread#yield() 方法允許其他排隊線程去執(zhí)行。

這種策略主要用于通過消耗 CPU 來實現低延遲的場景。當 EventHandler 數量消息邏輯 CPU 核數并且對延遲要求較高時，可以考慮這種等待策略。

• BusySpinWaitStrategy

BusySpinWaitStrategy 是性能最高的等待策略，它適用于低延遲系統(tǒng)，但是對部署環(huán)境要求很高。

這種等待策略的唯一適用場景是當 EventHandler 數量消息邏輯 CPU 核數并且超線程被禁用。

4 官方示例

下面是一個官方示例。這個例子比較簡單，就是生產者向消費者發(fā)送一個 long 類型的值。

• 首先定義一個 Event。

public class LongEvent
{
    private long value;

    public void set(long value)
    {
        this.value = value;
    }

    @Override
    public String toString()
    {
        return "LongEvent{" + "value=" + value + '}';
    }
}

• 為了能讓 Disruptor 預分配內存，這里定義一個 LongEventFactory。

public class LongEventFactory implements EventFactory<LongEvent>
{
    @Override
    public LongEvent newInstance()
    {
        return new LongEvent();
    }
}

• 創(chuàng)建一個消費者來處理事件

public class LongEventHandler implements EventHandler<LongEvent>
{
    @Override
    public void onEvent(LongEvent event, long sequence, boolean endOfBatch)
    {
        System.out.println("Event: " + event);
    }
}

• 編寫發(fā)送事件消息的邏輯

import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.examples.longevent.LongEvent;
import com.lmax.disruptor.util.DaemonThreadFactory;
import java.nio.ByteBuffer;

public class LongEventMain
{
    public static void main(String[] args) throws Exception
    {
        int bufferSize = 1024; 

        Disruptor<LongEvent> disruptor = 
                new Disruptor<>(LongEvent::new, bufferSize, DaemonThreadFactory.INSTANCE);

        disruptor.handleEventsWith((event, sequence, endOfBatch) ->
                System.out.println("Event: " + event)); 
        disruptor.start(); 


        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer(); 
        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; true; l++)
        {
            bb.putLong(0, l);
            ringBuffer.publishEvent((event, sequence, buffer) -> event.set(buffer.getLong(0)), bb);
            Thread.sleep(1000);
        }
    }
}

5 總結

作為一款高性能的內存隊列，Disruptor 有不少優(yōu)秀的設計思想值得我們學習，比如內存預分配、無鎖并發(fā)。同時它的使用非常簡單，推薦大家使用。