在ZMQ無鎖隊(duì)列的原理與實(shí)現(xiàn)一文中，我們已經(jīng)知道了ypipe可以實(shí)現(xiàn)一線程寫一線程讀的無鎖隊(duì)列，那么其劣勢就很明顯了，無法適應(yīng)多寫多讀的場景，因?yàn)槠湓谧x的時(shí)候沒有對(duì)r指針加鎖，在寫的時(shí)候沒有對(duì)w指針加鎖。那么如何實(shí)現(xiàn)一個(gè)多讀多寫的線程安全的無鎖隊(duì)列呢？

1. 互斥鎖：mutexqueue(太簡單不介紹了)
2. 互斥鎖+條件變量：blockqueue(太簡單不介紹了)
3. 內(nèi)存屏障：lockfreequeue(SimpleLockFreeQueue.h 暫時(shí)未寫文章介紹)
4. 雙重CAS原子操作：arraylockfreequeue(本文)

2. ArrayLockFreeQueue的類接?和變量

該程序使用 gcc 內(nèi)置的__sync_bool_compare_and_swap，但重新做了宏定義封裝。

所謂循環(huán)數(shù)組，就是RingBuffer

2.1 變量介紹

• m_count：隊(duì)列的元素個(gè)數(shù)
• m_writeIndex：新元素?列時(shí)存放位置在數(shù)組中的下標(biāo)
• m_readIndex：下?個(gè)出列元素在數(shù)組中的下標(biāo)
• m_maximumReadIndex：最后?個(gè)已經(jīng)完成?列操作的元素在數(shù)組中的下標(biāo)。如果它的值跟m_writeIndex不?致，表明有寫請(qǐng)求尚未完成。這意味著，有寫請(qǐng)求成功申請(qǐng)了空間但數(shù)據(jù)還沒完全寫進(jìn)隊(duì)列。所以如果有線程要讀取，必須要等到寫線程將數(shù)據(jù)完全寫?到隊(duì)列之后。

必須指明的是使?3種不同的下標(biāo)都是必須的，因?yàn)殛?duì)列允許任意數(shù)量的?產(chǎn)者和消費(fèi)者圍繞著它?作。已經(jīng)存在?種基于循環(huán)數(shù)組的?鎖隊(duì)列，使得唯?的?產(chǎn)者和唯?的消費(fèi)者可以良好的?作。它的實(shí)現(xiàn)相當(dāng)簡潔?常值得閱讀。

2.2 函數(shù)介紹

2.2.1 取余函數(shù)QUEUE_INT countToIndex(QUEUE_INT a_count);

這個(gè)函數(shù)非常有用，因?yàn)槲覀儗?shí)現(xiàn)的是循環(huán)隊(duì)列，所以一定要對(duì)數(shù)組長度取余。

隊(duì)列已滿判斷如下

隊(duì)列為空判斷如下

相關(guān)視頻推薦

C++無鎖隊(duì)列的設(shè)計(jì)與實(shí)現(xiàn)

高并發(fā)場景下，三種鎖方案：互斥鎖，自旋鎖，原子操作的優(yōu)缺點(diǎn)

需要C/C++ Linux服務(wù)器架構(gòu)師學(xué)習(xí)資料加qun812855908獲取（資料包括C/C++，Linux，golang技術(shù)，Nginx，ZeroMQ，MySQL，Redis，fastdfs，MongoDB，ZK，流媒體，CDN，P2P，K8S，Docker，TCP/IP，協(xié)程，DPDK，ffmpeg等），免費(fèi)分享

2.2.2 文字舉例

2.2.2.1 入隊(duì)函數(shù)bool enqueue(const ELEM_T &a_data);

下面以文字舉例說明函數(shù)，后續(xù)再圖示舉例。

假設(shè)現(xiàn)在有兩個(gè)線程都進(jìn)入了enqueue這個(gè)函數(shù)，m_writeIndex，m_readIndex和m_maximumReadIndex都為0。

