在美團呆了7年的架構師帶你解讀Disruptor系列并發框架-有解無憂

前言

理解Disruptor的最佳方式是，將其與一些容易理解和目的相似的東西比較，這里的參照物就是java里的阻塞佇列（BlockingQueue），

與BlockingQueue的異同：
同：目的相同，都是為了在同一行程的執行緒間傳輸資料，
異：對消費者多播事件；預分配事件記憶體；可選無鎖，

核心概念

Ring Buffer : 曾經的核心，自從3.0以上，環形緩沖器只作為Disruptor存盤和更新資料（事件）的容器，對于一些高級用法，可以完全替換為用戶提供的容器，
Sequence：Disruptor使用Sequence作為一種確定特定組件位置的方法，每個消費者（EventProcessor）都維護一個Sequence，Disruptor自己也是一樣，大部分并發代碼以來這些Sequence值的移動，因此Sequence支持AtomicLong的當前許多特性，事實上，兩者唯一的區別是Sequence包含了附加功能來防止Sequence和其他值的偽共享，
Sequencer：Disruptor的真正核心，此介面的兩個實作（單生產者和多生產者）都實作了用于在生產者和消費者間快速準確傳遞資料的并發演算法，
Sequence Barrier：由Sequencer產生，持有Sequencer的主要發布Sequence和任意獨立消費者的Sequence的索引，它包含判斷是否有可供消費者處理的可用事件的邏輯，
Wait Strategy：等待策略決定了一個消費者如何等待生產者發布到Disruptor的事件，
Event：生產者傳遞給消費者的資料單元，用戶自定義，
EventProcessor：處理Disruptor事件的主要回圈，擁有消費者的Sequence，有一個BatchEventProcessor包含了一個事件回圈的高效實作，會在事件可用時回呼叫戶提供的EventHandler介面實作，
EventHandler：用戶實作介面，代表Disruptor的一個消費者，
Producer：用戶呼叫Disruptor進行入隊的代碼，在框架中沒有代碼表示，

多播事件

這是queue和Disruptor最大的行為區別，佇列中的一個事件只能被一個消費者消費，而Disruptor中的時間會發布給所有消費者，這是由于Disruptor意圖處理同一資料的獨立并行處理操作（譯注：類似JMS的topic模式），比如LMAX中同一資料需要進行記錄日志、復制和業務邏輯操作，當然，在Disruptor中同時并行處理不同事件可以使用WorkerPool（譯注：類似JMS的queue模式中的多消費者實作），但需要注意的是，由于這種特性并非是Disruptor的首要作業，所以使用WorkerPool可能并不是最高效的做法，
查看上圖，三個消費者JournalConsumer、ReplicationConsumer和ApplicationConsumer將會以相同順序接收Disruptor所有可用訊息，這實作了這些消費者的并行作業，

消費者依賴圖

為了支持并發處理的真實世界應用，很有必要支持消費者間的協調作業，回顧上圖，在日志記錄和復制消費者完成作業前，有必要阻止業務邏輯消費者的進一步作業，我們稱這個概念為gating，更準確的說是這種行為的超集稱為gating，Gating發生在兩個地方：第一用來保證生產者不能超過消費者，這通過呼叫RingBuffer.addGatingConsumers()把相關消費者添加到Disruptor實作，第二，先前提到的情況是通過從必須首先完成其處理的組件構造包含序列的SequenceBarrier來實作的，
回顧圖1，有三個消費者監聽RingBuffer的事件，在這個例子中，有一個依賴圖，ApplicationConsumer依賴JournalConsumer和ReplicationConsumer，這意味著JournalConsumer和ReplicationConsumer可以相互并行運行，依賴關系可以從ApplicationConsumer的SequenceBarrier和JournalConsumer及ReplicationConsumer的Sequence觀察到，同時引起注意的是Sequencer和下游消費者的關系，它的一個角色是保證發布不會環繞RingBuffer，為了做到這點，下游消費者的Sequence不能小于RingBuffer的Sequence，然而，使用依賴圖會發生一個有趣的優化，由于ApplicationConsumer Sequence保證小于等于JournalConsumer和ReplicationConsumer（由依賴關系保證），Sequencer只需要觀察ApplicationConsumer的Sequence，從廣義上來說，Sequencer只需要關注依賴樹種葉子節點的消費者Sequence，

事件預分配

