1. 併發容器的歷史

大家好，今天我們來聊一個 Java 多線程開發中繞不開的核心話題：併發容器。可能你已經發現，當我們在多線程環境中使用 HashMap、ArrayList 這些集合類時，經常會遇到ConcurrentModificationException或數據不一致的問題，這就是因為這些普通集合類不是線程安全的。

JDK 提供的傳統解決方案是Collections.synchronizedXxx()方法，比如：

Map<String, String> synchronizedMap = Collections.synchronizedMap(new HashMap<>());
List<String> synchronizedList = Collections.synchronizedList(new ArrayList<>());

但這種方式有一個致命問題：它使用的是"一刀切"的粗粒度同步策略，導致多個線程競爭同一把鎖，性能非常低。隨着 Java 5 引入的 java.util.concurrent 包（簡稱 JUC），我們有了更高效的併發容器選擇。

下面我們圍繞三個核心問題展開：

1. 併發容器相比傳統同步容器有哪些優勢？
2. 各種併發容器的內部實現原理是怎樣的？
3. 如何在實際項目中正確選擇並高效使用這些容器？

2. ConcurrentHashMap 深度解析

2.1 進化歷程：從分段鎖到 CAS+紅黑樹

ConcurrentHashMap 是併發編程中使用最廣泛的 Map 實現，它的設計經歷了重大變革：

graph LR
    A[JDK 1.5-1.6] --> |分段鎖機制 Segment| B[JDK 1.7]
    B --> |CAS+紅黑樹| C[JDK 1.8+]

• JDK 1.7 版本：採用分段鎖(Segment)機制，將數據分成 16 個段，每段一把鎖，本質上是可重入鎖(ReentrantLock)
• JDK 1.8 版本：拋棄了 Segment 設計，改用 CAS+synchronized+紅黑樹的設計

讓我們通過一個圖解來理解這兩個版本的區別：

鏈表到紅黑樹的轉換機制

JDK 1.8 引入了一個重要的優化：當鏈表長度超過特定閾值時，會將鏈表轉換為紅黑樹，顯著提升查找性能。

• 鏈表轉紅黑樹閾值：默認為 8，當鏈表節點數超過 8 時轉為紅黑樹
• 紅黑樹退化為鏈表閾值：默認為 6，當節點數小於 6 時轉回鏈表
• 為什麼選擇 8 作為閾值？根據泊松分佈統計，鏈表長度超過 8 的概率不到千萬分之一，這是一種時間和空間的權衡，避免大多數情況下不必要的紅黑樹複雜度

2.2 源碼解析：put 和 get 操作的實現

讓我們看看 JDK 1.8 中 ConcurrentHashMap 的核心方法實現（簡化版）：

put 操作的關鍵步驟：

final V putVal(K key, V value, boolean onlyIfAbsent) {
    // 不允許空鍵值
    if (key == null || value == null) throw new NullPointerException();

    // 計算hash值
    int hash = spread(key.hashCode());
    int binCount = 0;

    // 無限循環，確保CAS操作成功
    for (Node<K,V>[] tab = table;;) {
        // 省略一些初始化代碼...

        // 如果對應桶為空，通過CAS操作創建新節點
        if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null)))
                break; // 成功則跳出循環
        }
        // 省略其他邏輯...
        else {
            // 桶不為空，鎖定當前桶的首節點
            synchronized (f) {
                // 再次檢查（雙重檢查鎖）
                if (tabAt(tab, i) == f) {
                    // 遍歷鏈表或紅黑樹，更新/添加節點
                    // 如果節點數超過TREEIFY_THRESHOLD (8)，則轉換為紅黑樹
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                }
            }
        }
    }
    return null;
}

這裏有幾個關鍵點：

1. 使用 CAS 操作嘗試插入新節點
2. 只有發生哈希衝突時才使用 synchronized 鎖定桶
3. 當鏈表長度超過閾值時轉換為紅黑樹（TREEIFY_THRESHOLD = 8）