• 線程1在第一個(gè)while和CAS處：

• 線程2在第一個(gè)while和CAS處：

• 線程2在第二個(gè)while和CAS處：

• 線程1在第二個(gè)while和CAS處：

• 線程2在第二個(gè)while和CAS處：

2.2.2.2 出隊(duì)函數(shù)bool dequeue(ELEM_T &a_data);

下面以文字舉例說明函數(shù)，后續(xù)再圖示舉例。

假設(shè)現(xiàn)在有兩個(gè)線程都進(jìn)入了dequeue這個(gè)函數(shù)，currentReadIndex為0,currentMaximumReadIndex為2。

• 線程1執(zhí)行

• 線程2執(zhí)行

如果沒有新數(shù)據(jù)寫入，再次讀取數(shù)據(jù)，則currentReadIndex(2)==currentMaximumReadIndex(2)相等，return false，沒有數(shù)據(jù)可讀。

2.2.3 圖示舉例

2.2.3.1 入隊(duì)函數(shù)bool enqueue(const ELEM_T &a_data);

下面的圖我就不分目錄了，直接一口氣說明完。

以下插圖展示了對(duì)隊(duì)列執(zhí)?操作時(shí)各下標(biāo)是如何變化的。如果?個(gè)位置被標(biāo)記為X，標(biāo)識(shí)這個(gè)位置?存放了數(shù)據(jù)。空?表示位置是空的。對(duì)于下圖的情況,隊(duì)列中存放了兩個(gè)元素。WriteIndex指示的位置是新元素將會(huì)被插?的位置。ReadIndex指向的位置中的元素將會(huì)在下?次pop操作中被彈出。

當(dāng)?產(chǎn)者準(zhǔn)備將數(shù)據(jù)插?到隊(duì)列中,它?先通過增加WriteIndex的值來申請(qǐng)空間。MaximumReadIndex指向最后?個(gè)存放有效數(shù)據(jù)的位置(也就是實(shí)際的隊(duì)列尾)。

?旦空間的申請(qǐng)完成,?產(chǎn)者就可以將數(shù)據(jù)拷?到剛剛申請(qǐng)到的位置中。完成之后增加MaximumReadIndex使得它與WriteIndex的?致。

現(xiàn)在隊(duì)列中有3個(gè)元素，接著?有?個(gè)?產(chǎn)者嘗試向隊(duì)列中插?元素。

在第?個(gè)?產(chǎn)者完成數(shù)據(jù)拷?之前，?有另外?個(gè)?產(chǎn)者申請(qǐng)了?個(gè)新的空間準(zhǔn)備拷?數(shù)據(jù)?，F(xiàn)在有兩個(gè)?產(chǎn)者同時(shí)向隊(duì)列插?數(shù)據(jù)。

現(xiàn)在?產(chǎn)者開始拷?數(shù)據(jù)，在完成拷?之后，對(duì)MaximumReadIndex的遞增操作必須嚴(yán)格遵循?個(gè)順序：第?個(gè)?產(chǎn)者線程?先遞增MaximumReadIndex，接著才輪到第?個(gè)?產(chǎn)者。這個(gè)順序必須被嚴(yán)格遵守的原因是，我們必須保證數(shù)據(jù)被完全拷?到隊(duì)列之后才允許消費(fèi)者線程將其出列。讓出cpu的?的也是為了讓排在最前?的?產(chǎn)者完成m_maximumReadIndex的更新

第?個(gè)?產(chǎn)者完成了數(shù)據(jù)拷?，并對(duì)MaximumReadIndex完成了遞增，現(xiàn)在第?個(gè)?產(chǎn)者可以遞增MaximumReadIndex了。

第?個(gè)?產(chǎn)者完成了對(duì)MaximumReadIndex的遞增,現(xiàn)在隊(duì)列中有5個(gè)元素。

