追蹤解析 Disruptor 原始碼

Disruptor · 發表 2019-02-14 14:33:58

摘要：零前期準備 0 FBI WARNING 文章異常囉嗦且繞彎。 1 版本 Disruptor 版本 : Disruptor 3.4.2 IDE : idea 2018.3 JDK 版本 : OpenJDK 11.0.1 2 Disruptor 簡介高效能執行緒...

零前期準備

0 FBI WARNING

文章異常囉嗦且繞彎。

1 版本

Disruptor 版本 : Disruptor 3.4.2

IDE : idea 2018.3

JDK 版本 : OpenJDK 11.0.1

2 Disruptor 簡介

高效能執行緒間訊息佇列框架 Disruptor，是金融與遊戲領域的常用開發元件之一，也是 java 日誌框架和流處理框架底層的常用依賴。

3 Demo

Disruptor 的 github 主頁有非常詳細的 quick start demo，本文依照此 demo 做追蹤的模板(做了很小的改動)。

另外，對於官方提供的 jdk8 lambda 簡化版 demo 暫不做討論。

import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import java.nio.ByteBuffer;

public class LongEventMain {

//main 方法，啟動入口
public static void main(String[] args) throws Exception {

//在該框架中，所有的 task 的包裝類被稱為 Event，EventFactory 則是 Event 的生產者
LongEventFactory factory = new LongEventFactory();

//RingBuffer 的大小，數字為位元組數
//RingBuffer 是框架啟動器內部的快取區，用來儲存 event 內的 task 資料
int bufferSize = 1024;

//建立一個 Disruptor 啟動器，其中 DaemonThreadFactory 是一個執行緒工廠的實現類
Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE);


//該框架本質上是 生產-消費 設計模式的應用。所有的消費者被冠名為 handler
//handleEventsWith(...) 方法會在啟動器中註冊 handler
//此處的引數是不定數量的，可以有多個消費者，每個消費者都可以獲取 Event
disruptor.handleEventsWith(new LongEventHandler("handler1"),new LongEventHandler("handler2"));

//啟動器開始執行，並獲取其內部的快取區
RingBuffer<LongEvent> ringBuffer = disruptor.start();

//建立一個生產者，負責往快取區內寫入資料
LongEventProducer producer = new LongEventProducer(ringBuffer);

//官方 demo 中使用了 ByteBuffer 來方便操作，其實非必須
ByteBuffer bb = ByteBuffer.allocate(8);

for (long l = 0; true; l++) {
//將變數 l 作為一個 long 型別的數存入 ByteBuffer 中
bb.putLong(0, l);
//將 ByteBuffer 傳入生產者的相關方法中，該方法會負責將 ByteBuffer 中的資料寫入 RingBuffer
producer.onData(bb);
//執行緒休眠
Thread.sleep(1000);
}
}
}

//Event 類，本質上是資料的封裝，是生產者和消費者之間進行資料傳遞的介質
class LongEvent {
private long value;

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

public long get() {
return value;
}
}

//Event 的生產工廠類，必須實現 Disruptor 自帶的 EventFactory 介面
class LongEventFactory implements EventFactory<LongEvent> {

@Override
public LongEvent newInstance() {
return new LongEvent();
}
}

//消費者，必須實現 Disruptor 自帶的 EventHandler 介面
class LongEventHandler implements EventHandler<LongEvent> {

private String handlerName;

public LongEventHandler(String handlerName){
this.handlerName = handlerName;
}

//此方法為最終的消費 Event 的方法
@Override
public void onEvent(LongEvent event, long sequence, boolean endOfBatch) {
System.out.println("Event " + handlerName + " : " + event.get());
}
}

//生產者，主要負責往 RingBuffer 中寫入資料
//生產者類在框架中並非必須，但是一般情況下都會做一定程度的封裝
class LongEventProducer {
private final RingBuffer<LongEvent> ringBuffer;

//生產者的構造器負責獲取並存儲啟動器中的 RingBuffer
public LongEventProducer(RingBuffer<LongEvent> ringBuffer) {
this.ringBuffer = ringBuffer;
}

public void onData(ByteBuffer bb) {
//sequence 是 RingBuffer 中的一個數據塊，類似於一個數據地址
long sequence = ringBuffer.next();
try {
//用資料地址去獲取到一個 Event 事件類例項
LongEvent event = ringBuffer.get(sequence);
//在例項中存入 ByteBuffer 中的資料
event.set(bb.getLong(0));
} finally {
//釋出該資料塊，此時消費者們都可以看到該資料塊了，可以進行消費
ringBuffer.publish(sequence);
}
}
}

一 DaemonThreadFactory