get 操作的關鍵步驟：

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // 計算hash值
    int h = spread(key.hashCode());

    // 如果表不為空且對應位置有節點
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {

        // 檢查首節點
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        // 特殊節點處理...
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;

        // 遍歷鏈表或紅黑樹
        while ((e = e.next) != null) {
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

get 操作的關鍵在於：完全不需要加鎖！這是因為：

1. Node 的 value 和 next 指針使用了 volatile 修飾，保證了內存可見性
2. Node 節點的不可變性設計 - key 和 hash 值在創建後不可修改，value 只能通過原子操作更新
3. JDK 1.8 中 TreeNode（紅黑樹節點）的 left 和 right 引用也使用了 volatile 修飾，保證樹結構變化對所有線程可見

這種設計確保了在不加鎖的情況下，讀操作也能看到最新的值，大大提升了讀性能。

2.3 性能對比實驗

我們來看一個簡單的性能對比實驗，分別測試 HashMap(單線程)、synchronizedMap 和 ConcurrentHashMap 在多線程環境下的性能：

public class MapPerformanceTest {
    private static final int THREAD_COUNT = 100;
    private static final int ITEM_COUNT = 1000;

    public static void main(String[] args) throws Exception {
        // 測試HashMap (僅作為對照，實際多線程中會出錯)
        testMap(new HashMap<>(), "HashMap");

        // 測試synchronizedMap
        testMap(Collections.synchronizedMap(new HashMap<>()), "synchronizedMap");

        // 測試ConcurrentHashMap
        testMap(new ConcurrentHashMap<>(), "ConcurrentHashMap");
    }

    private static void testMap(Map<String, Integer> map, String mapName) throws Exception {
        System.out.println("Test: " + mapName);
        CountDownLatch latch = new CountDownLatch(THREAD_COUNT);
        long startTime = System.nanoTime();

        for (int i = 0; i < THREAD_COUNT; i++) {
            final int threadNum = i;
            new Thread(() -> {
                try {
                    for (int j = 0; j < ITEM_COUNT; j++) {
                        // 執行put操作
                        map.put(threadNum + "-" + j, j);
                    }
                } catch (Exception e) {
                    // HashMap在多線程下可能拋出異常
                    System.err.println(mapName + " 發生異常: " + e.getMessage());
                    // 注意：即使沒有異常，HashMap在多線程下結果也可能不正確！
                } finally {
                    latch.countDown();
                }
            }).start();
        }

        latch.await();
        long endTime = System.nanoTime();
        System.out.println(mapName + " 耗時: " + (endTime - startTime) / 1000000 + "ms");
        System.out.println("Map大小: " + map.size() + " (HashMap結果不可靠，僅作對照)");
        System.out.println();
    }
}

典型的測試結果（僅供參考，具體會因環境而異）：

• HashMap: 約 80ms（但結果不可靠，且可能拋出異常）
• synchronizedMap: 約 450ms
• ConcurrentHashMap: 約 150ms

2.4 實戰應用：高併發緩存實現

讓我們看一個使用 ConcurrentHashMap 實現的簡單高併發緩存，並優化清理過期數據的邏輯：

public class SimpleConcurrentCache<K, V> {
    // 使用ConcurrentHashMap存儲緩存數據
    private final ConcurrentHashMap<K, V> cache = new ConcurrentHashMap<>();

    // 緩存過期時間（毫秒）
    private final long expireTime;

    // 存儲鍵的過期時間
    private final ConcurrentHashMap<K, Long> expireMap = new ConcurrentHashMap<>();

    // 清理線程
    private final ScheduledExecutorService cleanerExecutor;

    public SimpleConcurrentCache(long expireTimeMillis) {
        this.expireTime = expireTimeMillis;
        this.cleanerExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
            Thread t = new Thread(r, "cache-cleaner");
            t.setDaemon(true); // 設置為守護線程
            return t;
        });

        // 定期清理過期數據
        this.cleanerExecutor.scheduleAtFixedRate(
            this::cleanExpiredEntries,
            expireTime / 2,
            expireTime / 2,
            TimeUnit.MILLISECONDS
        );
    }

    public void put(K key, V value) {
        cache.put(key, value);
        expireMap.put(key, System.currentTimeMillis() + expireTime);
    }

    public V get(K key) {
        Long expireTime = expireMap.get(key);
        if (expireTime == null) {
            return null;
        }

        // 檢查是否過期
        if (System.currentTimeMillis() > expireTime) {
            // 原子性移除過期數據
            cache.remove(key);
            expireMap.remove(key);
            return null;
        }

        return cache.get(key);
    }

    // 優化的清理過期條目方法
    private void cleanExpiredEntries() {
        long now = System.currentTimeMillis();

        // 使用迭代器安全地刪除過期條目
        Iterator<Map.Entry<K, Long>> iterator = expireMap.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<K, Long> entry = iterator.next();
            if (now > entry.getValue()) {
                K key = entry.getKey();
                cache.remove(key);
                // 使用Iterator的remove方法安全刪除當前元素
                iterator.remove();
            }
        }
    }

    public void shutdown() {
        cleanerExecutor.shutdown();
    }
}

