国产一级a片免费看高清,亚洲熟女中文字幕在线视频,黄三级高清在线播放,免费黄色视频在线看

Disruptor

1. Disruptor是什么

1.1  技術(shù)背景

    LMAX是在英國(guó)注冊(cè)并受到FCA監(jiān)管（監(jiān)管號(hào)碼為509778）的外匯黃金交易所, LMAX架構(gòu)是LMAX內(nèi)部研發(fā)并應(yīng)用到交易系統(tǒng)的一種技術(shù)。它之所以引起人們的關(guān)注,是因?yàn)樗且粋€(gè)非常高性能系統(tǒng)，這個(gè)系統(tǒng)是建立在JVM平臺(tái)上，核心是一個(gè)業(yè)務(wù)邏輯處理器，官方號(hào)稱(chēng)它能夠在一個(gè)線(xiàn)程里每秒處理6百萬(wàn)訂單.

  一個(gè)僅僅部署在4臺(tái)服務(wù)器上的服務(wù)，每秒向Database寫(xiě)入數(shù)據(jù)超過(guò)100萬(wàn)行數(shù)據(jù)，每分鐘產(chǎn)生超過(guò)1G的數(shù)據(jù)。而每臺(tái)服務(wù)器（8核12G）上CPU占用不到100%，load不超過(guò)5。

1.2  對(duì)比阻塞隊(duì)列

 可以和BlockingQueue做對(duì)比，不過(guò)disruptor除了能完成同樣的工作場(chǎng)景外，能做更多的事，效率也更高。業(yè)務(wù)邏輯處理器完全是運(yùn)行在內(nèi)存中(in-memory)，使用事件源驅(qū)動(dòng)方式(event sourcing). 業(yè)務(wù)邏輯處理器的核心是Disruptors，這是一個(gè)并發(fā)組件，能夠在無(wú)鎖的情況下實(shí)現(xiàn)網(wǎng)絡(luò)的Queue并發(fā)操作。LMAX的研究表明，現(xiàn)在的所謂高性能研究方向似乎和現(xiàn)代CPU設(shè)計(jì)是相左的。

Thedisruptor component provides asynchronous SEDA behavior much as the standardSEDA Component, but utilizes a Disruptor instead of a BlockingQueue utilized bythe standard SEDA。

舉例部分典型的使用方式：

說(shuō)明： 2.0版本之后，Consumer的概念被EventProcessor代替，都是類(lèi)似事件處理消費(fèi)，下面的例子用Consumer表現(xiàn)，實(shí)際和EventProcessor一樣效果。

P=Producer,C=Consumer

所有的這些模式都可以在源碼里找到代碼例子，直接可參考

下面是官方給出的和ArrayBlockingQueue對(duì)比測(cè)試結(jié)果：

Comparative throughput (in ops per sec)

1.3  Disruptor構(gòu)成

先介紹幾個(gè)相關(guān)的核心概念。

①    環(huán)形隊(duì)列ringbuffer

數(shù)據(jù)緩沖區(qū)，不同線(xiàn)程之間傳遞數(shù)據(jù)的BUFFER。RingBuffer是存儲(chǔ)消息的地方，通過(guò)一個(gè)名為cursor的Sequence對(duì)象指示隊(duì)列的頭，協(xié)調(diào)多個(gè)生產(chǎn)者向RingBuffer中添加消息，并用于在消費(fèi)者端判斷RingBuffer是否為空。巧妙的是，表示隊(duì)列尾的Sequence并沒(méi)有在RingBuffer中，而是由消費(fèi)者維護(hù)。這樣的好處是多個(gè)消費(fèi)者處理消息的方式更加靈活，可以在一個(gè)RingBuffer上實(shí)現(xiàn)消息的單播，多播，流水線(xiàn)以及它們的組合。在RingBuffer中維護(hù)了一個(gè)名為gatingSequences的Sequence數(shù)組來(lái)跟蹤相關(guān)Seqence。

②    Producer/Consumer

Producer即生產(chǎn)者，比如下圖中的P1. 只是泛指調(diào)用 Disruptor 發(fā)布事件(我們把寫(xiě)入緩沖隊(duì)列的一個(gè)元素定義為一個(gè)事件)的用戶(hù)代碼。