在開始正式追蹤程式碼之前有必要先來理解 DaemonThreadFactory。這是 Disruptor 自身攜帶的執行緒工廠類：

public enum DaemonThreadFactory implements ThreadFactory{

//執行緒工廠使用列舉實現單例模式
INSTANCE;

@Override
public Thread newThread(final Runnable r){
Thread t = new Thread(r);
//此處建立的執行緒是守護執行緒
t.setDaemon(true);
return t;
}
}

二 Disruptor

本 part 主要追蹤 demo 中 Disruptor 相關的程式碼：

Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE);
disruptor.handleEventsWith(new LongEventHandler("handler1"),new LongEventHandler("handler2"));
RingBuffer<LongEvent> ringBuffer = disruptor.start();

1 disruptor 的建立

來看下方程式碼：

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

追蹤 Disruptor 的構造器：

//step 1
//Disruptor.class
public Disruptor(final EventFactory<T> eventFactory, final int ringBufferSize, final ThreadFactory threadFactory){
//RingBuffer.createMultiProducer(...) 方法會返回一個 RingBuffer
//BasicExecutor 是執行緒和執行緒工廠的封裝類
this(RingBuffer.createMultiProducer(eventFactory, ringBufferSize), new BasicExecutor(threadFactory));
}

//step 2
//Disruptor.class
private Disruptor(final RingBuffer<T> ringBuffer, final Executor executor){
//存入 RingBuffer 和 Executor
this.ringBuffer = ringBuffer;
this.executor = executor;
}

但是實際上 Disruptor 提供了很多的構造器，其中還有一個較高配置許可權的：

//Disruptor.class
public Disruptor(final EventFactory<T> eventFactory,final int ringBufferSize,
final ThreadFactory threadFactory,final ProducerType producerType,
final WaitStrategy waitStrategy){
//解釋傳入的引數：
//eventFactory 是 Event 類的建立工廠
//ringBufferSize 是 RingBuffer 的位元組數大小
//threadFactory 是執行緒工廠
//ProducerType 是生產者的型別，分為單生產者型別(single)和多生產者型別(multi)，預設為 multi
//waitStrategy 是框架中的一個介面，表示等待策略，預設為 BlockingWaitStrategy(阻塞等待)，WaitStrategy 的可講內容較多，在後頭開一個單獨 part
this(RingBuffer.create(producerType, eventFactory, ringBufferSize, waitStrategy),new BasicExecutor(threadFactory));
}

先來看 ProducerType：

public enum ProducerType{
SINGLE,
MULTI
}

僅僅只是個標記，不多贅述。

1.1 BasicExecutor

BasicExecutor 是 Executor 的實現類，其內部維護著一個執行緒工廠和一個執行緒佇列，核心方法為 execute(...)：

//BasicExecutor.class
public void execute(Runnable command){

//使用執行緒工廠建立一個執行緒，此處的 factory 即為 DaemonThreadFactory
final Thread thread = factory.newThread(command);
//有效性驗證
if (null == thread){
throw new RuntimeException("Failed to create thread to run: " + command);
}

//開啟執行緒
thread.start();
//threads 是一個 ConcurrentLinkedQueue<Thread> 型別的變數，用來儲存執行緒
threads.add(thread);
}

1.2 RingBuffer 的建立

再來追蹤一下 RingBuffer 的建立：

//RingBuffer.class
public static <E> RingBuffer<E> create(ProducerType producerType,EventFactory<E> factory,
int bufferSize,WaitStrategy waitStrategy){
//此處根據 ProducerType 進行分發操作
switch (producerType){
case SINGLE:
//建立單消費者的 producer
return createSingleProducer(factory, bufferSize, waitStrategy);
case MULTI:
//建立多消費者的 producer
return createMultiProducer(factory, bufferSize, waitStrategy);
default:
//丟擲錯誤
throw new IllegalStateException(producerType.toString());
}
}

本質上這兩種模式的 RingBuffer 的建立差距並不大，此處追蹤 createMultiProducer(...) 方法：

//step 1
//RingBuffer.class
public static <E> RingBuffer<E> createMultiProducer(EventFactory<E> factory,int bufferSize,WaitStrategy waitStrategy){
//MultiProducerSequencer 是 RingBuffer 中用來在生產者和消費者之間傳遞資料的元件
//sequencer 是 RingBuffer 中的核心元件，是區別 SINGLE 和 MULTI 的關鍵，後文會繼續理解
MultiProducerSequencer sequencer = new MultiProducerSequencer(bufferSize, waitStrategy);
//自身構造器
return new RingBuffer<E>(factory, sequencer);
}