這個實現的優點：

1. 利用 ConcurrentHashMap 的高併發特性支持多線程訪問
2. 無需加鎖即可進行讀寫操作
3. 使用迭代器安全地刪除過期數據，避免 ConcurrentModificationException
4. 使用守護線程處理清理工作，避免阻止 JVM 退出

3. CopyOnWriteArrayList 深度解析

3.1 寫時複製機制詳解

CopyOnWriteArrayList 是 ArrayList 的線程安全版本，它使用了一種叫"寫時複製"的策略：

graph TD
    A[讀操作] --> B{直接訪問當前數組}
    C[寫操作] --> D[複製整個數組]
    D --> E[在副本上進行修改]
    E --> F[替換引用]

核心原理是：

• 讀操作不需要加鎖，直接訪問內部數組
• 寫操作加鎖，並複製整個數組，在副本上修改完成後替換原數組

需要特別注意的是，CopyOnWriteArrayList 提供的是最終一致性而非強一致性：

• 當線程 A 正在進行寫操作時，線程 B 的讀操作看到的是修改前的數據
• 只有當寫操作完成（新數組引用替換舊數組引用）後，新的讀操作才能看到更新後的數據
• 這意味着讀線程可能會短暫地看到"舊"數據，適用於對實時性要求不高的場景

3.2 源碼分析

讓我們看看 CopyOnWriteArrayList 的關鍵方法實現：

// 添加元素
public boolean add(E e) {
    // 獲取獨佔鎖
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 獲取當前數組
        Object[] elements = getArray();
        int len = elements.length;

        // 創建新數組，長度+1
        Object[] newElements = Arrays.copyOf(elements, len + 1);

        // 在新數組末尾添加元素
        newElements[len] = e;

        // 更新數組引用
        setArray(newElements);
        return true;
    } finally {
        // 釋放鎖
        lock.unlock();
    }
}

// 獲取元素
public E get(int index) {
    // 直接訪問數組，無需加鎖
    return get(getArray(), index);
}

3.3 偽共享問題與性能影響

在 CopyOnWriteArrayList 中，偽共享問題可能嚴重影響性能。偽共享是指多個線程修改位於同一緩存行但不同的變量，導致緩存行失效的問題。

舉個例子，假設兩個線程分別修改數組中相鄰的元素：

// 線程1修改索引0的元素
list.set(0, newValue1);

// 線程2修改索引1的元素
list.set(1, newValue2);

如果這兩個元素恰好位於同一個 CPU 緩存行（通常為 64 字節），那麼當一個線程修改其中一個元素時，會導致整個緩存行失效，另一個線程必須從主內存重新加載數據，即使它實際上修改的是不同的元素。

JDK 9 引入了@Contended註解來解決這個問題，通過填充字節使變量不共享緩存行：

class PaddedElement {
    @Contended
    volatile Object value;
    // ...其他字段
}

