背景

Disruptor是英國外匯交易公司LMAX開發的一個高效能佇列，研發的初衷是解決記憶體佇列的延遲問題（在效能測試中發現竟然與I/O操作處於同樣的數量級）。基於Disruptor開發的系統單執行緒能支撐每秒600萬訂單，2010年在QCon演講後，獲得了業界關注。2011年，企業應用軟體專家Martin Fowler專門撰寫長文介紹。同年它還獲得了Oracle官方的Duke大獎。

目前，包括Apache Storm、Camel、Log4j 2在內的很多知名專案都應用了Disruptor以獲取高效能。在美團技術團隊它也有不少應用，有的專案架構借鑑了它的設計機制。本文從實戰角度剖析了Disruptor的實現原理。

需要特別指出的是，這裡所說的佇列是系統內部的記憶體佇列，而不是Kafka這樣的分散式佇列。另外，本文所描述的Disruptor特性限於3.3.4。

Java內建佇列

介紹Disruptor之前，我們先來看一看常用的執行緒安全的內建佇列有什麼問題。Java的內建佇列如下表所示。

佇列的底層一般分成三種：陣列、連結串列和堆。其中，堆一般情況下是為了實現帶有優先順序特性的佇列，暫且不考慮。

我們就從陣列和連結串列兩種資料結構來看，基於陣列執行緒安全的佇列，比較典型的是ArrayBlockingQueue，它主要通過加鎖的方式來保證執行緒安全；基於連結串列的執行緒安全佇列分成LinkedBlockingQueue和ConcurrentLinkedQueue兩大類，前者也通過鎖的方式來實現執行緒安全，而後者以及上面表格中的LinkedTransferQueue都是通過原子變數compare and swap（以下簡稱“CAS”）這種不加鎖的方式來實現的。

通過不加鎖的方式實現的佇列都是無界的（無法保證佇列的長度在確定的範圍內）；而加鎖的方式，可以實現有界佇列。在穩定性要求特別高的系統中，為了防止生產者速度過快，導致記憶體溢位，只能選擇有界佇列；同時，為了減少Java的垃圾回收對系統性能的影響，會盡量選擇array/heap格式的資料結構。這樣篩選下來，符合條件的佇列就只有ArrayBlockingQueue。

ArrayBlockingQueue的問題

ArrayBlockingQueue在實際使用過程中，會因為加鎖和偽共享等出現嚴重的效能問題，我們下面來分析一下。

加鎖

現實程式設計過程中，加鎖通常會嚴重地影響效能。執行緒會因為競爭不到鎖而被掛起，等鎖被釋放的時候，執行緒又會被恢復，這個過程中存在著很大的開銷，並且通常會有較長時間的中斷，因為當一個執行緒正在等待鎖時，它不能做任何其他事情。如果一個執行緒在持有鎖的情況下被延遲執行，例如發生了缺頁錯誤、排程延遲或者其它類似情況，那麼所有需要這個鎖的執行緒都無法執行下去。如果被阻塞執行緒的優先順序較高，而持有鎖的執行緒優先順序較低，就會發生優先順序反轉。

Disruptor論文中講述了一個實驗：

• 這個測試程式呼叫了一個函式，該函式會對一個64位的計數器迴圈自增5億次。
• 機器環境：2.4G 6核
• 運算： 64位的計數器累加5億次

CAS操作比單執行緒無鎖慢了1個數量級；有鎖且多執行緒併發的情況下，速度比單執行緒無鎖慢3個數量級。可見無鎖速度最快。

單執行緒情況下，不加鎖的效能 > CAS操作的效能 > 加鎖的效能。

在多執行緒情況下，為了保證執行緒安全，必須使用CAS或鎖，這種情況下，CAS的效能超過鎖的效能，前者大約是後者的8倍。

綜上可知，加鎖的效能是最差的。

關於鎖和CAS

保證執行緒安全一般分成兩種方式：鎖和原子變數。

鎖

圖1 通過加鎖的方式實現執行緒安全

採取加鎖的方式，預設執行緒會衝突，訪問資料時，先加上鎖再訪問，訪問之後再解鎖。通過鎖界定一個臨界區，同時只有一個執行緒進入。如上圖所示，Thread2訪問Entry的時候，加了鎖，Thread1就不能再執行訪問Entry的程式碼，從而保證執行緒安全。

下面是ArrayBlockingQueue通過加鎖的方式實現的offer方法，保證執行緒安全。