2.2.3.2 出隊(duì)函數(shù)bool dequeue(ELEM_T &a_data);

以下插圖展示了元素出列的時(shí)候各種下標(biāo)是如何變化的,隊(duì)列中初始有2個(gè)元素。WriteIndex指示的位置是新元素將會(huì)被插?的位置。ReadIndex指向的位置中的元素將會(huì)在下?次pop操作中被彈出。

消費(fèi)者線程拷?數(shù)組ReadIndex位置的元素，然后嘗試?CAS操作將ReadIndex加1。如果操作成功消費(fèi)者成功的將數(shù)據(jù)出列。因?yàn)镃AS操作是原?的，所以只有唯?的線程可以在同?時(shí)刻更新ReadIndex的值。如果操作失敗，讀取新的ReadIndex值，以重復(fù)以上操作(copy數(shù)據(jù)，CAS)。

現(xiàn)在?有?個(gè)消費(fèi)者將元素出列，隊(duì)列變成空。

現(xiàn)在有?個(gè)?產(chǎn)者正在向隊(duì)列中添加元素。它已經(jīng)成功的申請(qǐng)了空間，但尚未完成數(shù)據(jù)拷?。任何其它企圖從隊(duì)列中移除元素的消費(fèi)者都會(huì)發(fā)現(xiàn)隊(duì)列?空(因?yàn)閣riteIndex不等于readIndex)。但它不能讀取readIndex所指向位置中的數(shù)據(jù)，因?yàn)閞eadIndex與MaximumReadIndex相等(相等break)。直到?產(chǎn)者完成數(shù)據(jù)拷?增加MaximumReadIndex的值才能讀取這個(gè)數(shù)據(jù)。

當(dāng)?產(chǎn)者完成數(shù)據(jù)拷?，隊(duì)列的??是1，消費(fèi)者線程可以讀取這個(gè)數(shù)據(jù)了。

2.2.4 計(jì)算隊(duì)列的大小size函數(shù)的ABA問題與解決方案

下面的場景描述了 size 為何會(huì)返回一個(gè)不正確的值:

解決方案：添加一個(gè)用于保存隊(duì)列中元素?cái)?shù)量的成員 count.這個(gè)成員可以通過 AtomicAdd/AtomicSub 來實(shí)現(xiàn)原子的遞增和遞減。

但需要注意的是這增加了一定開銷,因?yàn)樵舆f增,遞減操作比較昂貴也很難被編譯器優(yōu)化。如果可以容忍size函數(shù)的ABA問題，則可以不用count與AtomicAdd/AtomicSub。使用者可以根據(jù)自己的使用場合選擇是否承受額外的運(yùn)行時(shí)開銷。

2.2.5 與智能指針一起使用，內(nèi)存無法得到釋放

如果你打算用這個(gè)隊(duì)列來存放智能指針對(duì)象.需要注意,將一個(gè)智能指針存入隊(duì)列之后,如果它所占用的位置沒有被另一個(gè)智能指針覆蓋,那么它所指向的內(nèi)存是無法被釋放的(因?yàn)樗囊糜?jì)數(shù)器無法下降為 0).這對(duì)于一個(gè)操作頻繁的隊(duì)列來說沒有什么問題,但是程序員需要注意的是,一旦隊(duì)列被填滿過一次那么應(yīng)用程序所占用的內(nèi)存就不會(huì)下降,即使隊(duì)列被清空.除非自己做改動(dòng)，每次 pop 手動(dòng) delete。

3. 多個(gè)?產(chǎn)者線程的情況下yielding處理器的必要性