3.4 適用場景與性能特性

CopyOnWriteArrayList 適用於以下場景：

• 讀操作遠多於寫操作的場景（讀寫比例通常應該在 10:1 以上）
• 集合規模較小（通常少於 1000 個元素）
• 對實時性要求不高的場景，能容忍短暫的數據不一致

內存開銷考慮：

• 每次寫操作都會創建一個新數組，導致大量臨時對象產生
• 集合元素越多，內存開銷越大，可能增加 GC 壓力
• 寫操作頻繁時會導致性能急劇下降

讓我們通過一個簡單的實驗來比較 CopyOnWriteArrayList 在不同讀寫比例下的性能表現：

public class ListPerformanceTest {
    private static final int THREAD_COUNT = 10;
    private static final int INITIAL_SIZE = 1000;

    // 測試不同讀寫比例
    public static void main(String[] args) throws Exception {
        // 讀多寫少 (10:1)
        testWithRatio(10, 1, "讀多寫少(10:1)");

        // 讀寫均衡 (1:1)
        testWithRatio(1, 1, "讀寫均衡(1:1)");

        // 寫多讀少 (1:10)
        testWithRatio(1, 10, "寫多讀少(1:10)");
    }

    private static void testWithRatio(int readRatio, int writeRatio, String scenario) throws Exception {
        System.out.println("===== 測試場景: " + scenario + " =====");

        List<Integer> cowList = new CopyOnWriteArrayList<>();
        List<Integer> syncList = Collections.synchronizedList(new ArrayList<>());

        // 初始化列表
        for (int i = 0; i < INITIAL_SIZE; i++) {
            cowList.add(i);
            syncList.add(i);
        }

        // 計算讀寫操作次數
        int readCount = 10000 * readRatio / (readRatio + writeRatio);
        int writeCount = 10000 * writeRatio / (readRatio + writeRatio);

        System.out.println("讀操作: " + readCount + ", 寫操作: " + writeCount);

        // 測試CopyOnWriteArrayList
        long cowTime = testList(cowList, "CopyOnWriteArrayList", readCount, writeCount);

        // 測試SynchronizedList
        long syncTime = testList(syncList, "SynchronizedList", readCount, writeCount);

        System.out.println("性能比較: CopyOnWriteArrayList 是 SynchronizedList 的 "
                + String.format("%.2f", (double)syncTime/cowTime) + " 倍");
        System.out.println();
    }

    private static long testList(List<Integer> list, String listName,
                               int readCount, int writeCount) throws Exception {
        CountDownLatch latch = new CountDownLatch(THREAD_COUNT);
        long startTime = System.nanoTime();

        for (int i = 0; i < THREAD_COUNT; i++) {
            final int threadNum = i;
            new Thread(() -> {
                try {
                    // 執行讀操作
                    for (int j = 0; j < readCount; j++) {
                        int index = threadNum % INITIAL_SIZE;
                        list.get(index);
                    }

                    // 執行寫操作
                    for (int j = 0; j < writeCount; j++) {
                        list.add(j);
                    }
                } catch (Exception e) {
                    System.err.println(listName + " 異常: " + e.getMessage());
                } finally {
                    latch.countDown();
                }
            }).start();
        }

        latch.await();
        long endTime = System.nanoTime();
        long duration = (endTime - startTime) / 1000000;
        System.out.println(listName + " 耗時: " + duration + "ms");

        return duration;
    }
}

典型測試結果表明：

• 讀多寫少(10:1)：CopyOnWriteArrayList 比 SynchronizedList 快約 1.5-2 倍
• 讀寫均衡(1:1)：兩者性能相近
• 寫多讀少(1:10)：SynchronizedList 比 CopyOnWriteArrayList 快約 3-5 倍

3.5 實戰應用：併發環境中的事件監聽器管理

一個典型的 CopyOnWriteArrayList 應用場景是事件監聽器管理，使用弱引用防止內存泄漏：