Disruptor的一個目標是可以用于低延遲環境中，在低延遲系統中，有必要減少或消除記憶體分配，在Java系統中，目標是減少由于垃圾回收造成的停頓次數（在低延遲的C/C++系統中，重記憶體分配會由于記憶體分配器的征用也可能導致問題），
為了支持這個目標，用戶可以預分配Disruptor中事件的存盤，用戶提供的EventFactory會在Disruptor中RingBuffer每個條目構建時呼叫，當發布新資料到Disruptor時，有API供用戶呼叫來持有構建出的物件，這樣可以呼叫物件的方法或更新物件屬性，在正確實作下，Disruptor保證這些操作操作是并發安全的，

可選的無鎖

對低延遲的渴望造就的另一個實作細節是無鎖演算法在Disruptor中的大量使用，所有記憶體可見性和正確性保證使用記憶體屏障和/或CAS操作實作，真正使用鎖的場景只有一個，那就是使用BlockingWatiStrategy，這樣做只為了使用Condition讓消費執行緒可以在等待新事件到達前進行park操作，許多低延遲系統使用忙等待(busy-wait)來避免使用Condition可能導致的抖動，然而一些系統的忙等待操作會導致性能的急劇下降，尤其是CPU資源被嚴重制約時，比方說在虛擬環境下的web服務器，

入門指南

基本的事件生產和消費

從簡單的事件開始：

public class LongEvent
{
    private long value;

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

為了讓Disruptor能夠預分配事件，我們需要提供一個EventFactory完成構建：

import com.lmax.disruptor.EventFactory;

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

事件定義好后，需要創建消費者處理這些事件，這里只做簡單的列印：

import com.lmax.disruptor.EventHandler;

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

我們還需要一個事件的生產源，舉個例子，假定資料是來自某種I/O設備，如網路或檔案的位元組緩沖(ByteBuffer)，

import com.lmax.disruptor.RingBuffer;

public class LongEventProducer
{
    private final RingBuffer<LongEvent> ringBuffer;

    public LongEventProducer(RingBuffer<LongEvent> ringBuffer)
    {
        this.ringBuffer = ringBuffer;
    }

    public void onData(ByteBuffer bb)
    {
        long sequence = ringBuffer.next();  // Grab the next sequence
        try
        {
            LongEvent event = ringBuffer.get(sequence); // Get the entry in the Disruptor
                                                        // for the sequence
            event.set(bb.getLong(0));  // Fill with data
        }
        finally
        {
            ringBuffer.publish(sequence);
        }
    }
}

可以發現，相比使用簡單的queue，事件的發布更具有相關性，這是由于需要事件預分配，事件發布需要（最低）2階段方式，先宣告環形緩沖器中的槽位，再發布可用資料，同時也需要把發布程序使用try/finally塊包裹起來，如果宣告了環形緩沖的一個槽位（通過呼叫RingBuffer.next()）然后必須發布這個序列，如果沒有這么做，會導致Disruptor的狀態損壞(corruption)，特別地，在多生產者的情況下，這將會導致消費者阻塞，只能通過重啟解決，

使用3.x版本的Translator

Disruptor3.0提供了一種富Lambda風格的API，旨在幫助開發者屏蔽直接操作RingBuffer的復雜性，所以3.0以上版本發布訊息更好的辦法是通過事件發布者(Event Publisher)或事件翻譯器(Event Translator)API，如下

import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.EventTranslatorOneArg;

public class LongEventProducerWithTranslator
{
    private final RingBuffer<LongEvent> ringBuffer;
   
    public LongEventProducerWithTranslator(RingBuffer<LongEvent> ringBuffer)
    {
        this.ringBuffer = ringBuffer;
    }
   
    private static final EventTranslatorOneArg<LongEvent, ByteBuffer> TRANSLATOR =
        new EventTranslatorOneArg<LongEvent, ByteBuffer>()
        {
            public void translateTo(LongEvent event, long sequence, ByteBuffer bb)
            {
                event.set(bb.getLong(0));
            }
        };

    public void onData(ByteBuffer bb)
    {
        ringBuffer.publishEvent(TRANSLATOR, bb);
    }
}

這種方法另一個好處是翻譯器代碼可以放到一個單獨的類中，以便于更容易進行單元測驗，Disruptor提供了一些用于翻譯器的不同的介面（EventTranslator，EventTranslatorOneArg，EventTranslatorTwoArg，等），這樣做的原因是，允許翻譯器表示為靜態類，或以非捕獲lambda運算式（使用java8時）作為翻譯方法引數，通過呼叫RingBuffer上的翻譯器進行傳遞，
最后一步是把上面這些步驟統一到一起，可以手工把這些組件都組裝到一起，但還是有點復雜，所以引入了DSL來簡化構建，盡管通過DSL的方式不能使用有些復雜選項，但這種方式還是適合絕大多數場景，

import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class LongEventMain
{
    public static void main(String[] args) throws Exception
    {
        // Executor that will be used to construct new threads for consumers
        Executor executor = Executors.newCachedThreadPool();

        // The factory for the event
        LongEventFactory factory = new LongEventFactory();

        // Specify the size of the ring buffer, must be power of 2.
        int bufferSize = 1024;

        // Construct the Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(factory, bufferSize, executor);

        // Connect the handler
        disruptor.handleEventsWith(new LongEventHandler());

        // Start the Disruptor, starts all threads running
        disruptor.start();

        // Get the ring buffer from the Disruptor to be used for publishing.
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

        LongEventProducer producer = new LongEventProducer(ringBuffer);

        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; true; l++)
        {
            bb.putLong(0, l);
            producer.onData(bb);
            Thread.sleep(1000);
        }
    }
}