//step 2
//RingBuffer.class
RingBuffer(EventFactory<E> eventFactory,Sequencer sequencer){
//呼叫父類 RingBufferFields 的構造方法
super(eventFactory, sequencer);
}

//step 3
//RingBufferFields.class
RingBufferFields(EventFactory<E> eventFactory,Sequencer sequencer){
//此處為 MultiProducerSequencer
this.sequencer = sequencer;
//獲取使用者自定義的 bufferSize 並記錄下來
this.bufferSize = sequencer.getBufferSize();

//bufferSize 的有效性驗證
if (bufferSize < 1){
throw new IllegalArgumentException("bufferSize must not be less than 1");
}

if (Integer.bitCount(bufferSize) != 1){
throw new IllegalArgumentException("bufferSize must be a power of 2");
}

//根據 bufferSize 確定序列號最大值，因為從 0 開始所以要減一
this.indexMask = bufferSize - 1;
//entries 是一個 Object 陣列，用於存放 Event
//BUFFER_PAD 是對整個緩衝區的填充
this.entries = new Object[sequencer.getBufferSize() + 2 * BUFFER_PAD];
//fill(...) 方法會重新設定 entries
fill(eventFactory);
}

//step 4
//RingBuffer.class
private void fill(EventFactory<E> eventFactory){
for (int i = 0; i < bufferSize; i++){
//遍歷陣列進行 Event 的填充
entries[BUFFER_PAD + i] = eventFactory.newInstance();
}
}

2 消費者的註冊

來看下方程式碼：

disruptor.handleEventsWith(new LongEventHandler("handler1"),new LongEventHandler("handler2"));

追蹤 handleEventsWith(...) 方法：

//step 1
//Disruptor.class
public final EventHandlerGroup<T> handleEventsWith(final EventHandler<? super T>... handlers){
//Sequence 可以看做是 long 型的封裝類
//此處的第一個引數是前置關卡，在處理 handler 之前會進行處理的事件
//handlers 即為消費者
return createEventProcessors(new Sequence[0], handlers);
}

//step 2
//Disruptor.class
EventHandlerGroup<T> createEventProcessors(final Sequence[] barrierSequences,final EventHandler<? super T>[] eventHandlers){

//Disruptor 中有一個 AtomicBoolean 型別的變數 started
//checkNotStarted() 會檢查該變數的值是否為 true，如果是的話就證明已經啟動了，則丟擲異常
checkNotStarted();

//processorSequences 是每個消費者對應的執行器的序列號
final Sequence[] processorSequences = new Sequence[eventHandlers.length];

//此處返回的 barrier 可以看做是上文 MultiProducerSequencer 的封裝增強
final SequenceBarrier barrier = ringBuffer.newBarrier(barrierSequences);

for (int i = 0,eventHandlersLength = eventHandlers.length; i < eventHandlersLength; i++){
final EventHandler<? super T> eventHandler = eventHandlers[i];

//batchEventProcessor 是儲存了消費者和生產者的執行器，實現了 Runnable 介面，內部會不斷迴圈去接收並處理事件
final BatchEventProcessor<T> batchEventProcessor =
new BatchEventProcessor<>(ringBuffer, barrier, eventHandler);

//exceptionHandler 是用於處理錯誤的消費者元件
if (exceptionHandler != null){
batchEventProcessor.setExceptionHandler(exceptionHandler);
}

//consumerRepository 可以看做是消費者的集合封裝
//consumerRepository 會將傳入的三個引數包裝成 EventProcessorInfo 並儲存在集合和 map 裡
consumerRepository.add(batchEventProcessor, eventHandler, barrier);

//記錄下消費者對應的執行器的序列號
processorSequences[i] = batchEventProcessor.getSequence();
}

//處理一些前置事件，在本例中沒有前置事件存在
updateGatingSequencesForNextInChain(barrierSequences, processorSequences);

return new EventHandlerGroup<>(this, consumerRepository, processorSequences);
}

2.1 newBarrier

來追蹤一下 ringBuffer.newBarrier(...) 方法：

//step 1
//RingBuffer.class
public SequenceBarrier newBarrier(Sequence... sequencesToTrack){
//在本例中，此處的 sequencesToTrack 是 Sequence[0]
//此處的 sequencer 即為 MultiProducerSequencer
return sequencer.newBarrier(sequencesToTrack);
}

//step 2
//AbstractSequencer.class
public SequenceBarrier newBarrier(Sequence... sequencesToTrack){
//此方法被定義在 MultiProducerSequencer 的父類 AbstractSequencer 中
//cursor 是在 AbstractSequencer 中例項化的一個 Sequence 型別物件，是 MultiProducerSequencer 的序列號
return new ProcessingSequenceBarrier(this, waitStrategy, cursor, sequencesToTrack);
}