public class WeakEventBus {
    // 使用CopyOnWriteArrayList存儲監聽器的弱引用
    private final CopyOnWriteArrayList<WeakReference<EventListener>> listeners =
        new CopyOnWriteArrayList<>();

    // 用於清理已被GC回收的弱引用
    private final ScheduledExecutorService cleanupExecutor =
        Executors.newSingleThreadScheduledExecutor();

    public WeakEventBus() {
        // 定期清理已被GC回收的弱引用
        cleanupExecutor.scheduleAtFixedRate(
            this::removeStaleListeners,
            10, 10, TimeUnit.SECONDS
        );
    }

    // 註冊監聽器
    public void register(EventListener listener) {
        // 使用弱引用包裝監聽器
        listeners.add(new WeakReference<>(listener));
    }

    // 顯式移除監聽器
    public void unregister(EventListener listener) {
        for (int i = 0; i < listeners.size(); i++) {
            WeakReference<EventListener> ref = listeners.get(i);
            EventListener refListener = ref.get();

            // 如果引用已失效或等於要移除的監聽器
            if (refListener == null || refListener == listener) {
                listeners.remove(i);
                i--; // 調整索引
            }
        }
    }

    // 觸發事件
    public void post(Event event) {
        // 遍歷所有監聽器並通知
        for (WeakReference<EventListener> ref : listeners) {
            EventListener listener = ref.get();
            if (listener != null) {
                try {
                    listener.onEvent(event);
                } catch (Exception e) {
                    // 處理異常
                    e.printStackTrace();
                }
            }
        }
    }

    // 清理已被GC回收的弱引用
    private void removeStaleListeners() {
        listeners.removeIf(ref -> ref.get() == null);
    }

    // 關閉清理執行器
    public void shutdown() {
        cleanupExecutor.shutdown();
    }

    // 事件接口
    public interface EventListener {
        void onEvent(Event event);
    }

    // 事件類
    public static class Event {
        private final String type;
        private final Object data;

        public Event(String type, Object data) {
            this.type = type;
            this.data = data;
        }

        public String getType() {
            return type;
        }

        public Object getData() {
            return data;
        }
    }
}

這個實現的優勢在於：

1. 使用弱引用存儲監聽器，避免因客户端忘記取消註冊導致的內存泄漏
2. 事件觸發時不會阻塞註冊/註銷操作
3. 定期清理無效引用，避免列表無限增長
4. 對於需要顯式註銷的監聽器，提供 unregister 方法

4. 併發隊列體系與實戰應用

4.1 併發隊列分類與實現

Java 併發隊列可以分為兩大類：

還有一個重要的有序 Map 容器：

graph LR
    A[併發有序Map] --> B[ConcurrentSkipListMap]
    A --> C[ConcurrentSkipListSet]

主要特點對比：

1. 阻塞隊列：當隊列滿/空時，入隊/出隊操作會阻塞等待
2. 非阻塞隊列：入隊/出隊操作不會阻塞，而是立即返回成功或失敗
3. 跳錶容器：基於跳錶實現的有序併發容器，提供 log(n)級別的查找性能

4.2 ConcurrentLinkedQueue 的無鎖實現與 ABA 問題

ConcurrentLinkedQueue 是一個基於鏈表的無界隊列，它使用 CAS 操作實現無鎖併發：

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final Node<E> newNode = new Node<E>(e);

    for (Node<E> t = tail, p = t;;) {
        Node<E> q = p.next;
        if (q == null) {
            // p是最後一個節點
            if (p.casNext(null, newNode)) {
                // 成功插入，嘗試更新tail
                if (p != t)
                    casTail(t, newNode);
                return true;
            }
        }
        else if (p == q)
            // 幫助初始化或處理已刪除節點
            p = (t != (t = tail)) ? t : head;
        else
            // 檢查下一個節點
            p = (p != t && t != (t = tail)) ? t : q;
    }
}

這段代碼的精髓在於通過 CAS 操作實現了無鎖的併發添加，大大提高了性能。

ABA 問題及其解決方案