Consumer和EventProcessor是一個(gè)概念，新的版本中由EventProcessor概念替代了Consumer。

有兩種實(shí)現(xiàn)策略，一個(gè)是SingleThreadedStrategy（單線(xiàn)程策略）另一個(gè)是MultiThreadedStrategy（多線(xiàn)程策略），兩種策略對(duì)應(yīng)的實(shí)現(xiàn)類(lèi)為SingleProducerSequencer、MultiProducerSequencer【都實(shí)現(xiàn)了Sequencer類(lèi)，之所以叫Sequencer是因?yàn)樗麄兌际峭ㄟ^(guò)Sequence來(lái)實(shí)現(xiàn)數(shù)據(jù)寫(xiě)，Sequence的概念參見(jiàn)③】 ，它們定義在生產(chǎn)者和消費(fèi)者之間快速、正確地傳遞數(shù)據(jù)的并發(fā)算法。具體使用哪個(gè)根據(jù)自己的場(chǎng)景來(lái)定，[多線(xiàn)程的策略使用了AtomicLong（Java提供的CAS操作），而單線(xiàn)程的使用long，沒(méi)有鎖也沒(méi)有CAS。這意味著單線(xiàn)程版本會(huì)非常快，因?yàn)樗挥幸粋€(gè)生產(chǎn)者，不會(huì)產(chǎn)生序號(hào)上的沖突]

Consumer即下圖中的C1,C2,C3，可以看到消費(fèi)者之間可以形成依賴(lài)關(guān)系。

老版：

（網(wǎng)絡(luò)配圖）

新版：

③Sequence

Sequence是一個(gè)遞增的序號(hào)，說(shuō)白了就是計(jì)數(shù)器；其次，由于需要在線(xiàn)程間共享，所以Sequence是引用傳遞，并且是線(xiàn)程安全的；再次，Sequence支持CAS操作；最后，為了提高效率，Sequence通過(guò)padding來(lái)避免偽共享，關(guān)于解決偽共享的問(wèn)題，可以參見(jiàn)下面章節(jié)詳細(xì)的介紹。

通過(guò)順序遞增的序號(hào)來(lái)編號(hào)管理通過(guò)其進(jìn)行交換的數(shù)據(jù)（事件），對(duì)數(shù)據(jù)(事件)的處理過(guò)程總是沿著序號(hào)逐個(gè)遞增處理。一個(gè) Sequence 用于跟蹤標(biāo)識(shí)某個(gè)特定的事件處理者(RingBuffer/Consumer )的處理進(jìn)度。生產(chǎn)者對(duì)RingBuffer的互斥訪問(wèn)，生產(chǎn)者與消費(fèi)者之間的協(xié)調(diào)以及消費(fèi)者之間的協(xié)調(diào)，都是通過(guò)Sequence實(shí)現(xiàn)。幾乎每一個(gè)重要的組件都包含Sequence。

說(shuō)明：雖然一個(gè)AtomicLong 也可以用于標(biāo)識(shí)進(jìn)度，但定義 Sequence 來(lái)負(fù)責(zé)該問(wèn)題還有另一個(gè)目的，那就是防止不同的 Sequence 之間的CPU緩存?zhèn)喂蚕?Flase Sharing)問(wèn)題。

④Sequence Barrier

用于保持對(duì)RingBuffer的main published Sequence 和Consumer依賴(lài)的其它Consumer的 Sequence 的引用。 Sequence Barrier 還定義了決定 Consumer 是否還有可處理的事件的邏輯。SequenceBarrier用來(lái)在消費(fèi)者之間以及消費(fèi)者和RingBuffer之間建立依賴(lài)關(guān)系。在Disruptor中，依賴(lài)關(guān)系實(shí)際上指的是Sequence的大小關(guān)系，消費(fèi)者A依賴(lài)于消費(fèi)者B指的是消費(fèi)者A的Sequence一定要小于等于消費(fèi)者B的Sequence，這種大小關(guān)系決定了處理某個(gè)消息的先后順序。因?yàn)樗邢M(fèi)者都依賴(lài)于RingBuffer，所以消費(fèi)者的Sequence一定小于等于RingBuffer中名為cursor的Sequence，即消息一定是先被生產(chǎn)者放到Ringbuffer中，然后才能被消費(fèi)者處理。不好理解的話(huà)，可以看下面章節(jié)事例配合了解。