讀者可能注意到了enqueue函數(shù)中使?了sched_yield()來主動(dòng)的讓出處理器，對(duì)于?個(gè)聲稱?鎖的算法??，這個(gè)調(diào)?看起來有點(diǎn)奇怪。正如?章開始的部分解釋過的，多線程環(huán)境下影響性能的其中?個(gè)因素就是Cache損壞。?產(chǎn)?Cache損壞的?種情況就是?個(gè)線程被搶占，操作系統(tǒng)需要保存被搶占線程的上下?，然后將被選中作為下?個(gè)調(diào)度線程的上下?載?。此時(shí)Cache中緩存的數(shù)據(jù)都會(huì)失效，因?yàn)樗潜粨屨季€程的數(shù)據(jù)?不是新線程的數(shù)據(jù)。

所以，當(dāng)此算法調(diào)?sched_yield()意味著告訴操作系統(tǒng)：“我要把處理器時(shí)間讓給其它線程，因?yàn)槲乙却臣虑榈陌l(fā)?”。?鎖算法和通過阻塞機(jī)制同步的算法的?個(gè)主要區(qū)別在于?鎖算法不會(huì)阻塞在線程同步上，那么為什么在這?我們要主動(dòng)請(qǐng)求操作系統(tǒng)搶占??呢？這個(gè)問題的答案沒那么簡單。它與有多少個(gè)?產(chǎn)者線程在并發(fā)的往隊(duì)列中存放數(shù)據(jù)有關(guān)：每個(gè)?產(chǎn)者線程所執(zhí)?的CAS操作都必須嚴(yán)格遵循FIFO次序，?個(gè)?于申請(qǐng)空間，另?個(gè)?于通知消費(fèi)者數(shù)據(jù)已經(jīng)寫?完成可以被讀取了。

如果我們的應(yīng)?程序只有唯?的?產(chǎn)者操作這個(gè)隊(duì)列，sche_yield()將永遠(yuǎn)沒有機(jī)會(huì)被調(diào)?，第2個(gè)CAS操作永遠(yuǎn)不會(huì)失敗。因?yàn)樵?個(gè)?產(chǎn)者的情況下沒有?能破壞?產(chǎn)者執(zhí)?這兩個(gè)CAS操作的FIFO順序。

?當(dāng)多于?個(gè)?產(chǎn)者線程往隊(duì)列中存放數(shù)據(jù)的時(shí)候，問題就出現(xiàn)了。概括來說，?個(gè)?產(chǎn)者通過第1個(gè)CAS操作申請(qǐng)空間，然后將數(shù)據(jù)寫?到申請(qǐng)到的空間中，然后執(zhí)?第2個(gè)CAS操作通知消費(fèi)者數(shù)據(jù)準(zhǔn)備完畢可供讀取了。這第2個(gè)CAS操作必須遵循FIFO順序，也就是說，如果A線程第?先執(zhí)?完第?個(gè)CAS操作，那么它也要第1個(gè)執(zhí)?完第2個(gè)CAS操作，如果A線程在執(zhí)?完第?個(gè)CAS操作之后停?，然后B線程執(zhí)?完第1個(gè)CAS操作，那么B線程將?法完成第2個(gè)CAS操作，因?yàn)樗却鼳先完成第2個(gè)CAS操作。?這就是問題產(chǎn)?的根源。讓我們考慮如下場景，3個(gè)消費(fèi)者線程和1個(gè)消費(fèi)者線程：

• 線程1，2，3按順序調(diào)?第1個(gè)CAS操作申請(qǐng)了空間。那么它們完成第2個(gè)CAS操作的順序也應(yīng)該與這個(gè)順序?致，1，2，3。
• 線程2?先嘗試執(zhí)?第2個(gè)CAS，但它會(huì)失敗，因?yàn)榫€程1還沒完成它的第2此CAS操作呢。同樣對(duì)于線程3也是?樣的。
• 線程2和3將會(huì)不斷的調(diào)?它們的第2個(gè)CAS操作，直到線程1完成它的第2個(gè)CAS操作為?。
• 線程1最終完成了它的第2個(gè)CAS，現(xiàn)在線程3必須等線程2先完成它的第2個(gè)CAS。
• 線程2也完成了，最終線程3也完成。