CAS 操作可能面臨 ABA 問題（值被修改後又改回原值）。在 ConcurrentLinkedQueue 中，由於操作的是對象引用，而每個 Node 是唯一的新對象，所以不會出現引用級別的 ABA 問題。

對於可能出現 ABA 問題的場景，JUC 提供了AtomicStampedReference解決方案：

// 創建帶版本號的原子引用
AtomicStampedReference<Integer> atomicRef =
    new AtomicStampedReference<>(100, 0); // 初始值100，版本號0

// 獲取當前值和版本號
int[] stampHolder = new int[1];
Integer value = atomicRef.get(stampHolder);
int initialStamp = stampHolder[0];

// CAS操作時同時檢查值和版本號
boolean success = atomicRef.compareAndSet(
    value, newValue, initialStamp, initialStamp + 1);

AtomicStampedReference通過引入版本號，可以檢測值是否被中間修改過，即使最終值相同也能察覺到變化。

4.3 各種 BlockingQueue 對比與選擇

不同 BlockingQueue 實現有不同的特點和適用場景：

1. ArrayBlockingQueue：基於數組的有界隊列

   • 適用：有界緩衝區，生產速度與消費速度相近
   • 特點：構造時必須指定容量，不會自動擴容

2. LinkedBlockingQueue：基於鏈表的可選有界隊列

   • 適用：吞吐量要求高，生產和消費速度差異大
   • 特點：不指定容量時默認為 Integer.MAX_VALUE(21 億)，需要注意 OOM 風險

3. PriorityBlockingQueue：支持優先級的無界隊列

   • 適用：任務優先級處理場景
   • 特點：元素必須實現 Comparable 接口或提供 Comparator

4. DelayQueue：延遲獲取元素的無界隊列

   • 適用：定時任務調度
   • 特點：元素必須實現 Delayed 接口，到期時間到達才能取出

5. SynchronousQueue：沒有緩衝的阻塞隊列

   • 適用：直接交付場景，生產者必須等待消費者取走元素
   • 特點：沒有存儲空間，put 必須等待 take

6. LinkedTransferQueue：融合 SynchronousQueue 和 LinkedBlockingQueue 的特性

   • 適用：既需要隊列存儲又需要直接交付的場景
   • 特點：支持 tryTransfer 操作，可選擇性地等待消費者接收

7. ConcurrentSkipListMap：基於跳錶的併發有序 Map

   • 適用：需要按鍵排序且併發訪問的場景
   • 特點：提供了與 TreeMap 類似的操作，但是線程安全的

4.4 實戰應用：高效生產者-消費者系統

下面是一個使用 BlockingQueue 實現的生產者-消費者模式示例：

public class TaskProcessor {
    // 使用有界阻塞隊列存儲任務
    private final BlockingQueue<Task> taskQueue;
    // 消費者線程池
    private final ExecutorService consumers;
    // 運行標誌
    private volatile boolean running = true;

    public TaskProcessor(int queueSize, int consumerCount) {
        // 顯式指定隊列大小，避免無界隊列可能的OOM風險
        this.taskQueue = new LinkedBlockingQueue<>(queueSize);
        this.consumers = Executors.newFixedThreadPool(consumerCount, r -> {
            Thread t = new Thread(r, "task-consumer");
            t.setUncaughtExceptionHandler((thread, ex) -> {
                System.err.println("消費者線程異常: " + ex.getMessage());
            });
            return t;
        });

        // 啓動消費者線程
        for (int i = 0; i < consumerCount; i++) {
            consumers.submit(this::consumeTask);
        }
    }

