Disruptor高性能之道-無鎖化
JUC中的隊列BlockingQueue是通過加鎖實現(xiàn)對生產(chǎn)者和消費者的協(xié)調(diào)。
加鎖就意味著需要犧牲高性能，換來線程安全。
有沒有辦法既能高性能，還能線程安全？
Disruptor給出的答案是，“無鎖”。
?無鎖，并不是完全消除鎖，而是指沒有OS層面的鎖。
Disruptor通過CAS（Compare And Swap）指令實現(xiàn)了無鎖化。具體的指令是cmpxchg，本文會做簡單講解。感興趣的讀者可以自行搜索資料了解詳細內(nèi)容。
?
簡單解釋下CAS具體干了什么事情。
?CAS, 比較并交換，Compare And Swap。顧名思義，就是通過比較值是否發(fā)生變化，決定是否要重新賦值。如果在操作期間，值沒有被其他線程操作，那么就將期望的值賦值給它，否則發(fā)現(xiàn)期望的值與舊值不等，說明值已經(jīng)變更，則不執(zhí)行操作，返回操作失敗。
?
舉個栗子
比如說，舊值oldValue為1，期望的值expectValue為1，新值newValue為2。如果沒有其他線程修改舊值，那么
expectValue == oldValue 將newValue寫入，當前值為2
如果在操作過程中，oldValue被其他線程操作修改為2，那么當前線程的expectValue（1）與oldValue（2）比較就不等，寫入失敗。
Disruptor如何進行CAS
我們知道Disruptor核心數(shù)據(jù)結(jié)構(gòu)為RingBuffer，Disruptor為RingBuffer分配了一個Sequence對象，用于標識RingBuffer的頭和尾，這個標識不是通過指針實現(xiàn)的，而是通過序號。
?這個序號也就是Sequence。
?
雖然邏輯上RingBuffer是一個環(huán)形數(shù)組，但是在內(nèi)存中是以一個普通的數(shù)組形式存在的。
RingBuffer中通過對比序號的方式對生產(chǎn)者和消費者間的資源進行協(xié)調(diào)。
每當生產(chǎn)者要往隊列中加入新數(shù)據(jù)，生產(chǎn)者都會將當前sequence + 準備加入隊列的數(shù)據(jù)量，與消費者所在位置進行比較，以判斷是否存在足夠的空間放這些數(shù)據(jù)，而不至于覆蓋掉消費者沒有消費的數(shù)據(jù)。
用體育術(shù)語就叫“套圈”。
如圖所示：ringBufferSize=16，生產(chǎn)者當前sequence指向20，消費者sequence指向27。
我們簡單計算一下這個場景下，生產(chǎn)者能否繼續(xù)寫入4個元素。
對于消費者而言，27 MOD 16 = 11 對于生產(chǎn)者而言，20 + 4 = 24（預(yù)計寫入的最大序號），24 MOD 16 = 8 生產(chǎn)者若成功寫入4個元素，則sequence指向數(shù)組的第8個位置，8 < 11, 表明生產(chǎn)者沒有覆蓋消費者的進度。 生產(chǎn)者不需要等待消費者，直接生產(chǎn)數(shù)據(jù)即可。而且并不會覆蓋消費者未處理完的數(shù)據(jù)。
實際上，Disruptor的代碼實現(xiàn)就是通過compareAndSet方法實現(xiàn)了CAS無鎖化操作。
????/**
?????*?Atomically?add?the?supplied?value.
?????*
?????*?@param?increment?The?value?to?add?to?the?sequence.
?????*?@return?The?value?after?the?increment.
?????*/
????public?long?addAndGet(final?long?increment)
????{
????????long?currentValue;
????????long?newValue;
????????do
????????{
????????????currentValue?=?get();
????????????newValue?=?currentValue?+?increment;
????????}
????????while?(!compareAndSet(currentValue,?newValue));
????????return?newValue;
????}
我們可以看到，這里通過while循環(huán)不斷嘗試CAS操作，如果CAS操作不成功就會自旋重試，這個操作并沒有使用OS層面的鎖，因此效率要大幅高于OS層面的鎖機制（管程）。
addAndGet調(diào)用了compareAndSet方法：
????/**
?????*?Perform?a?compare?and?set?operation?on?the?sequence.
?????*
?????*?@param?expectedValue?The?expected?current?value.
?????*?@param?newValue?The?value?to?update?to.
?????*?@return?true?if?the?operation?succeeds,?false?otherwise.
?????*/
????public?boolean?compareAndSet(final?long?expectedValue,?final?long?newValue)
????{
????????return?UNSAFE.compareAndSwapLong(this,?VALUE_OFFSET,?expectedValue,?newValue);
????}
可以看到最終是調(diào)用了UNSAFE的compareAndSwapLong方法，該方法為native方法，在JVM層面調(diào)用了CAS指令。
CAS指令
上文我們提到，Disruptor的CAS最終調(diào)用的是CPU層面的機器指令「cmpxchg」。
該指令的詳細描述：
??compxchg?[ax]?(隱式參數(shù)，EAX?累加器),?
???????????[bx]?(源操作數(shù)地址),?
???????????[cx]?(目標操作數(shù)地址)
簡單解釋下：
cmpxchg指令有三個操作數(shù)，操作數(shù)ax不在指令里面出現(xiàn)，是一個隱式的操作數(shù)，準確地說它是EAX累加寄存器里面的值。 操作數(shù)bx是源操作數(shù)，指令會對比這個操作數(shù)和上面的累加寄存器里面的值是否相等，如果相等 CPU 會把 ZF（也就是條件碼寄存器里面零標志位的值）設(shè)置為 1，然后再把操作數(shù)cx（也就是目標操作數(shù)）設(shè)置到源操作數(shù)的地址上。 如果不相等的話，就把源操作數(shù)里面的值設(shè)置到累加器寄存器里面
由于cmpxchg是cpu級別的指令，因此直接調(diào)用就可以保證cas操作的原子性。
由于去除了OS層面的鎖，即便CAS存在比較操作與自旋操作，其本質(zhì)也是無鎖化操作，這種無鎖化機制消除了上下文切換，對于CPU極為友好，因此運行效率很快。
事實上，在JUC包中，也提供了大量的CAS相關(guān)工作類方便我們操作，這些類一般以atomic開頭，如果去研究其實現(xiàn)，我們同樣會發(fā)現(xiàn)最終是通過UNSAFE調(diào)用了底層的CAS實現(xiàn)，實現(xiàn)無鎖化操作，減少上下文切換，提升代碼運行速率。
?加鎖導致的上下文切換之所以會顯著影響代碼運行速度，主要原因在于獲取鎖的過程中，CPU需要等待OS層面的鎖競爭結(jié)果，對于沒有獲取鎖的線程需要進行掛起，此時就需要進行上下文切換。
上下文切換會把掛起線程的寄存器中的數(shù)據(jù)放到線程棧，該操作會導致加載到高速緩存中的數(shù)據(jù)也失效，進而導致程序運行速率比無鎖更慢。
?
CAS就沒有什么問題么？
當然CAS操作同樣也會存在缺點，那就是由于CAS操作本身需要進行對比，如果不相等則會一直自旋（busy-wait），這樣的操作會使得cpu的負載升高，全功率滿負荷運行。