//step 3
//ProcessingSequenceBarrier.class
ProcessingSequenceBarrier(final Sequencer sequencer,final WaitStrategy waitStrategy,
final Sequence cursorSequence,final Sequence[] dependentSequences){
//即為 Disruptor 啟動器中的 MultiProducerSequencer
this.sequencer = sequencer;
//即為 Disruptor 啟動器中的阻塞策略
this.waitStrategy = waitStrategy;
//上述方法的 cursor
this.cursorSequence = cursorSequence;
if (0 == dependentSequences.length){
//此處的 dependentSequences 是長度是 0，所以此處
dependentSequence = cursorSequence;
}else{
dependentSequence = new FixedSequenceGroup(dependentSequences);
}
}

需要注意的是，此處的 sequencer 已經被抽象成了 SingleProducerSequencer 和 MultiProducerSequencer 的共同實現介面 Sequencer。

所以對於 SingleProducerSequencer 來說，這個流程也是沒有區別的。

2.2 updateGatingSequencesForNextInChain

回到上述程式碼：

//此處的 barrierSequences 是一個 Sequence[0] 陣列，processorSequences 是所有消費者的序列號集合
updateGatingSequencesForNextInChain(barrierSequences, processorSequences);

追蹤該方法的實現：

//step 1
//Disruptor.class
private void updateGatingSequencesForNextInChain(final Sequence[] barrierSequences, final Sequence[] processorSequences){

//processorSequences.length 大於 0 意味著消費者數量大於 0
if (processorSequences.length > 0){
ringBuffer.addGatingSequences(processorSequences);

//barrierSequences 是前置事件的集合
//由於此處的 barrierSequences 是長度為 0 的 Sequence 陣列，即沒有前置事件，所以此處不會進入迴圈，忽略
for (final Sequence barrierSequence : barrierSequences){
ringBuffer.removeGatingSequence(barrierSequence);
}
//unMarkEventProcessorsAsEndOfChain(...) 方法也是處理 barrierSequences 的，忽略
consumerRepository.unMarkEventProcessorsAsEndOfChain(barrierSequences);
}
}

addGatingSequences

追蹤 ringBuffer.addGatingSequences(...) 方法：

//step 1
//RingBuffer.class
public void addGatingSequences(Sequence... gatingSequences){
//sequencer 為 MultiProducerSequencer
sequencer.addGatingSequences(gatingSequences);
}

//step 2
//AbstractSequencer.class
public final void addGatingSequences(Sequence... gatingSequences){
//此處的 SEQUENCE_UPDATER 是一個 AtomicReferenceFieldUpdaterImpl 型別的變數，用於 CAS 操作 gatingSequences
SequenceGroups.addSequences(this, SEQUENCE_UPDATER, this, gatingSequences);
}

//step 3
//SequenceGroups.class
static <T> void addSequences(final T holder,final AtomicReferenceFieldUpdater<T, Sequence[]> updater,
final Cursored cursor,final Sequence... sequencesToAdd){
long cursorSequence;
Sequence[] updatedSequences;
Sequence[] currentSequences;

do{
//此處的 holder 即為 MultiProducerSequencer，此處獲取其內部的 gatingSequences 變數
currentSequences = updater.get(holder);
//此處為 copyOf(...) 方法為 java.util.Arrays.copyOf(...) 方法，用於將 currentSequences 複製一份
updatedSequences = copyOf(currentSequences, currentSequences.length + sequencesToAdd.length);
//此處的 cursor 即為 MultiProducerSequencer，getCursor() 方法會獲取其的序列號
cursorSequence = cursor.getCursor();

int index = currentSequences.length;
//此處的 sequencesToAdd 是之前消費者的序列號集合，更新 sequencesToAdd 中的每個序列號封裝
//將 MultiProducerSequencer 的序列號註冊進去，並填充到新集合的後面一半中
for (Sequence sequence : sequencesToAdd){
sequence.set(cursorSequence);
updatedSequences[index++] = sequence;
}
}while (!updater.compareAndSet(holder, currentSequences, updatedSequences));
//此處的 while 會死迴圈 CAS 操作直到更新成功

//在此獲取 MultiProducerSequencer 的序列號，更新到 sequencesToAdd 的每個序列號封裝類中
cursorSequence = cursor.getCursor();
for (Sequence sequence : sequencesToAdd){
sequence.set(cursorSequence);
}
}

3 Disruptor 的啟動

來看下方程式碼：

disruptor.start();

追蹤 start(...) 方法：

