窺探 Netty 源碼！Netty 中的 MpscLinkedQueue 究竟是什么鬼？

本文選自 Doocs 開源社區(qū)旗下“源碼獵人”項目，作者 tydhot。

項目將會持續(xù)更新，歡迎 Star 關(guān)注。

項目地址：https://github.com/doocs/source-code-hunter

該文所涉及的 netty 源碼版本為 4.1.6。

MpscLinkedQueue 是什么

在 Netty 的核心中的核心成員 NioEventLoop 中，其中任務(wù)隊列的實現(xiàn) taskQueue 便是 MpscLinkedQueue。MpscLinkedQueue 是 Netty 所實現(xiàn)的一個基于多生產(chǎn)者單消費者的無鎖隊列，針對 NioEventLoop 中任務(wù)隊列的特點，其單消費者的場景在一開始就避免了從隊列中取數(shù)據(jù)時加鎖的必要，而其最精妙的地方便是在多生產(chǎn)者并發(fā)從隊列中添加數(shù)據(jù)的時候也沒有加鎖，達到 Netty 所期望的高性能實現(xiàn)。這是如何實現(xiàn)的？

MpscLinkedQueue 無鎖并發(fā)線程安全寫入原理

MpscLinkedQueue 對于尾結(jié)點的維護

首先，MpscLinkedQueue 繼承自 AtomicReference，也就是說 MpscLinkedQueue 通過繼承自 AtomicReference 的方式，顯式地維護了一個提供原子讀寫能力的變量 value。而在 MpscLinkedQueue 中，這個 value 是其內(nèi)部維護的隊列的尾結(jié)點。

MpscLinkedQueue 對于頭結(jié)點的維護

而后，來看 MpscLinkedQueue 的構(gòu)造方法。

MpscLinkedQueue() {    MpscLinkedQueueNode<E> tombstone = new DefaultNode<E>(null);    headRef = new FullyPaddedReference<MpscLinkedQueueNode<E>>();    headRef.set(tombstone);    setTail(tombstone);}

在 MpscLinkedQueue 中，維護著 headRef 頭結(jié)點字段，其隊列內(nèi)部節(jié)點的實現(xiàn)是一個 MpscLinkedQueueNode。MpscLinkedQueueNode 是一個除了存放具體隊列元素外只有 next 字段的節(jié)點，也就是說，MpscLinkedQueue 的隊列是單向的。在構(gòu)造方法的最后，通過 setTail()方法的，將 MpscLinkedQueue 的尾結(jié)點字段 value 也設(shè)置為頭結(jié)點。MpscLinkedQueue 的頭結(jié)點字段 headRef 的存在可以方便后續(xù)直接從頭結(jié)點開始的隊列操作，消費者可以簡單判斷頭尾節(jié)點是否相等來確認隊列中是否有元素可以消費。

MpscLinkedQueue 如何做到線程安全的無鎖加入

@Override@SuppressWarnings("unchecked")public boolean offer(E value) {    if (value == null) {        throw new NullPointerException("value");    }    final MpscLinkedQueueNode<E> newTail;    if (value instanceof MpscLinkedQueueNode) {        newTail = (MpscLinkedQueueNode<E>) value;        newTail.setNext(null);    } else {        newTail = new DefaultNode<E>(value);    }    MpscLinkedQueueNode<E> oldTail = replaceTail(newTail);    oldTail.setNext(newTail);    return true;}private MpscLinkedQueueNode<E> replaceTail(MpscLinkedQueueNode<E> node) {    return getAndSet(node);}

MpscLinkedQueue 的 offer()方法很簡短，但是恰恰就是整個添加隊列元素加入的流程，當(dāng)元素被加入的時候，首先判斷加入的元素是否是 MpscLinkedQueueNode，如果不是則進行封裝。之后便是整個操作的重點：

通過 replaceTail()方法，將當(dāng)前被加入的節(jié)點通過 AtomicReference 所提供的 getAndSet()方法將其設(shè)為隊列的尾結(jié)點，并返回先前的尾結(jié)點。這次操作由 UNSAFE 的 CAS 來保證操作的原子性。

之后將之前的尾結(jié)點的 next 指向新加入的節(jié)點，本次加入宣告結(jié)束。

整個操作就到此結(jié)束，這里可以看出，MpscLinkedQueue 利用了 AtomicReference 底層 UNSAFE 的能力，通過 CAS 確保新設(shè)置進入 value 的節(jié)點必定能夠和原先的節(jié)點達成一個且唯一的聯(lián)系，那么只需要自頂向下不斷通過將這個聯(lián)系變成引用，那么一條隊列便形成了。由于其實現(xiàn)是鏈表而不是數(shù)組，也就沒有涉及到資源的競爭，在不加鎖的前提下其隊列順序可能不會嚴格按照加入順序，但這在當(dāng)前場景下并不是問題。在這個前提，高并發(fā)的插入場景下，每個新進入的新節(jié)點都將獲取原尾位置 value 上的節(jié)點，而自身將會被設(shè)置為其后驅(qū)節(jié)點重新放到尾結(jié)點位置上，CAS 在不加鎖的前提下保證了前后節(jié)點對應(yīng)關(guān)系的唯一性，完成了并發(fā)條件下不加鎖的線程安全寫入。

MpscLinkedQueue 不支持 remove()

在 MpscLinkedQueue 中，是不支持 remove()的方法去從隊列中移除任意一個元素的。原因很簡單，消費者和生產(chǎn)者是無鎖的，消費者可以通過比較隊首和隊尾元素是否一致來保證線程安全地從隊首取數(shù)據(jù)，但是 remove()從隊列中任意位置修改數(shù)據(jù)是線程不安全的，主要體現(xiàn)在移除隊尾元素可能會導(dǎo)致正在加入的新元素被丟棄。

MpscLinkedQueue 另外的實現(xiàn)細節(jié)

MpscLinkedQueue 中的頭節(jié)點被通過 FullyPaddedReference 封裝。其內(nèi)部前后分別填充 56 字節(jié)和 64 字節(jié)來進行填充以避免偽共享導(dǎo)致的性能損耗，使得其頭結(jié)點可以高效被訪問。關(guān)于偽共享的相關(guān)知識可以通過搜索引擎進行查詢。

MpscLinkedQueue 在消費者消費數(shù)據(jù)后，當(dāng)將下一個節(jié)點設(shè)置為頭結(jié)點的時候，并不是直接進行賦值，而是通過 UNSAFE 來根據(jù)偏移量賦值，這樣做將略微提高性能，主要是內(nèi)存屏障 storestrore 和 loadstrore 之間的性能差異。

全文完！

希望本文對大家有所幫助。如果感覺本文有幫助，有勞轉(zhuǎn)發(fā)或點一下“在看”！讓更多人收獲知識！

推薦閱讀

?窺探 Netty 源碼！Recycler 對象池實現(xiàn)原理剖析?窺探 Netty 源碼！FastThreadLocal 究竟快在哪里？?牛逼！一文看懂 Java 并發(fā)編程在各主流框架中的應(yīng)用

長按識別下圖二維碼，關(guān)注公眾號「Doocs 開源社區(qū)」，第一時間跟你們分享好玩、實用的技術(shù)文章與業(yè)內(nèi)最新資訊。