使用Java8

Disruptor API的設計影響之一是Java 8將依靠功能介面的概念作為Java Lambdas的型別宣告， Disruptor API中的大多數介面定義符合功能介面的要求，因此可以使用Lambda而不是自定義類，這樣可以減少所需的重復代碼(boiler place)，

import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class LongEventMain
{
    public static void main(String[] args) throws Exception
    {
        // Executor that will be used to construct new threads for consumers
        Executor executor = Executors.newCachedThreadPool();

        // Specify the size of the ring buffer, must be power of 2.
        int bufferSize = 1024;

        // Construct the Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, executor);

        // Connect the handler
        disruptor.handleEventsWith((event, sequence, endOfBatch) -> System.out.println("Event: " + event));

        // Start the Disruptor, starts all threads running
        disruptor.start();

        // Get the ring buffer from the Disruptor to be used for publishing.
        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);
        }
    }
}

注意有一些類（如handler，translator）不再需要了，還要注意用于publishEvent()的lambda是如何參考傳入的引數的，如果使用如下代碼代替：

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

這會創建一個capturing lambda，意味著需要實體化一個物件來持有ByteBuffer bb變數，通過呼叫publishEvent()來傳遞lambda，這樣會創建額外不必須的垃圾，所以如果需要低GC壓力就需要傳遞引數給lambda，
使用這種方法參考可以代替匿名的lamdba，以這種方式重寫這個例子是可能的，

import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.RingBuffer;
import java.nio.ByteBuffer;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class LongEventMain
{
    public static void handleEvent(LongEvent event, long sequence, boolean endOfBatch)
    {
        System.out.println(event);
    }

    public static void translate(LongEvent event, long sequence, ByteBuffer buffer)
    {
        event.set(buffer.getLong(0));
    }

    public static void main(String[] args) throws Exception
    {
        // Executor that will be used to construct new threads for consumers
        Executor executor = Executors.newCachedThreadPool();

        // Specify the size of the ring buffer, must be power of 2.
        int bufferSize = 1024;

        // Construct the Disruptor
        Disruptor<LongEvent> disruptor = new Disruptor<>(LongEvent::new, bufferSize, executor);

        // Connect the handler
        disruptor.handleEventsWith(LongEventMain::handleEvent);

        // Start the Disruptor, starts all threads running
        disruptor.start();

        // Get the ring buffer from the Disruptor to be used for publishing.
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();

        ByteBuffer bb = ByteBuffer.allocate(8);
        for (long l = 0; true; l++)
        {
            bb.putLong(0, l);
            ringBuffer.publishEvent(LongEventMain::translate, bb);
            Thread.sleep(1000);
        }
    }
}

基本調優選項

使用上述的方法可以在最廣泛的部署場景中作業正常，然而，如果你能夠確定Disruptor將要運行的硬體和軟體環境，就可以調整引數提升性能，主要有以下兩種調優方式：單vs.多生產者和替換等待策略，

單vs.多生產者