//Disruptor.class
public RingBuffer<T> start(){
//確認該 Disruptor 沒有啟動
checkOnlyStartedOnce();
//此處的 consumerInfo 是 EventProcessorInfo 型別的變數
for (final ConsumerInfo consumerInfo : consumerRepository){
consumerInfo.start(executor);
}

return ringBuffer;
}

先來看 checkOnlyStartedOnce() 方法：

//Disruptor.class
private void checkOnlyStartedOnce(){
//如果在呼叫該 CAS 方法之前已經為 true 了，會丟擲錯誤
//其實就是確保在呼叫該方法之前還處於未開啟的狀態
if (!started.compareAndSet(false, true)){
throw new IllegalStateException("Disruptor.start() must only be called once.");
}
}

再來追蹤 EventProcessorInfo 的 start(...) 方法：

//EventProcessorInfo.class
public void start(final Executor executor){
//此處的 executor 即為 BasicExecutor
executor.execute(eventprocessor);
}

所以本質上 Disruptor 的啟動就是開啟 BasicExecutor，藉此啟動執行緒。

3.1 BatchEventProcessor

上述程式碼中啟動執行緒的時候會傳入 eventprocessor 物件作為 task 去啟動消費者。eventprocessor 物件本質上是上文中提到過的 BatchEventProcessor。

BatchEventProcessor 能夠被傳入 execute(...) 方法，證明其實現了 Runnable 介面：

//step 1
//BatchEventProcessor.class
@Override
public void run(){
//running 是一個定義在 BatchEventProcessor 中的 AtomicInteger 型別的變數
//CAS 操作，先判斷 running 的值是否等於 IDLE，如果是的話就修改成 RUNNING，並返回 true
//IDLE = 1，RUNNING = 2，皆為 int 型別的常量
if (running.compareAndSet(IDLE, RUNNING)){

//此處修改 sequenceBarrier 中 alert 變數的狀態值，清除掉中斷狀態
sequenceBarrier.clearAlert();

//如果傳入的消費者實現了 LifecycleAware 介面，就會在 notifyStart() 方法中去執行相關方法
//LifecycleAware 中定義了 onStart() 和 onShutdown() 方法，會分別在消費者真正執行之前和關閉之前執行一次
//執行 LifecycleAware 的 onStart() 方法
notifyStart();
try{
//如果 running 是 RUNNING 狀態，就會進入死迴圈
if (running.get() == RUNNING){
//核心方法
processEvents();
}
}finally{
//執行 LifecycleAware 的 onShutdown() 方法
notifyShutdown();
//切換 running 的狀態值
running.set(IDLE);
}
}else{
if (running.get() == RUNNING){
throw new IllegalStateException("Thread is already running");
}else{
earlyExit();
}
}
}

//step 2
//BatchEventProcessor.class
private void processEvents(){
T event = null;
//此處的 sequence 記錄著當前消費者已經處理過的事件的編號，初始化的時候為 -1，所以 nextSequence 初始為 0，每次加一
//nextSequence 是當前消費者下一項準備處理的事件的編號
long nextSequence = sequence.get() + 1L;

//死迴圈
while (true){
try{
//當沒有事件發生的時候，消費者所在的執行緒會在此等待，具體的實現依照使用者設定的等待策略的不同而不同
//本例中使用的是 BlockingWaitStrategy，所以會在此阻塞直到出現了事件
//返回的 availableSequence 是最新的事件的編號，在任務量較小的情況下和 nextSequence 數值相同，在任務量較大的情況下小於 nextSequence
//等待策略留在後頭展開
final long availableSequence = sequenceBarrier.waitFor(nextSequence);

if (batchStartAware != null){
batchStartAware.onBatchStart(availableSequence - nextSequence + 1);
}

//nextSequence 大於 availableSequence 的情況理論上不會出現
while (nextSequence <= availableSequence){
//dataProvider 就是之前初始化的 RingBuffer，RingBuffer 在此處會去獲取當前編號的 Event
event = dataProvider.get(nextSequence);
//onEvent(...) 是 EventHandler 介面定義的方法，是消費者消費 Event 的最重要方法，方法體由使用者進行定義
eventHandler.onEvent(event, nextSequence, nextSequence == availableSequence);
//編號自增
nextSequence++;
}

//在消費完當前的所有事件之後，記錄下事件編號
sequence.set(availableSequence);
}catch (final TimeoutException e){
//如果消費者實現了 TimeoutHandler 介面，就可以在這裡處理超時問題
notifyTimeout(sequence.get());
}catch (final AlertException ex){
//running 的狀態值非 RUNNING，就會退出死迴圈
if (running.get() != RUNNING){
break;
}
}catch (final Throwable ex){
//如果當前的消費者實現了 ExceptionHandler 介面的話，可以在此處進行錯誤處理
exceptionHandler.handleEventException(ex, nextSequence, event);
sequence.set(nextSequence);
nextSequence++;
}
}
}