SequenceBarrier在初始化的時(shí)候會(huì)收集需要依賴(lài)的組件的Sequence，RingBuffer的cursor會(huì)被自動(dòng)的加入其中。需要依賴(lài)其他消費(fèi)者和/或RingBuffer的消費(fèi)者在消費(fèi)下一個(gè)消息時(shí)，會(huì)先等待在SequenceBarrier上，直到所有被依賴(lài)的消費(fèi)者和RingBuffer的Sequence大于等于這個(gè)消費(fèi)者的Sequence。當(dāng)被依賴(lài)的消費(fèi)者或RingBuffer的Sequence有變化時(shí)，會(huì)通知SequenceBarrier喚醒等待在它上面的消費(fèi)者。

⑤Wait Strategy

當(dāng)消費(fèi)者等待在SequenceBarrier上時(shí)，有許多可選的等待策略，不同的等待策略在延遲和CPU資源的占用上有所不同，可以視應(yīng)用場(chǎng)景選擇：

• BusySpinWaitStrategy ： 自旋等待，類(lèi)似Linux Kernel使用的自旋鎖。低延遲但同時(shí)對(duì)CPU資源的占用也多。
• BlockingWaitStrategy ： 使用鎖和條件變量。CPU資源的占用少，延遲大。
• SleepingWaitStrategy ： 在多次循環(huán)嘗試不成功后，選擇讓出CPU，等待下次調(diào)度，多次調(diào)度后仍不成功，嘗試前睡眠一個(gè)納秒級(jí)別的時(shí)間再?lài)L試。這種策略平衡了延遲和CPU資源占用，但延遲不均勻。
• YieldingWaitStrategy ： 在多次循環(huán)嘗試不成功后，選擇讓出CPU，等待下次調(diào)。平衡了延遲和CPU資源占用，但延遲也比較均勻。
• PhasedBackoffWaitStrategy ： 上面多種策略的綜合，CPU資源的占用少，延遲大。

⑥    Event

在 Disruptor 的語(yǔ)義中，生產(chǎn)者和消費(fèi)者之間進(jìn)行交換的數(shù)據(jù)被稱(chēng)為事件(Event)。它不是一個(gè)被 Disruptor 定義的特定類(lèi)型，而是由 Disruptor 的使用者定義并指定。

⑦     EventProcessor

EventProcessor 持有特定消費(fèi)者(Consumer)的 Sequence，并提供用于調(diào)用事件處理實(shí)現(xiàn)的事件循環(huán)(Event Loop)。通過(guò)把EventProcessor提交到線(xiàn)程池來(lái)真正執(zhí)行，有兩類(lèi)Processor:

其中一類(lèi)消費(fèi)者是BatchEvenProcessor。每個(gè)BatchEvenProcessor有一個(gè)Sequence，來(lái)記錄自己消費(fèi)RingBuffer中消息的情況。所以，一個(gè)消息必然會(huì)被每一個(gè)BatchEvenProcessor消費(fèi)。

另一類(lèi)消費(fèi)者是WorkProcessor。每個(gè)WorkProcessor也有一個(gè)Sequence，多個(gè)WorkProcessor還共享一個(gè)Sequence用于互斥的訪問(wèn)RingBuffer。一個(gè)消息被一個(gè)WorkProcessor消費(fèi)，就不會(huì)被共享一個(gè)Sequence的其他WorkProcessor消費(fèi)。這個(gè)被WorkProcessor共享的Sequence相當(dāng)于尾指針

⑧     EventHandler

Disruptor定義的事件處理接口，由用戶(hù)實(shí)現(xiàn)，用于處理事件，是 Consumer 的真正實(shí)現(xiàn)。開(kāi)發(fā)者實(shí)現(xiàn)EventHandler，然后作為入?yún)鬟f給EventProcessor的實(shí)例。

