Disruptor 是線程內(nèi)通信框架，用于線程里共享數(shù)據(jù)。LMAX 創(chuàng)建Disruptor 作為可靠消息架構(gòu)的一部分并將它設(shè)計成一種在不同組件中共享數(shù)據(jù)非?？斓姆椒ā?基于 Mechanical Sympathy（對于計算機(jī)底層硬件的理解），基本的計算機(jī)科學(xué)以及領(lǐng)域驅(qū)動設(shè)計，Disruptor 已經(jīng)發(fā)展成為一個幫助開發(fā)人員解決很多繁瑣并發(fā)編程問題的框架。很多架構(gòu)都普遍使用一個隊列共享線程間的數(shù)據(jù)（即傳送消息）。圖1 展示了一個在不同的階段中通過使用隊列來傳送消息的例子（每個藍(lán)色的圈代表一個線程）。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-1.png" alt="" />

圖 1

這種架構(gòu)允許生產(chǎn)者線程（圖1中的 stage 1）在 stage2 很忙以至于無法立刻處理的時候能夠繼續(xù)執(zhí)行下一步操作，從而提供了解決系統(tǒng)中數(shù)據(jù)擁堵的方法。這里隊列可以看成是不同線程之間的緩沖。

在這種最簡單的情況下，Disruptor 可以用來代替隊列作為在不同的線程傳遞消息的工具（如圖 2 所示）。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-2.png" alt="" />

圖2

這種數(shù)據(jù)結(jié)構(gòu)叫著 RingBuffer，是用數(shù)組實現(xiàn)的。Stage1 線程把數(shù)據(jù)放進(jìn) RingBuffer，而 Stage2 線程從 RingBuffer 中讀取數(shù)據(jù)。

圖2 中，可以看到 RingBuffer 中每格中都有序號，并且 RingBuffer 實時監(jiān)測值最大（最新）的序號，該序號指向 RingBuffer 中最后一格。序號會伴隨著越來越多的數(shù)據(jù)增加進(jìn) RingBuffer 中而增長。

Disruptor 的關(guān)鍵在于是它的設(shè)計目標(biāo)是在框架內(nèi)沒有競爭.這是通過遵守single-writer 原則，即只有一塊數(shù)據(jù)可以寫入一個數(shù)據(jù)塊中，而達(dá)到的。遵循這樣的規(guī)則使得 Disruptor 避免了代價高昂的 CAS 鎖，這也使得 Disruptor 非?？?。

Disruptor 通過使用 RingBuffer 以及每個事件處理器（EventProcessor）監(jiān)測各自的序號從而減少了競爭。這樣，事件處理器只能更新自己所獲得的序號。當(dāng)介紹向 RingBuffer 讀取和寫入數(shù)據(jù)時會對這個概念作進(jìn)一步闡述。

發(fā)布到 Disruptor

向 RingBuffer 寫入數(shù)據(jù)需要通過兩階段提交（two-phase commit）。首先，Stage 1 線程即發(fā)布者必須確定 RingBuffer 中下一個可以插入的格,如圖 3 所示。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-3.png" alt="" />

圖 3

RingBuffer 持有最近寫入格的序號（圖3中的18格），從而確定下一個插入格的序號。

RingBuffer 通過檢查所有事件處理器正在從 RingBuffe 中讀取的當(dāng)前序號來判斷下一個插入格是否空閑。

圖 4 顯示發(fā)現(xiàn)了下一個插入格。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-4.png" alt="" />

圖 4

當(dāng)發(fā)布者得到下一個序號后，它可以獲得該格中的對象，并可以對該對象進(jìn)行任意操作。你可以把格想象成一個簡單的可以寫入任意值的容器。

同時，在發(fā)布者處理 19 格數(shù)據(jù)的時候，RingBuffer 的序號依然是 18，所以其他事件處理器將不會讀到19格中的數(shù)據(jù)。

圖5表示對象的改動保存進(jìn)了 RingBuffer。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-5.png" alt="" />

圖5

最終，發(fā)布者最終將數(shù)據(jù)寫入19 格后，通知 RingBuffer 發(fā)布 19 格的數(shù)據(jù)。這時，RingBuffer 更新序號并且所有從 RingBuffer 讀數(shù)據(jù)的事件處理器都可以看到 19 格中的數(shù)據(jù)。

RingBuffer 中數(shù)據(jù)讀取

Disruptor 框架中包含了可以從 RingBuffer 中讀取數(shù)據(jù)的BatchEventProcessor，下面將概述它如何工作并著重介紹它的設(shè)計。

當(dāng)發(fā)布者向 RingBuffer 請求下一個空格以便寫入時，一個實際上并不真的從 RingBuffer 消費事件的事件處理器，將監(jiān)控它處理的最新的序號并請求它所需要的下一個序號。

圖5顯示事件處理器等待下一個序號。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-6.png" alt="" />

圖6

事件處理器不是直接向 RingBuffer 請求序號，而是通過 SequenceBarrie r向 RingBuffer 請求序號。其中具體實現(xiàn)細(xì)節(jié)對我們的理解并不重要，但是下面可以看到這樣做的目的很明顯。

如圖 6 中 Stage2 所示，事件處理器的最大序號是 16.它向 SequenceBarrier 調(diào)用 waitFor（17）以獲得 17 格中的數(shù)據(jù)。因為沒有數(shù)據(jù)寫入 RingBuffer，Stage2 事件處理器掛起等待下一個序號。如果這樣，沒有什么可以處理。但是，如圖6所示的情況，RingBuffer 已經(jīng)被填充到 18 格，所以 waitFor 函數(shù)將返回18并通知事件處理器，它可以讀取包括直到 18 格在內(nèi)的數(shù)據(jù)，如圖7所示。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-7.png" alt="" />