public boolean offer(E e) {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == items.length)
            return false;
        else {
            insert(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}

原子變數

原子變數能夠保證原子性的操作，意思是某個任務在執行過程中，要麼全部成功，要麼全部失敗回滾，恢復到執行之前的初態，不存在初態和成功之間的中間狀態。例如CAS操作，要麼比較並交換成功，要麼比較並交換失敗。由CPU保證原子性。

通過原子變數可以實現執行緒安全。執行某個任務的時候，先假定不會有衝突，若不發生衝突，則直接執行成功；當發生衝突的時候，則執行失敗，回滾再重新操作，直到不發生衝突。

圖2 通過原子變數CAS實現執行緒安全

如圖所示，Thread1和Thread2都要把Entry加1。若不加鎖，也不使用CAS，有可能Thread1取到了myValue=1，Thread2也取到了myValue=1，然後相加，Entry中的value值為2。這與預期不相符，我們預期的是Entry的值經過兩次相加後等於3。

CAS會先把Entry現在的value跟執行緒當初讀出的值相比較，若相同，則賦值；若不相同，則賦值執行失敗。一般會通過while/for迴圈來重新執行，直到賦值成功。

程式碼示例是AtomicInteger的getAndAdd方法。CAS是CPU的一個指令，由CPU保證原子性。

/**
 * Atomically adds the given value to the current value.
 *
 * @param delta the value to add
 * @return the previous value
 */
public final int getAndAdd(int delta) {
    for (;;) {
        int current = get();
        int next = current + delta;
        if (compareAndSet(current, next))
            return current;
    }
}

/**
 * Atomically sets the value to the given updated value
 * if the current value {@code ==} the expected value.
 *
 * @param expect the expected value
 * @param update the new value
 * @return true if successful. False return indicates that
 * the actual value was not equal to the expected value.
 */
public final boolean compareAndSet(int expect, int update) {
    return unsafe.compareAndSwapInt(this, valueOffset, expect, update);
}

在高度競爭的情況下，鎖的效能將超過原子變數的效能，但是更真實的競爭情況下，原子變數的效能將超過鎖的效能。同時原子變數不會有死鎖等活躍性問題。

偽共享

什麼是共享

下圖是計算的基本結構。L1、L2、L3分別表示一級快取、二級快取、三級快取，越靠近CPU的快取，速度越快，容量也越小。所以L1快取很小但很快，並且緊靠著在使用它的CPU核心；L2大一些，也慢一些，並且仍然只能被一個單獨的CPU核使用；L3更大、更慢，並且被單個插槽上的所有CPU核共享；最後是主存，由全部插槽上的所有CPU核共享。

圖3 計算機CPU與快取示意圖

當CPU執行運算的時候，它先去L1查詢所需的資料、再去L2、然後是L3，如果最後這些快取中都沒有，所需的資料就要去主記憶體拿。走得越遠，運算耗費的時間就越長。所以如果你在做一些很頻繁的事，你要儘量確保資料在L1快取中。

另外，執行緒之間共享一份資料的時候，需要一個執行緒把資料寫回主存，而另一個執行緒訪問主存中相應的資料。

下面是從CPU訪問不同層級資料的時間概念:

可見CPU讀取主存中的資料會比從L1中讀取慢了近2個數量級。

快取行

Cache是由很多個cache line組成的。每個cache line通常是64位元組，並且它有效地引用主記憶體中的一塊兒地址。一個Java的long型別變數是8位元組，因此在一個快取行中可以存8個long型別的變數。

CPU每次從主存中拉取資料時，會把相鄰的資料也存入同一個cache line。

在訪問一個long陣列的時候，如果陣列中的一個值被載入到快取中，它會自動載入另外7個。因此你能非常快的遍歷這個陣列。事實上，你可以非常快速的遍歷在連續記憶體塊中分配的任意資料結構。

下面的例子是測試利用cache line的特性和不利用cache line的特性的效果對比。

package com.meituan.FalseSharing;

/**
 * @author gongming
 * @description
 * @date 16/6/4
 */
public class CacheLineEffect {
    //考慮一般快取行大小是64位元組，一個 long 型別佔8位元組
    static  long[][] arr;

    public static void main(String[] args) {
        arr = new long[1024 * 1024][];
        for (int i = 0; i < 1024 * 1024; i++) {
            arr[i] = new long[8];
            for (int j = 0; j < 8; j++) {
                arr[i][j] = 0L;
            }
        }
        long sum = 0L;
        long marked = System.currentTimeMillis();
        for (int i = 0; i < 1024 * 1024; i+=1) {
            for(int j =0; j< 8;j++){
                sum = arr[i][j];
            }
        }
        System.out.println("Loop times:" + (System.currentTimeMillis() - marked) + "ms");

        marked = System.currentTimeMillis();
        for (int i = 0; i < 8; i+=1) {
            for(int j =0; j< 1024 * 1024;j++){
                sum = arr[j][i];
            }
        }
        System.out.println("Loop times:" + (System.currentTimeMillis() - marked) + "ms");
    }
}

在2G Hz、2核、8G記憶體的執行環境中測試，速度差一倍。

結果：
Loop times:30ms
Loop times:65ms

什麼是偽共享

ArrayBlockingQueue有三個成員變數：

• takeIndex：需要被取走的元素下標
• putIndex：可被元素插入的位置的下標
• count：佇列中元素的數量

這三個變數很容易放到一個快取行中，但是之間修改沒有太多的關聯。所以每次修改，都會使之前快取的資料失效，從而不能完全達到共享的效果。

圖4 ArrayBlockingQueue偽共享示意圖

如上圖所示，當生產者執行緒put一個元素到ArrayBlockingQueue時，putIndex會修改，從而導致消費者執行緒的快取中的快取行無效，需要從主存中重新讀取。

這種無法充分使用快取行特性的現象，稱為偽共享。

對於偽共享，一般的解決方案是，增大陣列元素的間隔使得由不同執行緒存取的元素位於不同的快取行上，以空間換時間。

package com.meituan.FalseSharing;

public class FalseSharing implements Runnable{
        public final static long ITERATIONS = 500L * 1000L * 100L;
        private int arrayIndex = 0;

        private static ValuePadding[] longs;
        public FalseSharing(final int arrayIndex) {
            this.arrayIndex = arrayIndex;
        }

        public static void main(final String[] args) throws Exception {
            for(int i=1;i<10;i++){
                System.gc();
                final long start = System.currentTimeMillis();
                runTest(i);
                System.out.println("Thread num "+i+" duration = " + (System.currentTimeMillis() - start));
            }

        }

        private static void runTest(int NUM_THREADS) throws InterruptedException {
            Thread[] threads = new Thread[NUM_THREADS];
            longs = new ValuePadding[NUM_THREADS];
            for (int i = 0; i < longs.length; i++) {
                longs[i] = new ValuePadding();
            }
            for (int i = 0; i < threads.length; i++) {
                threads[i] = new Thread(new FalseSharing(i));
            }

            for (Thread t : threads) {
                t.start();
            }

            for (Thread t : threads) {
                t.join();
            }
        }

        public void run() {
            long i = ITERATIONS + 1;
            while (0 != --i) {
                longs[arrayIndex].value = 0L;
            }
        }

        public final static class ValuePadding {
            protected long p1, p2, p3, p4, p5, p6, p7;
            protected volatile long value = 0L;
            protected long p9, p10, p11, p12, p13, p14;
            protected long p15;
        }
        public final static class ValueNoPadding {
            // protected long p1, p2, p3, p4, p5, p6, p7;
            protected volatile long value = 0L;
            // protected long p9, p10, p11, p12, p13, p14, p15;
        }
}

在2G Hz，2核，8G記憶體, jdk 1.7.0_45 的執行環境下，使用了共享機制比沒有使用共享機制，速度快了4倍左右。

結果：
Thread num 1 duration = 447
Thread num 2 duration = 463
Thread num 3 duration = 454
Thread num 4 duration = 464
Thread num 5 duration = 561
Thread num 6 duration = 606
Thread num 7 duration = 684
Thread num 8 duration = 870
Thread num 9 duration = 823

把程式碼中ValuePadding都替換為ValueNoPadding後的結果：
Thread num 1 duration = 446
Thread num 2 duration = 2549
Thread num 3 duration = 2898
Thread num 4 duration = 3931
Thread num 5 duration = 4716
Thread num 6 duration = 5424
Thread num 7 duration = 4868
Thread num 8 duration = 4595
Thread num 9 duration = 4540

備註：在jdk1.8中，有專門的註解@Contended來避免偽共享，更優雅地解決問題。

Disruptor的設計方案

Disruptor通過以下設計來解決佇列速度慢的問題：

• 環形陣列結構

為了避免垃圾回收，採用陣列而非連結串列。同時，陣列對處理器的快取機制更加友好。

• 元素位置定位

陣列長度2^n，通過位運算，加快定位的速度。下標採取遞增的形式。不用擔心index溢位的問題。index是long型別，即使100萬QPS的處理速度，也需要30萬年才能用完。

• 無鎖設計

每個生產者或者消費者執行緒，會先申請可以操作的元素在陣列中的位置，申請到之後，直接在該位置寫入或者讀取資料。

下面忽略陣列的環形結構，介紹一下如何實現無鎖設計。整個過程通過原子變數CAS，保證操作的執行緒安全。

一個生產者

寫資料

生產者單執行緒寫資料的流程比較簡單：

1. 申請寫入m個元素；
2. 若是有m個元素可以寫入，則返回最大的序列號。這兒主要判斷是否會覆蓋未讀的元素；
3. 若是返回的正確，則生產者開始寫入元素。

圖5 單個生產者生產過程示意圖

多個生產者

多個生產者的情況下，會遇到“如何防止多個執行緒重複寫同一個元素”的問題。Disruptor的解決方法是，每個執行緒獲取不同的一段陣列空間進行操作。這個通過CAS很容易達到。只需要在分配元素的時候，通過CAS判斷一下這段空間是否已經分配出去即可。

但是會遇到一個新問題：如何防止讀取的時候，讀到還未寫的元素。Disruptor在多個生產者的情況下，引入了一個與Ring Buffer大小相同的buffer：available Buffer。當某個位置寫入成功的時候，便把availble Buffer相應的位置置位，標記為寫入成功。讀取的時候，會遍歷available Buffer，來判斷元素是否已經就緒。

下面分讀資料和寫資料兩種情況介紹。

讀資料

生產者多執行緒寫入的情況會複雜很多：

1. 申請讀取到序號n；
2. 若writer cursor >= n，這時仍然無法確定連續可讀的最大下標。從reader cursor開始讀取available Buffer，一直查到第一個不可用的元素，然後返回最大連續可讀元素的位置；
3. 消費者讀取元素。

如下圖所示，讀執行緒讀到下標為2的元素，三個執行緒Writer1/Writer2/Writer3正在向RingBuffer相應位置寫資料，寫執行緒被分配到的最大元素下標是11。

讀執行緒申請讀取到下標從3到11的元素，判斷writer cursor>=11。然後開始讀取availableBuffer，從3開始，往後讀取，發現下標為7的元素沒有生產成功，於是WaitFor(11)返回6。

然後，消費者讀取下標從3到6共計4個元素。

圖6 多個生產者情況下，消費者消費過程示意圖

寫資料

多個生產者寫入的時候：

1. 申請寫入m個元素；
2. 若是有m個元素可以寫入，則返回最大的序列號。每個生產者會被分配一段獨享的空間；
3. 生產者寫入元素，寫入元素的同時設定available Buffer裡面相應的位置，以標記自己哪些位置是已經寫入成功的。

如下圖所示，Writer1和Writer2兩個執行緒寫入陣列，都申請可寫的陣列空間。Writer1被分配了下標3到下表5的空間，Writer2被分配了下標6到下標9的空間。

Writer1寫入下標3位置的元素，同時把available Buffer相應位置置位，標記已經寫入成功，往後移一位，開始寫下標4位置的元素。Writer2同樣的方式。最終都寫入完成。

圖7 多個生產者情況下，生產者生產過程示意圖

防止不同生產者對同一段空間寫入的程式碼，如下所示：

public long tryNext(int n) throws InsufficientCapacityException
{
    if (n < 1)
    {
        throw new IllegalArgumentException("n must be > 0");
    }

    long current;
    long next;

    do
    {
        current = cursor.get();
        next = current + n;

        if (!hasAvailableCapacity(gatingSequences, n, current))
        {
            throw InsufficientCapacityException.INSTANCE;
        }
    }
    while (!cursor.compareAndSet(current, next));

    return next;
}

通過do/while迴圈的條件cursor.compareAndSet(current, next)，來判斷每次申請的空間是否已經被其他生產者佔據。假如已經被佔據，該函式會返回失敗，While迴圈重新執行，申請寫入空間。

消費者的流程與生產者非常類似，這兒就不多描述了。

總結

Disruptor通過精巧的無鎖設計實現了在高併發情形下的高效能。

在美團內部，很多高併發場景借鑑了Disruptor的設計，減少競爭的強度。其設計思想可以擴充套件到分散式場景，通過無鎖設計，來提升服務效能。

程式碼樣例

使用Disruptor比使用ArrayBlockingQueue略微複雜，為方便讀者上手，增加程式碼樣例。

程式碼實現的功能：每10ms向disruptor中插入一個元素，消費者讀取資料，並列印到終端。詳細邏輯請細讀程式碼。

以下程式碼基於3.3.4版本的Disruptor包。

package com.meituan.Disruptor;

/**
 * @description disruptor程式碼樣例。每10ms向disruptor中插入一個元素，消費者讀取資料，並列印到終端
 */
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.ThreadFactory;


public class DisruptorMain
{
    public static void main(String[] args) throws Exception
    {
        // 佇列中的元素
        class Element {

            private int value;

            public int get(){
                return value;
            }

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

        }

        // 生產者的執行緒工廠
        ThreadFactory threadFactory = new ThreadFactory(){
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "simpleThread");
            }
        };

        // RingBuffer生產工廠,初始化RingBuffer的時候使用
        EventFactory<Element> factory = new EventFactory<Element>() {
            @Override
            public Element newInstance() {
                return new Element();
            }
        };

        // 處理Event的handler
        EventHandler<Element> handler = new EventHandler<Element>(){
            @Override
            public void onEvent(Element element, long sequence, boolean endOfBatch)
            {
                System.out.println("Element: " + element.get());
            }
        };

        // 阻塞策略
        BlockingWaitStrategy strategy = new BlockingWaitStrategy();

        // 指定RingBuffer的大小
        int bufferSize = 16;

        // 建立disruptor，採用單生產者模式
        Disruptor<Element> disruptor = new Disruptor(factory, bufferSize, threadFactory, ProducerType.SINGLE, strategy);

        // 設定EventHandler
        disruptor.handleEventsWith(handler);

        // 啟動disruptor的執行緒
        disruptor.start();

        RingBuffer<Element> ringBuffer = disruptor.getRingBuffer();

        for (int l = 0; true; l++)
        {
            // 獲取下一個可用位置的下標
            long sequence = ringBuffer.next();  
            try
            {
                // 返回可用位置的元素
                Element event = ringBuffer.get(sequence); 
                // 設定該位置元素的值
                event.set(l); 
            }
            finally
            {
                ringBuffer.publish(sequence);
            }
            Thread.sleep(10);
        }
    }
}