綜上所述，附官方類(lèi)圖：

2. Disruptor什么時(shí)候用

Disruptor適用于兩個(gè)獨(dú)立的處理過(guò)程(兩個(gè)線(xiàn)程)之間交換數(shù)據(jù)。下面以?xún)蓚€(gè)簡(jiǎn)單場(chǎng)景舉例：

例如場(chǎng)景一：

           停車(chē)批量入場(chǎng)數(shù)據(jù)上報(bào)，數(shù)據(jù)上報(bào)后需要對(duì)每條入場(chǎng)數(shù)據(jù)存入DB，還需要發(fā)送kafka消息給其他業(yè)務(wù)系統(tǒng)。如果執(zhí)行完所有的操作，再返回，那么接口耗時(shí)比較長(zhǎng)，我們可以批量上報(bào)后驗(yàn)證數(shù)據(jù)正確性，通過(guò)后按單條入場(chǎng)數(shù)據(jù)寫(xiě)入環(huán)形隊(duì)列，然后直接返回成功。

實(shí)現(xiàn)方式一：啟 動(dòng)2個(gè)消費(fèi)者線(xiàn)程，一個(gè)消費(fèi)者去執(zhí)行db入庫(kù)，一個(gè)消費(fèi)者去發(fā)送kafka消息。

實(shí)現(xiàn)方式二：啟動(dòng)4個(gè)消費(fèi)者，2個(gè)消費(fèi)者并發(fā)執(zhí)行db入庫(kù)，兩個(gè)消費(fèi)者并發(fā)發(fā)送kafka消息，充分利用cpu多核特性，提高執(zhí)行效率。

實(shí)現(xiàn)方式三：如果要求寫(xiě)入DB和kafka后，需要給用戶(hù)發(fā)送短信。那么可以啟動(dòng)三個(gè)消費(fèi)者線(xiàn)程，一個(gè)執(zhí)行db插入，一個(gè)執(zhí)行kafka消息發(fā)布，最后一個(gè)依賴(lài)前兩個(gè)線(xiàn)程執(zhí)行成功，前兩個(gè)線(xiàn)程都執(zhí)行成功后，該線(xiàn)程執(zhí)行短信發(fā)送。

例如場(chǎng)景二：

你在網(wǎng)上使用信用卡下訂單。一個(gè)簡(jiǎn)單的零售系統(tǒng)將獲取您的訂單信息，使用信用卡驗(yàn)證服務(wù)，以檢查您的信用卡號(hào)碼，然后確認(rèn)您的訂單– 所有這些都在一個(gè)單一過(guò)程中操作。當(dāng)進(jìn)行信用卡有效性檢查時(shí)，服務(wù)器這邊的線(xiàn)程會(huì)阻塞等待，當(dāng)然這個(gè)對(duì)于用戶(hù)來(lái)說(shuō)停頓不會(huì)太長(zhǎng)。

在MAX架構(gòu)中，你將此單一操作過(guò)程分為兩個(gè)，第一部分將獲取訂單信息，然后輸出事件(請(qǐng)求信用卡檢查有效性的請(qǐng)求事件)給信用卡公司. 業(yè)務(wù)邏輯處理器將繼續(xù)處理其他客戶(hù)的訂單，直至它在輸入事件中發(fā)現(xiàn)了信用卡已經(jīng)檢查有效的事件，然后獲取該事件來(lái)確認(rèn)該訂單有效。

3.Disruptor為什么快

2.1數(shù)組實(shí)現(xiàn)

用數(shù)組實(shí)現(xiàn), 解決了鏈表節(jié)點(diǎn)分散, 不利于cache預(yù)讀問(wèn)題，可以預(yù)分配用于存儲(chǔ)事件內(nèi)容的內(nèi)存空間；并且解決了節(jié)點(diǎn)每次需要分配和釋放, 需要大量的垃圾回收GC問(wèn)題 (數(shù)組內(nèi)元素的內(nèi)存地址的連續(xù)性存儲(chǔ)的,在硬件級(jí)別，數(shù)組中的元素是會(huì)被預(yù)加載的,因?yàn)橹灰粋€(gè)元素被加載到緩存行，其他相鄰的幾個(gè)元素也會(huì)被加載進(jìn)同一個(gè)緩存行)