3.2 WaitStrategy

回到上述程式碼的以下這句：

final long availableSequence = sequenceBarrier.waitFor(nextSequence);

追蹤一下 waitFor(...) 方法：

//ProcessingSequenceBarrier.class
public long waitFor(final long sequence) throws AlertException, InterruptedException, TimeoutException{

//如果變數 alert 為 true 的話會丟擲錯誤
checkAlert();

//呼叫等待策略的相關方法
//返回最新的事件的編號
long availableSequence = waitStrategy.waitFor(sequence, cursorSequence, dependentSequence, this);

//如果當前可用的最新事件編號小於傳入的 sequence，就直接返回可用編號即可
if (availableSequence < sequence){
return availableSequence;
}

//getHighestPublishedSequence(...) 方法會判斷最大的可用的事件編號
return sequencer.getHighestPublishedSequence(sequence, availableSequence);
}

等待策略的所有實現類都實現了 WaitStrategy 介面：

public interface WaitStrategy{
//休眠方法
long waitFor(long sequence, Sequence cursor, Sequence dependentSequence, SequenceBarrier barrier)
throws AlertException, InterruptedException, TimeoutException;
//喚醒方法
void signalAllWhenBlocking();
}

Disruptor 自帶的策略中，常用的有以下幾種：

阻塞策略BlockingWaitStrategy：預設策略，沒有獲取到任務的情況下執行緒會進入等待狀態。cpu 消耗少，但是延遲高。
阻塞限時策略TimeoutBlockingWaitStrategy：相對於BlockingWaitStrategy來說，設定了等待時間，超過後拋異常。
自旋策略BusySpinWaitStrategy：執行緒一直自旋等待。cpu 佔用高，延遲低.
Yield 策略 YieldingWaitStrategy：嘗試自旋 100 次，然後呼叫 Thread.yield() 讓出 cpu。cpu 佔用高，延遲低。
分段策略SleepingWaitStrategy：嘗試自旋 100 此，然後呼叫 Thread.yield() 100 次，如果經過這兩百次的操作還未獲取到任務，就會嘗試階段性掛起自身執行緒。此種方式是對 cpu 佔用和延遲的一種平衡，效能不太穩定。

還有幾種譬如 PhasedBackoffWaitStrategy 和 LiteBlockingWaitStrategy 等，不多介紹。

詳細看一下 BlockingWaitStrategy 的實現：

public final class BlockingWaitStrategy implements WaitStrategy{
//重入鎖
private final Lock lock = new ReentrantLock();
//Condition 用來控制執行緒的休眠和喚醒
private final Condition processorNotifyCondition = lock.newCondition();

@Override
public long waitFor(long sequence, Sequence cursorSequence, 
Sequence dependentSequence, SequenceBarrier barrier)
throws AlertException, InterruptedException{

long availableSequence;
if (cursorSequence.get() < sequence){
//上鎖
lock.lock();
try{
while (cursorSequence.get() < sequence){
//檢查執行緒是否中斷了，如果已經中斷了就會丟擲異常
barrier.checkAlert();
//休眠執行緒
processorNotifyCondition.await();
}
}finally{
//解鎖
lock.unlock();
}
}

//生產者進度小於消費者的消費進度，此迴圈進行等待
//正常情況下都會在上方阻塞，不會進入該迴圈
while ((availableSequence = dependentSequence.get()) < sequence){
barrier.checkAlert();
ThreadHints.onSpinWait();
}

return availableSequence;
}

@Override
public void signalAllWhenBlocking(){
lock.lock();
try{
//用 Condition 喚醒全部的執行緒
processorNotifyCondition.signalAll();
}finally{
lock.unlock();
}
}

//toString() 方法，忽略
@Override
public String toString(){
return "BlockingWaitStrategy{" +
"processorNotifyCondition=" + processorNotifyCondition +
'}';
}
}

3.3 DataProvider

回到上述程式碼的以下這句：

event = dataProvider.get(nextSequence);

dataProvider 是一個 DataProvider 型別的變數。DataProvider 本質上是一個 Disruptor 內的介面：

public interface DataProvider<T>{
T get(long sequence);
}

其存在唯一實現類 RingBuffer。所以 get(...) 方法也在 RingBuffer 中：

//step 1
//RingBuffer.class
@Override
public E get(long sequence){
//elementAt(...) 方法定義在 RingBuffer 的抽象父類 RingBufferFields 中
return elementAt(sequence);
}