    // 提交任務（生產者）
    public boolean submitTask(Task task) {
        if (!running) {
            return false;
        }

        try {
            // 嘗試放入隊列，最多等待100ms
            return taskQueue.offer(task, 100, TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            return false;
        }
    }

    // 消費者線程邏輯
    private void consumeTask() {
        while (running) {
            try {
                // 從隊列取任務，最多等待1s
                Task task = taskQueue.poll(1, TimeUnit.SECONDS);
                if (task != null) {
                    try {
                        // 處理任務
                        task.process();
                    } catch (Exception e) {
                        // 處理任務異常
                        System.err.println("Task process error: " + e.getMessage());
                    }
                }
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                break;
            }
        }
    }

    // 返回當前隊列大小（用於監控）
    public int getQueueSize() {
        return taskQueue.size();
    }

    // 關閉處理器
    public void shutdown() {
        running = false;
        consumers.shutdown();
        try {
            if (!consumers.awaitTermination(5, TimeUnit.SECONDS)) {
                consumers.shutdownNow();
            }
        } catch (InterruptedException e) {
            consumers.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }

    // 任務接口
    public interface Task {
        void process();
    }
}

這個實現的優點：

1. 利用 BlockingQueue 自動處理生產者和消費者的速度差異
2. 提供超時機制避免生產者無限阻塞
3. 優雅地處理關閉邏輯，確保資源釋放
4. 顯式指定隊列大小，避免潛在的內存溢出問題

5. 併發容器選擇指南與性能優化

5.1 如何正確選擇併發容器？

5.2 分佈式環境下的併發容器選擇

在分佈式系統中，JUC 併發容器僅能保證單 JVM 內的線程安全，對於跨 JVM 的併發訪問，需要結合分佈式組件：

對於需要跨 JVM 共享的數據，可以考慮：

1. Redis：適用於高性能分佈式緩存和數據結構
2. Hazelcast/Ignite：分佈式內存數據網格
3. ZooKeeper：分佈式鎖和協調服務
4. Kafka：分佈式消息隊列替代 BlockingQueue

5.3 常見問題與解決方案

1. 內存佔用問題

   • CopyOnWriteArrayList 每次寫操作都會複製數組，容易導致 GC 壓力增大
   • 解決方案：限制集合大小，合併修改操作，必要時考慮其他集合實現

2. ABA 問題

   • CAS 操作可能遇到 ABA 問題（值被修改後又改回原值）
   • 解決方案：使用 AtomicStampedReference 或 AtomicMarkableReference 添加版本號或標記位

3. 死鎖問題

   • 即使使用併發容器，不當的使用模式仍可能導致死鎖
   • 解決方案：避免嵌套鎖、使用帶超時的操作、遵循固定的鎖定順序

4. 偽共享問題

   • 多核 CPU 緩存行共享導致的性能下降（一個線程修改變量導致其他 CPU 緩存失效）
   • 解決方案：使用@Contended 註解、填充變量避免共享緩存行

5.4 性能優化技巧

1. 容器預設容量

   // 預設合理容量避免擴容
   Map<String, String> map = new ConcurrentHashMap<>(1024);

2. 批量操作優化

   // 使用ConcurrentHashMap的原子複合操作
   map.compute(key, (k, v) -> v == null ? 1 : v + 1);

3. 選擇合適的負載因子

   // 調整負載因子平衡空間與時間
   new ConcurrentHashMap<>(initialCapacity, loadFactor);

4. 避免不必要的同步

   // 優先使用putIfAbsent而非containsKey+put
   map.putIfAbsent(key, value);

5. 隊列容量設置

   // 設置合理的隊列容量，避免生產者等待或消費者空轉
   new ArrayBlockingQueue<>(Runtime.getRuntime().availableProcessors() * 2);

6. 總結

Java 併發容器是多線程編程的重要工具，正確理解它們的實現原理和適用場景，能夠幫助我們寫出更高效、更可靠的併發程序。在實際使用中，需要根據具體的業務場景和性能需求，選擇合適的併發容器實現。

感謝您耐心閲讀到這裏！如果覺得本文對您有幫助，歡迎點贊 👍、收藏 ⭐、分享給需要的朋友，您的支持是我持續輸出技術乾貨的最大動力！

如果想獲取更多 Java 技術深度解析，歡迎點擊頭像關注我，後續會每日更新高質量技術文章，陪您一起進階成長～