效能

以下面這些模式測試效能:

吞吐量測試資料（每秒的數量）如下。

環境：

• CPU:Intel Core i7 860 @ 2.8 GHz without HT
• JVM:Java 1.6.0_25 64-bit
• OS:Windows 7

環境：

• CPU:Intel Core i7-2720QM
• JVM:Java 1.6.0_25 64-bit
• OS:Ubuntu 11.04

依據併發競爭的激烈程度的不同，Disruptor比ArrayBlockingQueue吞吐量快4~7倍。

按照Pipeline: 1P – 3C的連線模式測試延遲，生產者兩次寫入之間的延遲為1ms。

執行環境：

• CPU:2.2GHz Core i7-2720QM
• Java: 1.6.0_25 64-bit
• OS:Ubuntu 11.04.

可見，平均延遲差了3個數量級。

等待策略

生產者的等待策略

暫時只有休眠1ns。

LockSupport.parkNanos(1);

消費者的等待策略

Log4j 2應用場景

Log4j 2相對於Log4j 1最大的優勢在於多執行緒併發場景下效能更優。該特性源自於Log4j 2的非同步模式採用了Disruptor來處理。
在Log4j 2的配置檔案中可以配置WaitStrategy，預設是Timeout策略。下面是Log4j 2中對WaitStrategy的配置官方文件：

效能差異

loggers all async採用的是Disruptor，而Async Appender採用的是ArrayBlockingQueue佇列。

由圖可見，單執行緒情況下，loggers all async與Async Appender吞吐量相差不大，但是在64個執行緒的時候，loggers all async的吞吐量比Async Appender增加了12倍，是Sync模式的68倍。

圖8 Log4j 2各個模式效能比較

美團在公司內部統一推行日誌接入規範，要求必須使用Log4j 2，使普通單機QPS的上限不再只停留在幾千，極高地提升了服務效能。

參考文件