提高并發系統性能的最佳方法之一就是遵守單作者原則(Single Writer Principle https://mechanical-sympathy.blogspot.tw/2011/09/single-writer-principle.html，這適用于Disruptor，如果你的情況是只有一個執行緒會在Disruptor中發布事件，那就可以利用此功能獲得額外的性能提升，

public class LongEventMain
{
    public static void main(String[] args) throws Exception
    {
        //.....
        // Construct the Disruptor with a SingleProducerSequencer
        Disruptor<LongEvent> disruptor = new Disruptor(
            factory, bufferSize, ProducerType.SINGLE, new BlockingWaitStrategy(), executor);
        //.....
    }
}

OneToOne 性能測驗（https://github.com/LMAX-Exchange/disruptor/blob/master/src/perftest/java/com/lmax/disruptor/sequenced/OneToOneSequencedThroughputTest.java）可以說明這種技術能夠提升多少性能，以下測驗使用i7 Sandy Bridge MacBook Air，
多生產者

Run 0, Disruptor=26,553,372 ops/sec
Run 1, Disruptor=28,727,377 ops/sec
Run 2, Disruptor=29,806,259 ops/sec
Run 3, Disruptor=29,717,682 ops/sec
Run 4, Disruptor=28,818,443 ops/sec
Run 5, Disruptor=29,103,608 ops/sec
Run 6, Disruptor=29,239,766 ops/sec

單生產者

Run 0, Disruptor=89,365,504 ops/sec
Run 1, Disruptor=77,579,519 ops/sec
Run 2, Disruptor=78,678,206 ops/sec
Run 3, Disruptor=80,840,743 ops/sec
Run 4, Disruptor=81,037,277 ops/sec
Run 5, Disruptor=81,168,831 ops/sec
Run 6, Disruptor=81,699,346 ops/sec

替換等待策略

BlockingWaitStategy

Disruptor默認的等待策略是BlockingWaitStategy，在BlockingWaitStategy內部使用一個典型的鎖和條件(a typical lock and condition)變數處理執行緒喚醒，BlockingWaitStategy是可用等待策略中最慢的，但也是在CPU使用上最保守的，同時也將在最廣泛的部署選項中提供最一致的行為，然而，再說一次，了解部署系統可以獲得額外的性能提升，

SleepingWaitStrategy

類似BlockingWaitStategy，SleepingWaitStrategy也試圖在CPU使用上保持保守，這通過一個忙等待(busy wait loop)回圈實作，但在回圈中間會呼叫LockSupport.parkNanos(1)，在一個典型的Linux系統，這樣會暫停執行緒大概60μs(譯注1μs=1000ns)，但它的好處是生產者執行緒除了增加回應的計數器外，不需要采取任何行動，而且不需要給條件變數發信號的成本(cost of signalling a condition variable)，然而，生產者和消費者轉移事件的平均延遲會增加，這種方式最好作業在不需要低延遲，但對生產者執行緒影響最小的情況下，一個常見的使用場景是異步日志，

YieldingWaitStrategy

可用于低延遲系統的兩種等待策略其中之一，這種策略通過消耗CPU時鐘周期來達到優化延遲的目的，這種策略使用忙回圈(busy spin)等待正確的序號到達，在回圈內部，Thread.yield()將被呼叫，來允許其他排隊中的執行緒運行，當需要很高的性能，而且事件處理者EventHandler的執行緒數少于CPU邏輯核心數時（比如使用超執行緒時），推薦使用這種策略，

BusySpinWaitStrategy

這種策略有最高的性能，但也有最高的部署邊境限制，這種等待策略應該只用于事件處理者執行緒小于CPU物理核心數，

清除環形緩沖的物件

使用Disruptor傳輸資料時，物件的存活周期有可能比預期更長，為了避免發生這種情況，有必要在事件處理完畢后做清理，如果有一個事件處理器，在這個事件處理器中做清理就足夠了，如果有一個事件處理鏈，那就可能會在鏈尾需要一個特定的處理器來清理這個物件，

class ObjectEvent<T>
{
    T val;

    void clear()
    {
        val = null;
    }
}

public class ClearingEventHandler<T> implements EventHandler<ObjectEvent<T>>
{
    public void onEvent(ObjectEvent<T> event, long sequence, boolean endOfBatch)
    {
        // Failing to call clear here will result in the
        // object associated with the event to live until
        // it is overwritten once the ring buffer has wrapped
        // around to the beginning.
        event.clear();
    }
}

public static void main(String[] args)
{
    Disruptor<ObjectEvent<String>> disruptor = new Disruptor<>(
        () -> ObjectEvent<String>(), bufferSize, executor);

    disruptor
        .handleEventsWith(new ProcessingEventHandler())
        .then(new ClearingObjectHandler());
}

文章到此就結束了！

最后來自小編的福利

以下是小編整理的一份大廠真題的面試資料，以及2020最新Java核心技術整理的資料集錦，需要領取的小伙伴可以點我免費領取，編程的世界永遠向所有熱愛編程的人開放，這是一個自由，平等，共享的世界，我始終是這樣堅信的，

部分資料圖片：

喜歡小編的分享可以點贊關注哦，小編持續為你分享最新文章和福利領取哦

轉載請註明出處，本文鏈接：https://www.uj5u.com/ruanti/238587.html

標籤：其他

上一篇：作業幫面經（已拿offer）

下一篇：B-樹基本操作的實作利用C++模板類封裝