圖7

這種方法提供了非常好的批處理功能，可以在 BatchEventProcessor 源碼中看到。源碼中直接向 RingBuffer 批量獲取從下一個序號直到最大可以獲得的序號中的數(shù)據(jù)。

你可以通過實現(xiàn) EventHandler 使用批處理功能。在 Disruptor 性能測試中有關(guān)于如何使用批處理的例子，例如 FizzBuzzEventHandler。

是低延遲隊列？

當(dāng)然，Disruptor 可以被當(dāng)作低延遲隊列來使用。我們對于 Disruptor 之前版本的測試數(shù)據(jù)顯示了，運行在一個 2.2 GHz 的英特爾酷睿 i7-2720 QM 處理器上使用 Java 1.6.0_25 64位的 Ubuntu 的 11.04 三層管道模式架構(gòu)中，Disruptor 比 ArrayBlockingQueue 快了多少。表1顯示了在管道中的每跳延遲。有關(guān)此測試的更多詳細(xì)信息，請參閱 Disruptor 技術(shù)文件。

但是不要根據(jù)延遲數(shù)據(jù)得出 Disruptor 只是一種解決某種特定性能問題的方案，因為它不是。

更酷的東西

一個有意思的事是 Disruptor 是如何支持系統(tǒng)組件之間的依賴關(guān)系，并在線程之間共享數(shù)據(jù)時不產(chǎn)生競爭。

Disruptor 在設(shè)計上遵守 single-writer 原則從而實現(xiàn)零競爭，即每個數(shù)據(jù)位只能被一個線程寫入。但是，這不代表你不可以使用多個線程讀數(shù)據(jù)，而這正是 Disrupto r所支持的。

Disruptor 系統(tǒng)的最初設(shè)計是為了支持需要按照特定的順序發(fā)生的階段性類似流水線事件，這種需求在企業(yè)應(yīng)用系統(tǒng)開發(fā)中并不少見。圖 8 顯示了標(biāo)準(zhǔn)的 3 級流水線。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-8.png" alt="" />

圖 8

首先，每個事件都被寫入硬盤（日志）作為日后恢復(fù)用。其次，這些事件被復(fù)制到備份服務(wù)器。只有在這兩個階段后，系統(tǒng)開始業(yè)務(wù)邏輯處理。

按順序執(zhí)行上次操作是一個合乎邏輯的方法，但是并不是最有效的方法。日志和復(fù)制操作可以同步執(zhí)行，因為他們互相獨立。但是業(yè)務(wù)邏輯必須在他們都執(zhí)行完后才能執(zhí)行。圖9顯示他們可以并行互不依賴。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-9.png" alt="" />

圖 9

如果使用 Disruptor，前兩個階段（日志和復(fù)制）可以直接從 RingBuffer中讀取數(shù)據(jù)。正如圖7種的簡化圖所示，他們都使用一個單一的Sequence Barrier從RingBuffer 獲取下一個可用的序號。他們記錄他們使用過的序號，這樣他們知道那些事件已經(jīng)讀過并可以使用 BatchEventProcessor 批量獲取事件。

業(yè)務(wù)邏輯同樣可以從同一個 RingBuffer 中讀取事件，但是只限于前兩個階段已經(jīng)處理過事件。這是通過加入第二個 SequenceBarrier 實現(xiàn)的，用它來監(jiān)控處理日志的事件處理器和復(fù)制的事件處理器，當(dāng)請求最大可讀的序號時，它返回兩個處理器中較小的序號。

當(dāng)每個事件處理器都使用 SequenceBarrier 來確定哪些事件可以安全的從 RingBuffer 中讀出，那么就從中讀出這些事件。

http://wiki.jikexueyuan.com/project/disruptor-getting-started/images/12-10.png" alt="" />

圖10

有很多事件處理器都可以從 RingBuffer 中讀取序號，包括日志事件處理器，復(fù)制事件處理器等，但是只有一個處理器可以增加序號。這保證了共享數(shù)據(jù)沒有競爭。

如果有多個發(fā)布者？

Disruptor 也支持多個發(fā)布者向 RingBuffer 寫入。當(dāng)然，因為這樣的話必然會發(fā)生兩個不同的事件處理器寫入同一格的情況，這樣就會產(chǎn)生競爭。Disruptor 提供 ClaimStrategy 的處理方式應(yīng)對有多個發(fā)布者的情況。

結(jié)論

在這里，我已經(jīng)在總體上介紹了 Disruptor 框架是如何高性能在線程中共享數(shù)據(jù)，并簡單闡述了它的原理。有關(guān)更高級事件處理器以及向 RingBuffer 申請空間并等待下一個序號等很多策略在這里都沒有涉及， Disruptor 是開源的，到代碼中去搜索吧。

注1：源自 Oracle 出版的 Java 雜志，http://www.oracle.com/technetwork/cn/java/javamagazine/index.html

原創(chuàng)文章，轉(zhuǎn)載請注明：轉(zhuǎn)載自并發(fā)編程網(wǎng) – ifeve.com

本文鏈接地址: 線程間共享數(shù)據(jù)無需競爭

上一篇：LMAX Disruptor——一個高性能、低延遲且簡單的框架下一篇：LMAX架構(gòu)

線程間共享數(shù)據(jù)無需競爭

Disruptor 是什么？