//step 2
//RingBufferFields.class
protected final E elementAt(long sequence){
//呼叫 UNSAFE 的相關方法，通過地址去直接獲取
//entries 在上文程式碼中申請了一系列地址連續的記憶體
//REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT) 是一個很巧妙的演算法，結果永遠只會在申請下來的記憶體中迴圈
return (E) UNSAFE.getObject(entries, REF_ARRAY_BASE + ((sequence & indexMask) << REF_ELEMENT_SHIFT));
}

由此可知，Disruptor 中的所有的事件都非儲存在虛擬機器中，而是儲存在虛擬機器外，由 Unsafe 類直接呼叫。

Unsafe 具有 "呼叫記憶體物件很快，但是申請記憶體塊很慢" 的特性，所以也就可以解釋為什麼在初始化的時候要一次性將儲存 Event 的陣列進行逐個初始化了(程式碼在上述 1.2 小節的 step 4 中)。

有一個注意點，entries 上的元素實際上是在 jvm 管轄範圍內的，並不一定需要使用 Unsafe 去呼叫，這裡只是為了更高的效能。

三 Event 的產生

在開頭的 demo 中，可以看到 LongEventProducer 中有一個核心方法：

//LongEventProducer.class
public void onData(ByteBuffer bb) {
//sequence 是 RingBuffer 中的一個數據塊，類似於一個數據地址
long sequence = ringBuffer.next();
try {
//用資料地址去獲取到一個 Event 事件類例項
LongEvent event = ringBuffer.get(sequence);
//在例項中存入 ByteBuffer 中的資料
event.set(bb.getLong(0));
} finally {
//釋出該資料塊，此時消費者們都可以看到該資料塊了，可以進行消費
ringBuffer.publish(sequence);
}
}

這個方法內通過呼叫 ringBuffer.next() 方法獲取陣列內物件的地址，然後通過 ringBuffer.get(...) 方法獲取物件。

在 finally 程式碼塊中呼叫 ringBuffer.publish(...) 方法去釋出該資訊。

1 next

回到上述程式碼的以下這句：

long sequence = ringBuffer.next();

追蹤 next() 方法：

//step 1
//RingBuffer.class
@Override
public long next(){
//呼叫 RingBuffer 內的 MultiProducerSequencer 的相關方法
return sequencer.next();
}

//step 2
//MultiProducerSequencer.class
@Override
public long next(){
//呼叫自身的相關方法
return next(1);
}

//step 3
//MultiProducerSequencer.class
@Override
public long next(int n){
//引數有效性驗證，此處 n = 1
if (n < 1){
throw new IllegalArgumentException("n must be > 0");
}

long current;
long next;

//死迴圈
do{
//current 是當前最新的事件編號
current = cursor.get();
//此處為 current + 1，用作下一個事件的編號
next = current + n;

//wrapPoint 是事件編號和陣列大小的差
long wrapPoint = next - bufferSize;

//gatingSequenceCache 的設計很巧妙，它是一個 Sequence 型別的變數，可以看做是一個 long 整數
//gatingSequenceCache 的存在意義是每隔一段時間去檢查一次消費者的處理進度
//gatingSequenceCache 在每次檢查進度的時候都會更新成 "當前處理最慢的消費者已經處理完成的事件編號"
//處理邏輯在下方 if 判斷中
long cachedGatingSequence = gatingSequenceCache.get();

//cachedGatingSequence > current 的情況就不會發生，因為 cachedGatingSequence 是消費者處理進度，current 是目前的事件總編號，所以最多相等
//在消費者算力充足的情況下，cachedGatingSequence 會和 current 相等
//wrapPoint > cachedGatingSequence 的情況，在極端情況下可能是因為生產者的速度太快了，已經遠超過最慢的那個消費者，超過了 "一圈"(即 bufferSize 的大小)

//此處可以這麼理解，由於 RingBuffer 內陣列的大小是有限的，如果事件生產的多了，就會覆蓋掉最開始的幾個事件
//但是如果消費者的進度沒有跟上，來不及消費就被覆蓋了，就造成了 bug，此處即為抑制策略
if (wrapPoint > cachedGatingSequence || cachedGatingSequence > current){

//getMinimumSequence(...) 方法會獲取當前處理事件最慢的那個消費者的處理位置
long gatingSequence = Util.getMinimumSequence(gatingSequences, current);

//wrapPoint - gatingSequence = next - bufferSize - gatingSequence >0
//即 next > bufferSize + gatingSequence，落後了 "一圈"
if (wrapPoint > gatingSequence){
//執行緒掛起 1 納秒，然後跳過本次迴圈進行下一次迴圈
//此處會陷入死迴圈，阻塞掉生產者，去等待消費者的進度
LockSupport.parkNanos(1);
continue;
}

//跳出上述迴圈之後在這裡更新 gatingSequenceCache 的值
gatingSequenceCache.set(gatingSequence);
}else if (cursor.compareAndSet(current, next)){
//如果消費者的進度正常，那麼會在此用 CAS 操作更新 cursor 的值，並且跳出 while 迴圈
break;
}
}while (true);

//返回
return next;
}

