從零開始實現lmax-Disruptor佇列（三）多執行緒消費者WorkerPool原理解析
2022-06-18 06:02:27
MyDisruptor V3版本介紹
在v2版本的MyDisruptor實現多消費者、消費者組間依賴功能後。按照計劃，v3版本的MyDisruptor需要支援多執行緒消費者的功能。
由於該文屬於系列部落格的一部分，需要先對之前的部落格內容有所瞭解才能更好地理解本篇部落格
- v1版本部落格：從零開始實現lmax-Disruptor佇列（一）RingBuffer與單生產者、單消費者工作原理解析
- v2版本部落格：從零開始實現lmax-Disruptor佇列（二）多消費者、消費者組間消費依賴原理解析
MyDisruptor支援多執行緒消費者
- 之前的版本中我們已經實現了單執行緒消費者序列的消費，但在某些場景下我們需要更快的消費速度，所以disruptor也提供了多執行緒的消費者機制。
- 多執行緒消費者對外功能上和單執行緒消費者基本一樣，也是全量的消費從序列0到序列N的完整事件，但內部卻是區域性並行亂序消費的。在一定的範圍內，具體哪個執行緒消費哪個事件是通過CAS爭搶隨機獲得的。
WorkerPool實現解析
- disruptor中多執行緒消費者的載體是WorkerPool。
- 在V3版本的MyDisruptor中，MyWorkerPool和單執行緒消費者MyBatchEventProcessor一樣，建構函式都是傳入三個關鍵元件：RingBuffer、序列屏障mySequenceBarrier和使用者自定義的事件處理器。
- 和單執行緒消費者不同，多執行緒消費者允許傳入一個使用者自定義的事件處理器MyWorkHandler集合。傳入的每個MyWorkHandler都會建立一個MyWorkProcessor物件將其封裝、包裹起來（下文會展開介紹MyWorkProcessor）。
- 雖然同為使用者自定義的消費處理器介面，disruptor中WorkHandler和單執行緒消費者中傳入的EventHandler有些不一樣。其消費處理介面只傳入了事件物件本身，並沒有sequence和endOfBatch引數。
主要原因是因為多執行緒消費者內部的消費是並行、亂序的，因此sequence序列號意義不大，且endOfBatch也無法準確定義。 - WorkerPool對外提供了一個用於啟動消費者的方法start，要求外部傳入一個juc下的Executor實現用於啟動所有的MyWorkProcessor任務。
/**
* 多執行緒消費者（仿Disruptor.WorkerPool）
* */
public class MyWorkerPool<T> {
private final MySequence workSequence = new MySequence(-1);
private final MyRingBuffer<T> myRingBuffer;
private final List<MyWorkProcessor<T>> workEventProcessorList;
public MyWorkerPool(
MyRingBuffer<T> myRingBuffer,
MySequenceBarrier mySequenceBarrier,
MyWorkHandler<T>... myWorkHandlerList) {
this.myRingBuffer = myRingBuffer;
final int numWorkers = myWorkHandlerList.length;
this.workEventProcessorList = new ArrayList<>(numWorkers);
// 為每個自定義事件消費邏輯MyEventHandler，建立一個對應的MyWorkProcessor去處理
for (MyWorkHandler<T> myEventConsumer : myWorkHandlerList) {
workEventProcessorList.add(new MyWorkProcessor<>(
myRingBuffer,
myEventConsumer,
mySequenceBarrier,
this.workSequence));
}
}
/**
* 返回包括每個workerEventProcessor + workerPool自身的序列號集合
* */
public MySequence[] getCurrentWorkerSequences() {
final MySequence[] sequences = new MySequence[this.workEventProcessorList.size() + 1];
for (int i = 0, size = workEventProcessorList.size(); i < size; i++) {
sequences[i] = workEventProcessorList.get(i).getCurrentConsumeSequence();
}
sequences[sequences.length - 1] = workSequence;
return sequences;
}
public MyRingBuffer<T> start(final Executor executor) {
final long cursor = myRingBuffer.getCurrentProducerSequence().get();
workSequence.set(cursor);
for (MyWorkProcessor<?> processor : workEventProcessorList) {
processor.getCurrentConsumeSequence().set(cursor);
executor.execute(processor);
}
return this.myRingBuffer;
}
}
/**
* 多執行緒消費者-事件處理器介面
* */
public interface MyWorkHandler<T> {
/**
* 消費者消費事件
* @param event 事件物件本身
* */
void consume(T event);
}
MyWorkProcessor實現解析
接下來是本篇部落格的重點部分，MyWorkProcessor的實現。
/**
* 多執行緒消費者工作執行緒 （仿Disruptor.WorkProcessor）
* */
public class MyWorkProcessor<T> implements Runnable{
private final MySequence currentConsumeSequence = new MySequence(-1);
private final MyRingBuffer<T> myRingBuffer;
private final MyWorkHandler<T> myWorkHandler;
private final MySequenceBarrier sequenceBarrier;
private final MySequence workGroupSequence;
public MyWorkProcessor(MyRingBuffer<T> myRingBuffer,
MyWorkHandler<T> myWorkHandler,
MySequenceBarrier sequenceBarrier,
MySequence workGroupSequence) {
this.myRingBuffer = myRingBuffer;
this.myWorkHandler = myWorkHandler;
this.sequenceBarrier = sequenceBarrier;
this.workGroupSequence = workGroupSequence;
}
public MySequence getCurrentConsumeSequence() {
return currentConsumeSequence;
}
@Override
public void run() {
long nextConsumerIndex = this.currentConsumeSequence.get() + 1;
// 設定哨兵值，保證第一次迴圈時nextConsumerIndex <= cachedAvailableSequence一定為false，走else分支通過序列屏障獲得最大的可用序列號
long cachedAvailableSequence = Long.MIN_VALUE;
// 最近是否處理過了序列
boolean processedSequence = true;
while (true) {
try {
if(processedSequence) {
// 爭搶到了一個新的待消費序列，但還未實際進行消費（標記為false）
processedSequence = false;
// 如果已經處理過序列，則重新cas的爭搶一個新的待消費序列
do {
nextConsumerIndex = this.workGroupSequence.get() + 1L;
// 由於currentConsumeSequence會被註冊到生產者側，因此需要始終和workGroupSequence worker組的實際sequence保持協調
// 即當前worker的消費序列currentConsumeSequence = 當前消費者組的序列workGroupSequence
this.currentConsumeSequence.lazySet(nextConsumerIndex - 1L);
// 問題：只使用workGroupSequence，每個worker不維護currentConsumeSequence行不行？
// 回答：這是不行的。因為和單執行緒消費者的行為一樣，都是具體的消費者eventHandler/workHandler執行過之後才更新消費者的序列號，令其對外部可見（生產者、下游消費者）
// 因為消費依賴關係中約定，對於序列i事件只有在上游的消費者消費過後（eventHandler/workHandler執行過），下游才能消費序列i的事件
// workGroupSequence主要是用於通過cas協調同一workerPool內消費者執行緒序列爭搶的，對外的約束依然需要workProcessor原生的消費者序列currentConsumeSequence來控制
// cas更新，保證每個worker執行緒都會獲取到唯一的一個sequence
} while (!workGroupSequence.compareAndSet(nextConsumerIndex - 1L, nextConsumerIndex));
}else{
// processedSequence == false(手頭上存在一個還未消費的序列)
// 走到這裡說明之前拿到了一個新的消費序列，但是由於nextConsumerIndex > cachedAvailableSequence，沒有實際執行消費邏輯
// 而是被阻塞後返回獲得了最新的cachedAvailableSequence，重新執行一次迴圈走到了這裡
// 需要先把手頭上的這個序列給消費掉，才能繼續拿下一個消費序列
}
// cachedAvailableSequence只會存在兩種情況
// 1 第一次迴圈，初始化為Long.MIN_VALUE，則必定會走到下面的else分支中
// 2 非第一次迴圈，則cachedAvailableSequence為序列屏障所允許的最大可消費序列
if (nextConsumerIndex <= cachedAvailableSequence) {
// 爭搶到的消費序列是滿足要求的（小於序列屏障值，被序列屏障允許的），則呼叫消費者進行實際的消費
// 取出可以消費的下標對應的事件，交給eventConsumer消費
T event = myRingBuffer.get(nextConsumerIndex);
this.myWorkHandler.consume(event);
// 實際呼叫消費者進行消費了，標記為true.這樣一來就可以在下次迴圈中cas爭搶下一個新的消費序列了
processedSequence = true;
} else {
// 1 第一次迴圈會獲取當前序列屏障的最大可消費序列
// 2 非第一次迴圈，說明爭搶到的序列超過了屏障序列的最大值，等待生產者推進到爭搶到的sequence
cachedAvailableSequence = sequenceBarrier.getAvailableConsumeSequence(nextConsumerIndex);
}
} catch (final Throwable ex) {
// 消費者消費時發生了異常，也認為是成功消費了，避免阻塞消費序列
// 下次迴圈會cas爭搶一個新的消費序列
processedSequence = true;
}
}
}
}
- WorkProcessor和單執行緒消費者BatchEventProcessor有不少相似之處：
- 也實現了Runnable介面，作為一個獨立的執行緒通過一個主迴圈不斷的監聽上游進度而進行工作。
- 也通過建構函式傳入的sequenceBarrier來控制消費速度。
- WorkProcessor是隸屬於特定WorkerPool的，一個WorkerPool下的所有WorkProcessor執行緒都通過WorkerPool的序列號物件workSequence來協調爭搶序列。
- WorkerPool內的每個MyWorkProcessor執行緒都可以通過CAS爭搶到一個消費者內獨一無二的序列號，保證不會出現多執行緒間的重複消費
（假設一個WorkPool有三個WorkProcessor執行緒A、B、C，如果workerA執行緒爭搶到了序列號1，則workerB、workerC執行緒就不會再處理序列號1，而是去爭搶序列號2了）
雖然disruptor一直在努力避免使用CAS，但多執行緒消費者並行爭搶序列號的場景下也沒有特別好的辦法了。 - CAS爭搶到了一個序列號後，如果當前序列號可用（小於或等於序列屏障給出的當前最大可消費序列號），則會呼叫使用者自定義消費邏輯myWorkHandler進行業務處理。
如果當前序列號不可用,則會被阻塞於序列屏障的getAvailableConsumeSequence方法中。 - WorkerPool通過getCurrentWorkerSequences對外暴露workerSequence和每個worker執行緒的本地消費序列合併在一起的集合。
因此CAS爭搶時，通過this.currentConsumeSequence.lazySet(nextConsumerIndex - 1L)，保證當前workerProcessor的消費序列不會落後WorkerPool的序列太多
只使用workGroupSequence，每個MyWorkProcessor不單獨維護currentConsumeSequence行不行？
- 這是不行的。因為和單執行緒消費者的行為一樣，都是具體的消費者eventHandler/workHandler執行過之後才更新消費者的序列號，令其對外部可見（生產者、下游消費者）。
消費依賴關係中約定，對於序列i事件只有在上游的消費者消費過後（eventHandler/workHandler執行過），下游才能消費序列i的事件。 - 如果只使用workGroupSequence，則cas爭搶成功後（但具體的消費者還未消費），其對應的消費序列便立即對外部可見了，這是不符合上述約定的。
- workGroupSequence主要是用於通過cas協調同一workerPool內消費者執行緒序列爭搶，對外的約束依然需要每個workProcessor內部的消費者序列currentConsumeSequence來控制。
MyDisruptor v3版本demo解析
v3版本支援了多執行緒消費者功能，下面通過一個demo來展示如何使用該功能。
public class MyRingBufferV3Demo {
/**
* -> 多執行緒消費者B（依賴A）
* 單執行緒消費者A -> 單執行緒消費者D（依賴B、C）
* -> 單執行緒消費者C（依賴A）
* */
public static void main(String[] args) throws InterruptedException {
// 環形佇列容量為16（2的4次方）
int ringBufferSize = 16;
// 建立環形佇列
MyRingBuffer<OrderEventModel> myRingBuffer = MyRingBuffer.createSingleProducer(
new OrderEventProducer(), ringBufferSize, new MyBlockingWaitStrategy());
// 獲得ringBuffer的序列屏障（最上游的序列屏障內只維護生產者的序列）
MySequenceBarrier mySequenceBarrier = myRingBuffer.newBarrier();
// ================================== 基於生產者序列屏障，建立消費者A
MyBatchEventProcessor<OrderEventModel> eventProcessorA =
new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerA"), mySequenceBarrier);
MySequence consumeSequenceA = eventProcessorA.getCurrentConsumeSequence();
// RingBuffer監聽消費者A的序列
myRingBuffer.addGatingConsumerSequenceList(consumeSequenceA);
// ================================== 消費者組依賴上游的消費者A，通過消費者A的序列號建立序列屏障（構成消費的順序依賴）
MySequenceBarrier workerSequenceBarrier = myRingBuffer.newBarrier(consumeSequenceA);
// 基於序列屏障，建立多執行緒消費者B
MyWorkerPool<OrderEventModel> workerPoolProcessorB =
new MyWorkerPool<>(myRingBuffer, workerSequenceBarrier,
new OrderWorkHandlerDemo("workerHandler1"),
new OrderWorkHandlerDemo("workerHandler2"),
new OrderWorkHandlerDemo("workerHandler3"));
MySequence[] workerSequences = workerPoolProcessorB.getCurrentWorkerSequences();
// RingBuffer監聽消費者C的序列
myRingBuffer.addGatingConsumerSequenceList(workerSequences);
// ================================== 通過消費者A的序列號建立序列屏障（構成消費的順序依賴），建立消費者C
MySequenceBarrier mySequenceBarrierC = myRingBuffer.newBarrier(consumeSequenceA);
MyBatchEventProcessor<OrderEventModel> eventProcessorC =
new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerC"), mySequenceBarrierC);
MySequence consumeSequenceC = eventProcessorC.getCurrentConsumeSequence();
// RingBuffer監聽消費者C的序列
myRingBuffer.addGatingConsumerSequenceList(consumeSequenceC);
// ================================== 基於多執行緒消費者B，單執行緒消費者C的序列屏障，建立消費者D
MySequence[] bAndCSequenceArr = new MySequence[workerSequences.length+1];
// 把多執行緒消費者B的序列複製到合併的序列陣列中
System.arraycopy(workerSequences, 0, bAndCSequenceArr, 0, workerSequences.length);
// 陣列的最後一位是消費者C的序列
bAndCSequenceArr[bAndCSequenceArr.length-1] = consumeSequenceC;
MySequenceBarrier mySequenceBarrierD = myRingBuffer.newBarrier(bAndCSequenceArr);
MyBatchEventProcessor<OrderEventModel> eventProcessorD =
new MyBatchEventProcessor<>(myRingBuffer, new OrderEventHandlerDemo("consumerD"), mySequenceBarrierD);
MySequence consumeSequenceD = eventProcessorD.getCurrentConsumeSequence();
// RingBuffer監聽消費者D的序列
myRingBuffer.addGatingConsumerSequenceList(consumeSequenceD);
// 啟動消費者執行緒A
new Thread(eventProcessorA).start();
// 啟動workerPool多執行緒消費者B
workerPoolProcessorB.start(Executors.newFixedThreadPool(100, new ThreadFactory() {
private final AtomicInteger mCount = new AtomicInteger(1);
@Override
public Thread newThread(Runnable r) {
return new Thread(r,"worker" + mCount.getAndIncrement());
}
}));
// 啟動消費者執行緒C
new Thread(eventProcessorC).start();
// 啟動消費者執行緒D
new Thread(eventProcessorD).start();
// 生產者釋出100個事件
for(int i=0; i<100; i++) {
long nextIndex = myRingBuffer.next();
OrderEventModel orderEvent = myRingBuffer.get(nextIndex);
orderEvent.setMessage("message-"+i);
orderEvent.setPrice(i * 10);
System.out.println("生產者釋出事件：" + orderEvent);
myRingBuffer.publish(nextIndex);
}
}
}
- WorkPool對外直接面向用戶，而WorkProcessor則被封裝隱藏起來，對使用者不可見。
- WorkPool作為一個消費者有著自己的消費序列，其也是通過往ringBuffer的生產者中註冊消費序列限制生產速度，讓下游消費者維護消費序列實現上下游依賴關係的。
但WorkPool是多執行緒的，其消費序列是一個包含WorkPool總序列號和各個子執行緒內序列號的集合。因此在上述場景中，需要將這個集合（陣列）視為一個整體維護起來。
總結
- 在v3版本中我們實現了多執行緒的消費者。多執行緒消費者中，每個worker執行緒通過cas排它的爭搶序列號，因此多執行緒消費者WorkPool可以在同一時刻並行的同時消費多個事件。
在消費邏輯較耗時的場景下，可以考慮使用disruptor的多執行緒消費者來加速消費。 - 由於java中執行緒排程主要由作業系統控制，在某一執行緒cas爭搶到序列號n後但還未實際呼叫workerHandler的使用者自定義消費介面前，可能會被作業系統暫時掛起，此時爭搶到更大序列號n+1的其它執行緒則可以先呼叫workerHandler的使用者自定義消費介面。
帶來的直接影響就是多執行緒消費者內部的消費是區域性亂序的（可能n+1序列的事件先消費、n序列的事件後消費）。
disruptor無論在整體設計還是最終程式碼實現上都有很多值得反覆琢磨和學習的細節，希望能幫助到對disruptor感興趣的小夥伴。
本篇部落格的完整程式碼在我的github上：https://github.com/1399852153/MyDisruptor 分支：feature/lab3