在上?的場景中，?產(chǎn)者可能會(huì)在第2個(gè)CAS操作上?旋?段時(shí)間，?于等待先于它執(zhí)?第1個(gè)CAS操作的線程完成它的第2次CAS操作。在?個(gè)物理處理器數(shù)量?于操作隊(duì)列線程數(shù)量的系統(tǒng)上，這不會(huì)有太嚴(yán)重的問題：因?yàn)槊總€(gè)線程都可以分配在??的處理器上執(zhí)?，它們最終都會(huì)很快完成各?的第2次CAS操作。雖然算法導(dǎo)致線程處理忙等狀態(tài)，但這正是我們所期望的，因?yàn)檫@使得操作更快的完成。也就是說在這種情況下我們是不需要sche_yield()的，它完全可以從代碼中刪除。

但是，在?個(gè)物理處理器數(shù)量少于線程數(shù)量的系統(tǒng)上，sche_yield()就變得?關(guān)重要了。讓我們?cè)俅慰疾樯?3個(gè)線程的場景，當(dāng)線程3準(zhǔn)備向隊(duì)列中插?數(shù)據(jù)：如果線程1在執(zhí)?完第1個(gè)CAS操作，在執(zhí)?第2個(gè)CAS操作之前被搶占，那么線程2，3就會(huì)?直在它們的第2個(gè)CAS操作上忙等(它們忙等，不讓出處理器，線程1也就沒機(jī)會(huì)執(zhí)?，它們就只能繼續(xù)忙等，也就是說，如果不適用 sched_yield，一直自旋，那么可能多個(gè)線程同時(shí)阻塞在第二個(gè) CAS 那兒)，直到線程1重新被喚醒，完成它的第2個(gè)CAS操作。這就是需要sche_yield()的場合了，操作系統(tǒng)應(yīng)該避免讓線程2，3處于忙等狀態(tài)。它們應(yīng)該盡快的讓出處理器讓線程1執(zhí)?，使得線程1可以把它的第2個(gè)CAS操作完成。這樣線程2和3才能繼續(xù)完成它們的操作。

4. 循環(huán)數(shù)組無鎖隊(duì)列的性能測試

4.1 性能測試

互斥鎖隊(duì)列 vs 互斥鎖+條件變量隊(duì)列 vs 內(nèi)存屏障鏈表 vs RingBuffer CAS 實(shí)現(xiàn)。

4寫1讀：性能中等

4寫4讀：性能中上

1寫4讀：性能最好

7寫7讀：性能比互斥鎖隊(duì)列還差

4.2 分析

雖然沒有分析第三個(gè)內(nèi)存屏障鏈表的代碼，但是我們不難看出互斥鎖+條件變量 與 內(nèi)存屏障鏈表 的性能差別不大。為什么呢？鏈表的方式需要不斷的申請(qǐng)和釋放元素。當(dāng)然，用內(nèi)存池可以適當(dāng)改善這個(gè)影響，但是內(nèi)存池在分配內(nèi)存與釋放內(nèi)存的時(shí)候也會(huì)涉及到線程間的數(shù)據(jù)競爭，所以用鏈表的方式性能相對(duì)提升不多。

隨著生產(chǎn)者數(shù)量的增加,無鎖隊(duì)列的效率迅速下降。因?yàn)樵诙鄠€(gè)生產(chǎn)者的情況下，第 2 個(gè) CAS 將對(duì)性能產(chǎn)生影響。我們通過測試得出循環(huán)數(shù)組的?鎖隊(duì)列在1寫4讀的場景下性能提升是最高的，因?yàn)橹挥幸粋€(gè)生產(chǎn)者，那么第二個(gè)CAS不會(huì)有yield的情況出現(xiàn)。由此我們可以得出一個(gè)結(jié)論，在一寫多讀的場景，我們可以優(yōu)先使用循環(huán)數(shù)組的?鎖隊(duì)列，比如下圖的場景。