2.2求余操作優(yōu)化

               求余操作本身也是一種高耗費(fèi)的操作, 所以ringbuffer的size設(shè)成2的n次方, 可以利用位操作來(lái)高效實(shí)現(xiàn)求余。要找到數(shù)組中當(dāng)前序號(hào)指向的元素，可以通過(guò)mod操作，正常通過(guò)sequence mod array length = arrayindex，優(yōu)化后可以通過(guò)：sequence & （arraylength－1） = array index實(shí)現(xiàn)。比如一共有8槽，3&（8－1）=3，HashMap就是用這個(gè)方式來(lái)定位數(shù)組元素的，這種方式比取模的速度更快。

2.3 預(yù)讀與批量

相比鏈表隊(duì)列，實(shí)現(xiàn)數(shù)組預(yù)讀，減少結(jié)點(diǎn)操作空間釋放和申請(qǐng)，從而減少gc次數(shù)。生產(chǎn)者支持單生產(chǎn)，多生產(chǎn)者模式，單生產(chǎn)者cursor使用普通long實(shí)現(xiàn)，無(wú)鎖加快速度，多生產(chǎn)者才使用Sequence(AtomicLong)

 生產(chǎn)和消費(fèi)元素支持單線(xiàn)程批量操作數(shù)據(jù)。

2.4 Lock-Free

系統(tǒng)態(tài)的鎖會(huì)導(dǎo)致線(xiàn)程cache丟失.鎖競(jìng)爭(zhēng)的時(shí)候需要進(jìn)行仲裁. 這個(gè)仲裁會(huì)涉及到操作系統(tǒng)的內(nèi)核切換,并且在此過(guò)程中操作系統(tǒng)需要做一系列操作, 導(dǎo)致原有線(xiàn)程的指令緩存和數(shù)據(jù)緩很可能被丟掉

- 用戶(hù)態(tài)的鎖往往是通過(guò)自旋鎖來(lái)實(shí)現(xiàn)(自旋即忙等), 而自旋在競(jìng)爭(zhēng)激烈的時(shí)候開(kāi)銷(xiāo)是很大的(一直在消耗CPU資源)

disruptor不使用鎖, 使用CAS（Compare And Swap/Set）,嚴(yán)格意義上說(shuō)仍然是使用鎖, 因?yàn)镃AS本質(zhì)上也是一種樂(lè)觀鎖, 只不過(guò)是CPU級(jí)別指令, 不涉及到操作系統(tǒng), 所以效率很高(AtomicLong實(shí)現(xiàn)Sequence)

CAS說(shuō)明：

l   CAS依賴(lài)于處理器的支持, 當(dāng)然大部分現(xiàn)代處理器都支持.

l   CAS相對(duì)于鎖是非常高效的, 因?yàn)樗恍枰婕皟?nèi)核上下文切換進(jìn)行仲裁.

l   CAS并不是免費(fèi)的,它會(huì)涉及到對(duì)指令pipeline加鎖, 并且會(huì)用到內(nèi)存barrier（用來(lái)刷新內(nèi)存狀態(tài)，簡(jiǎn)單理解就是把緩存中，寄存器中的數(shù)據(jù)同步到內(nèi)存中去）

2.5 解決偽共享{False Sharing}

   Cpu cache簡(jiǎn)單示意圖：

 上面談到lock的耗費(fèi), 主要也是由于內(nèi)核的切換導(dǎo)致cache的丟失

所以cache是優(yōu)化的關(guān)鍵,cache越接近c(diǎn)ore就越快,也越小 。

其中L1，L2，L3等級(jí)緩存都是由緩存行組成的, 通常是64字節(jié), 一個(gè)Java的long類(lèi)型是8字節(jié)，因此在一個(gè)緩存行中可以存8個(gè)long類(lèi)型的變量. 緩存行是緩存更新的基本單位, 就算你只讀一個(gè)變量, 系統(tǒng)也會(huì)預(yù)讀其余7個(gè), 并cache這一行, 并且這行中的任一變量發(fā)生改變, 都需要重新加載整行, 而非僅僅重新加載一個(gè)變量.

 偽共享舉例：