線上程池(比如筆者比較瞭解的 ThreadPoolExecutor)的實現中，對於 task 過多，溢位等待佇列的情況，一般會有一種策略去應對。在 ThreadPoolExecutor 中，預設的策略為丟擲錯誤，直接終止程式。

在 Disruptor 中，其實 RingBuffer 就類似一個等待佇列，溢位策略則是暫停 task 的產生，等待執行緒池去執行。

【此處僅為類比，不能簡單的把 Disruptor 想成是一個執行緒池】

2 publish

ringBuffer.publish(...) 是事件釋出的核心方法：

//step 1
//RingBuffer.class
@Override
public void publish(long sequence){
sequencer.publish(sequence);
}

//step 2
//MultiProducerSequencer.class
@Override
public void publish(final long sequence){
//此處更新資料
setAvailable(sequence);
//此處呼叫等待策略的 signalAllWhenBlocking() 方法喚醒所有等待的執行緒
//具體實現依照 waitStrategy 的不同而不同
waitStrategy.signalAllWhenBlocking();
}

//step 3
//MultiProducerSequencer.class
private void setAvailable(final long sequence){
//calculateAvailabilityFlag(sequence) 可以簡單理解為是計算出的圈數，即 (sequence / bufferSize)
//calculateIndex(sequence) 會計算出新的 sequence 對應組中的哪一個位置，即 (sequence % bufferSize)
setAvailableBufferValue(calculateIndex(sequence), calculateAvailabilityFlag(sequence));
}

//step 4
//MultiProducerSequencer.class
private void setAvailableBufferValue(int index, int flag){
//SCALE 是本機 Object[] 引用的大小，一般為 4
long bufferAddress = (index * SCALE) + BASE;
//使用 Unsafe 更新元素
//availableBuffer 是一個 int 陣列，大小為 bufferSize，即和 entries 相同
//Unsafe.putOrderedInt(...) 會將 availableBuffer 的指定位置(bufferAddress)的元素修改成 flag
UNSAFE.putOrderedInt(availableBuffer, bufferAddress, flag);
}

四其它補充

筆者在專案過程中還看到了 Disruptor 的另一種用法，即和執行緒池一樣，每個 Event 只被一個消費者取用一次，然後就會失效。

具體的改造是將消費者繼承的 EventHandler 替換成 WorkHandler：

class LongEventHandler2 implements WorkHandler<LongEvent>{

private String handlerName;

public LongEventHandler2(String handlerName){
this.handlerName = handlerName;
}

@Override
public void onEvent(LongEvent event) throws Exception {
System.out.println("Event " + handlerName + " : " + event.get());
}
}

然後在 Disruptor 啟動器註冊消費者的時候不再使用 handleEventsWith(...) 方法，而是改用 handleEventsWithWorkerPool(...) 方法：

LongEventHandler2[] workers = new LongEventHandler2[5];
for (int i = 0; i < workers.length; i++) {
workers[i] = new LongEventHandler2("worker " + i);
}

//此方法會註冊執行緒池類的消費者
disruptor.handleEventsWithWorkerPool(workers);

具體原始碼不做追蹤了，筆者在專案中做過嘗試，挺好用的。

五一點嘮叨

· Disruptor 的封裝很薄(比起 Netty、Spring 之類的重量級框架)，呼叫鏈路都相對較短

· Disruptor 的環裝快取區(RingBuffer)的很多概念還有待理解

· 對於筆者這樣的數學苦手來說看底層演算法程式碼略頭疼

· 僅為個人的學習筆記，可能存在錯誤或者表述不清的地方，有緣補充

追蹤解析 Disruptor 原始碼

零 前期準備

0 FBI WARNING

1 版本

2 Disruptor 簡介

3 Demo

一 DaemonThreadFactory

二 Disruptor

1 disruptor 的建立

1.1 BasicExecutor

1.2 RingBuffer 的建立

2 消費者的註冊

2.1 newBarrier

2.2 updateGatingSequencesForNextInChain

addGatingSequences

3 Disruptor 的啟動

3.1 BatchEventProcessor

3.2 WaitStrategy

3.3 DataProvider

三 Event 的產生

1 next

2 publish

四 其它補充

五 一點嘮叨

您可能也會喜歡…

零前期準備

四其它補充

五一點嘮叨