一、概述

ConcurrentLinkedQueue 是 Java 并發(fā)包（java.util.concurrent）中提供的無界非阻塞線程安全隊(duì)列，基于單向鏈表實(shí)現(xiàn)，采用 CAS（Compare-and-Swap） 操作和 無鎖算法 保證并發(fā)安全。其核心設(shè)計(jì)目標(biāo)是高吞吐量和低延遲，適用于高并發(fā)場景下的生產(chǎn)者-消費(fèi)者模型。

關(guān)鍵特性

1. 無界性：理論上容量無限，但受內(nèi)存限制。
2. 非阻塞：操作永不阻塞線程，失敗立即返回（如 poll() 在隊(duì)列為空時(shí)返回 null）。
3. 無鎖設(shè)計(jì)：通過 CAS 和自旋機(jī)制實(shí)現(xiàn)線程安全，避免傳統(tǒng)鎖的開銷。
4. FIFO 順序：嚴(yán)格遵循先進(jìn)先出原則。
5. 弱一致性迭代器：遍歷時(shí)可能看到部分更新，但不會(huì)拋出 ConcurrentModificationException。

二、內(nèi)部數(shù)據(jù)結(jié)構(gòu)

1. 節(jié)點(diǎn)類Node<E>

private static class Node<E> {
    volatile E item;          // 存儲(chǔ)元素
    volatile Node<E> next;    // 指向下一個(gè)節(jié)點(diǎn)
    // CAS 操作方法
    boolean casItem(E cmp, E val) { ... }
    boolean casNext(Node<E> cmp, Node<E> val) { ... }
}

• volatile 修飾：保證多線程可見性。
• CAS 操作：通過 Unsafe 類實(shí)現(xiàn)原子性更新。

2. 隊(duì)列指針

• head：指向隊(duì)列頭部（可能滯后于實(shí)際頭節(jié)點(diǎn)）。
• tail：指向隊(duì)列尾部（可能滯后于實(shí)際尾節(jié)點(diǎn)）。
• 哨兵節(jié)點(diǎn)：初始化時(shí) head 和 tail 均指向一個(gè) item=null 的哨兵節(jié)點(diǎn)。

三、核心方法與實(shí)現(xiàn)原理

1.入隊(duì)操作（offer/add）

public boolean offer(E e) {
    checkNotNull(e);
    Node<E> newNode = new Node<>(e);
    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        if (q == null) {  // p 是尾節(jié)點(diǎn)
            if (p.casNext(null, newNode)) {
                if (p != t) casTail(t, newNode);  // 更新 tail
                return true;
            }
        } else if (p == q) {  // 自引用節(jié)點(diǎn)，重置 tail
            p = (t != (t = tail)) ? t : head;
        } else {
            p = (p != t && t != (t = tail)) ? t : q;
        }
    }
}

• CAS 競爭：多個(gè)線程可能同時(shí)嘗試插入，僅一個(gè)成功。
• tail 滯后更新：僅在 p != t 時(shí)更新 tail，減少 CAS 操作頻率。

2.出隊(duì)操作（poll）

public E poll() {
    restartFromHead:
    for (;;) {
        for (Node<E> h = head, p = h, q;;) {
            E item = p.item;
            if (item != null && p.casItem(item, null)) {
                if (p != h) updateHead(h, p);  // 更新 head
                return item;
            } else if ((q = p.next) == null) {
                updateHead(h, p);
                return null;
            } else if (p == q) continue restartFromHead;
            else p = q;
        }
    }
}

• CAS 移除：將頭節(jié)點(diǎn)的 item 設(shè)為 null，延遲物理刪除。
• head 更新：若 p != h，則更新 head 指針。

3.其他方法

• peek()：獲取頭元素但不移除，邏輯與 poll() 類似，不修改 item。
• size()：遍歷鏈表統(tǒng)計(jì)元素?cái)?shù)，非線程安全（高并發(fā)下結(jié)果可能不準(zhǔn)確）。
• remove(Object o)：遍歷鏈表移除首個(gè)匹配元素，返回是否成功。

四、無鎖并發(fā)控制機(jī)制

1. CAS 操作

• 原子性更新：通過 Unsafe.compareAndSwapObject 實(shí)現(xiàn)對(duì) item 和 next 的原子修改。
• 自旋重試：CAS 失敗時(shí)循環(huán)重試，而非阻塞線程。

2. 松弛不變量（Relaxed Invariants）

• head 滯后：可能指向已刪除節(jié)點(diǎn)，僅在必要時(shí)更新（如遍歷時(shí)遇到自引用節(jié)點(diǎn)）。
• tail 滯后：減少更新頻率，提升吞吐量。

3. 自引用節(jié)點(diǎn)

• 標(biāo)記刪除：出隊(duì)后，原頭節(jié)點(diǎn)的 next 指向自身，防止其他線程誤用。
• 垃圾回收：物理刪除由 GC 處理，避免頻繁內(nèi)存操作。

五、適用場景

1. 高并發(fā)生產(chǎn)者-消費(fèi)者模型：如日志處理、實(shí)時(shí)事件分發(fā)。
2. 低延遲系統(tǒng)：如高頻交易、游戲服務(wù)器。
3. 無界緩沖需求：需動(dòng)態(tài)擴(kuò)展隊(duì)列長度的場景。
4. 弱一致性要求：允許短暫的數(shù)據(jù)不一致（如遍歷時(shí)）。

與LinkedBlockingQueue對(duì)比

六、最佳實(shí)踐

1. 避免頻繁調(diào)用 size()：高并發(fā)下性能差，建議通過外部計(jì)數(shù)器統(tǒng)計(jì)。
2. 合理預(yù)估容量：雖無界，但內(nèi)存耗盡可能引發(fā) OOM。
3. 結(jié)合其他同步機(jī)制：如需精確控制，可與 Semaphore 或 CountDownLatch 聯(lián)用。
4. 弱一致性遍歷：接受遍歷時(shí)可能遺漏新元素，適用于非強(qiáng)一致性場景。

七、源碼設(shè)計(jì)細(xì)節(jié)

1. 哨兵節(jié)點(diǎn)：初始化時(shí) head 和 tail 指向同一個(gè)哨兵節(jié)點(diǎn)，簡化邊界條件處理。
2. 自引用節(jié)點(diǎn)：出隊(duì)后原頭節(jié)點(diǎn) next=self，標(biāo)記為待回收。
3. CAS 優(yōu)化：節(jié)點(diǎn)構(gòu)造時(shí)使用 Unsafe.putObject 替代 volatile 寫操作，減少內(nèi)存屏障開銷。

八、總結(jié)

ConcurrentLinkedQueue 是 Java 并發(fā)編程中高性能無鎖隊(duì)列的典范，通過 CAS 和松弛不變量設(shè)計(jì)，在保證線程安全的同時(shí)最大化吞吐量。適用于對(duì)延遲敏感、無需嚴(yán)格容量控制的場景，但需注意其無界特性可能帶來的內(nèi)存風(fēng)險(xiǎn)。理解其底層機(jī)制（如 CAS、自旋、自引用節(jié)點(diǎn)）有助于在實(shí)際工程中合理應(yīng)用。

到此這篇關(guān)于Java中的ConcurrentLinkedQueue的使用小結(jié)的文章就介紹到這了,更多相關(guān)Java ConcurrentLinkedQueue內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論