比如在鏈表中往往會(huì)連續(xù)定義head和tail指針, 所以對(duì)于cache-line的預(yù)讀, 很有可能會(huì)導(dǎo)致head和tail在同一cache-line。在實(shí)際使用中, 往往producer線(xiàn)程會(huì)持續(xù)更改tail指針, 而consumer線(xiàn)程會(huì)持續(xù)更改head指針

當(dāng)producer線(xiàn)程和consumer線(xiàn)程分別被分配到core2和core1, 就會(huì)出現(xiàn)以下?tīng)顩r,由于core1不斷改變h, 導(dǎo)致該cache-line過(guò)期, 對(duì)于core2,雖然他不需要讀h, 或者t也沒(méi)有改變, 但是由于cache-line的整行更新, 所以core2仍然需要不停的更新它的cache,core2的緩存未命中被一個(gè)和它本身完全不相干的值h, 而被大大提高, 導(dǎo)致cache效率底下，而實(shí)際情況下, core1會(huì)不斷更新h, 而core2會(huì)不斷更新t, 導(dǎo)致core1和core2都需要頻繁的重新load cache, 這就是偽共享問(wèn)題

在Disruptor里我們對(duì)RingBuffer的cursor和BatchEventProcessor的序列進(jìn)行了緩存行填充，如下：

2.6 使用內(nèi)存屏障

內(nèi)存屏障本身不是一種優(yōu)化方式, 而是你使用lock-free(CAS)的時(shí)候, 必須要配合使用內(nèi)存屏障，因?yàn)镃PU和memory之間有多級(jí)cache,CPU core只會(huì)更新cache-line, 而cache-line什么時(shí)候flush到memory, 這個(gè)是有一定延時(shí)的 ,在這個(gè)延時(shí)當(dāng)中, 其他CPU core是無(wú)法得知你的更新的, 因?yàn)橹挥邪裞ache-line flush到memory后, 其他core中的相應(yīng)的cache-line才會(huì)被置為過(guò)期數(shù)據(jù),所以如果要保證使用CAS能保證線(xiàn)程間互斥, 即樂(lè)觀鎖, 必須當(dāng)一個(gè)core發(fā)生更新后, 其他所有core立刻知道并把相應(yīng)的cache-line設(shè)為過(guò)期, 否則在這些core上執(zhí)行CAS讀到的都是過(guò)期數(shù)據(jù).

內(nèi)存屏障 = “立刻將cache-lineflush到memory, 沒(méi)有延時(shí)”

注:可參考java中volatile的原理，同樣實(shí)現(xiàn)了內(nèi)存屏障。

4.使用Disruptor開(kāi)發(fā)

下面以車(chē)輛入場(chǎng)為例，入場(chǎng)后需要存入數(shù)據(jù)庫(kù)，需要發(fā)送kafka消息，兩步執(zhí)行完后，給用戶(hù)發(fā)送短信。代碼實(shí)現(xiàn)如下：(參見(jiàn)代碼運(yùn)行)

開(kāi)發(fā)步驟：

  1. 定義事件

  2. 定義事件工廠

  3. 定義事件處理的具體實(shí)現(xiàn)

4. 發(fā)布事件類(lèi)實(shí)現(xiàn)(Disruptor 要求 RingBuffer.publish 必須得到調(diào)用，如果發(fā)生異常也一樣要調(diào)用publish ，那么，很顯然這個(gè)時(shí)候需要調(diào)用者在事件處理的實(shí)現(xiàn)上來(lái)判斷事件攜帶的數(shù)據(jù)是否是正確的或者完整的)

5. 定義用于事件處理的線(xiàn)程池, 指定等待策略, 啟動(dòng) Disruptor,執(zhí)行完畢后關(guān)閉Disruptor

6. 關(guān)閉 Disruptor

5. 相關(guān)資料

 Disruptor源碼地址: https://github.com/LMAX-Exchange/disruptor