Java 並發包源碼深度解析：atomic、lock、concurrent 三大核心模塊全景剖析

Java 並發包源碼深度解析

隨著多核 CPU 普及，高效利用並行計算能力成為大規模應用的關鍵。Java java.util.concurrent (JUC) 包提供了工業級的並發工具集，本文將從源碼層面深度剖析 atomic (原子類)、locks (鎖機制)、concurrent (並發集合/線程池) 三大核心模塊，揭示其底層實現原理與設計智慧。

JUC 包整體架構

• atomic：基於 CAS 的無鎖原子操作類 (AtomicInteger、AtomicReference 等)
• locks：顯式鎖框架 (ReentrantLock、ReadWriteLock、Condition 等)
• concurrent：高性能並發集合 (ConcurrentHashMap、BlockingQueue)、線程池 (ThreadPoolExecutor)、同步工具 (CountDownLatch、CyclicBarrier、Semaphore)

atomic 包：CAS 無鎖並發的基石

核心設計思想

atomic 包提供 12 個原子類，涵蓋四類更新方式：基本類型 (AtomicInteger/Long/Boolean)、數組 (AtomicIntegerArray/LongArray/ReferenceArray)、引用 (AtomicReference/StampedReference/MarkableReference)、字段更新器 (AtomicIntegerFieldUpdater 等)。其核心均基於 CAS (Compare-And-Swap) 指令實現樂觀鎖語義：

• 三個操作數：內存值 V、預期值 A、新值 B
• 原子性判斷：若 V == A，則 V = B 並返回 true；否則不修改返回 false
• 失敗重試：自旋循環直到成功，避免線程阻塞上下文切換開銷

Unsafe 與 JVM 級支持

所有原子類底層委派 sun.misc.Unsafe 的原生方法：

// AtomicInteger.incrementAndGet() 源碼鏈路
public final int incrementAndGet() {
    return unsafe.getAndAddInt(this, valueOffset, 1) + 1;
}

// Unsafe.getAndAddInt 實現 (JVM 內部)
public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    do {
        v = getIntVolatile(o, offset); // volatile 讀取保證可見性
    } while (!compareAndSwapInt(o, offset, v, v + delta)); // CAS 自旋
    return v;
}

JDK 5+ 將 CAS 映射為 CPU 級指令 (x86: LOCK CMPXCHG，ARM: LDREX/STREX)，由硬體保證原子性。JVM 會根據平台選擇最高效實現，甚至在無 CAS 支持的架構上退化為內部鎖。

實戰演示：多線程計數器

public class AtomicDemo {
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newCachedThreadPool();
        AtomicInteger counter = new AtomicInteger(0);
        AtomicIntegerArray arr = new AtomicIntegerArray(10);
        AtomicReference ref = new AtomicReference<>(new User("init", 0));

        // 並發自增 10 萬次
        for (int i = 0; i < 10; i++) {
            executor.execute(() -> {
                for (int j = 0; j < 10000; j++) counter.incrementAndGet();
            });
            executor.execute(() -> {
                for (int j = 0; j < 10; j++) 
                    for (int k = 0; k < 10; k++) arr.getAndAdd(k, 1);
            });
            executor.execute(() -> ref.set(new User("luoxn28", 23)));
        }
        executor.shutdown();
        Thread.sleep(100);
        System.out.println("Counter: " + counter); // 精確 100000
        System.out.println("Array[0]: " + arr.get(0)); // 精確 100
        System.out.println("Ref: " + ref.get());
    }
}

locks 包：顯式鎖與 Condition 機制

Lock 接口核心契約

public interface Lock {
    void lock();                    // 阻塞獲取鎖
    void lockInterruptibly() throws InterruptedException; // 可中斷獲取
    boolean tryLock();              // 非阻塞嘗試
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException; // 超時獲取
    void unlock();                  // 釋放鎖
    Condition newCondition();       // 關聯條件變量
}

ReentrantLock 實現精髓

• 基於 AQS (AbstractQueuedSynchronizer) 狀態機 + CLH 隊列實現
• 支援公平/非公平模式 (構造參數 fair)，非公平預設吞吐更高
• 可重入：同一線程多次 lock() 需同次數 unlock()
• Condition 替代 Object.wait/notify，支援多條件變量、可中斷等待、超時等待

使用範例：線程安全計數器

public class LockDemo {
    public static void main(String[] args) throws InterruptedException {
        Lock lock = new ReentrantLock();
        AtomicInteger data = new AtomicInteger(0);
        ExecutorService executor = Executors.newCachedThreadPool();
        
        for (int i = 0; i < 2; i++) {
            executor.execute(() -> {
                for (int j = 0; j < 10; j++) {
                    lock.lockInterruptibly();
                    try { data.getAndIncrement(); }
                    catch (InterruptedException e) { e.printStackTrace(); }
                    finally { lock.unlock(); }
                }
            });
        }
        executor.shutdown();
        Thread.sleep(10);
        System.out.println(data.get()); // 20
    }
}

concurrent 包：高性能並發集合與線程池

BlockingQueue：生產者-消費者核心協作組件

核心方法語義：

• put(e)：隊列滿阻塞等待
• take()：隊列空阻塞等待
• offer(e, timeout)/poll(timeout)：超時控制

ArrayBlockingQueue：有界環形數組隊列

• 單 ReentrantLock + 兩個 Condition (notEmpty/notFull)
• putIndex/takeIndex 環形游標，count 計數器
• 所有操作串行化，公平鎖可保證 FIFO 獲取順序

LinkedBlockingQueue：無界/有界鏈表隊列 (吞吐更高)

• 雙鎖設計：putLock/takeLock 分離，生產消費並行
• 各自關聯 notFull/notEmpty Condition
• count 用 AtomicInteger 避免雙鎖競爭
• 入隊 enqueue 尾插，出隊 dequeue 頭取並斷鏈助 GC

ConcurrentHashMap (JDK 8+)：分段鎖 → CAS + synchronized 進化

核心數據結構

• Node<K,V>：基本鏈表節點，val/next 為 volatile，禁止 setValue 直接修改
• TreeNode<K,V>：紅黑樹節點，繼承 Node 保留 next 指針
• TreeBin<K,V>：樹根包裝器，持有讀寫鎖，實際存放在 table 數組中
• table：Node<K,V>[] 數組，長度為 2 的冪次

put 操作核心流程 (putVal)

final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)
            tab = initTable(); // 懶初始化
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 空槽位：CAS 無鎖插入
            if (casTabAt(tab, i, null, new Node<>(hash, key, value, null)))
                break;
        }
        else if ((fh = f.hash) == MOVED)
            tab = helpTransfer(tab, f); // 協助擴容
        else {
            V oldVal = null;
            synchronized (f) { // 細粒度鎖：鎖住鏈表/樹頭節點
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) { // 鏈表模式
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            if (e.hash == hash && keyEquals(e.key, key)) {
                                oldVal = e.val; if (!onlyIfAbsent) e.val = value; break;
                            }
                            if (e.next == null) { e.next = new Node<>(hash, key, value, null); break; }
                            e = e.next;
                        }
                    }
                    else if (f instanceof TreeBin) { // 紅黑樹模式
                        binCount = 2;
                        TreeNode<K,V> p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value);
                        if (p != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; }
                    }
                }
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); // 鏈表轉樹
                if (oldVal != null) return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount); // 更新 sizeCtl
    return null;
}

關鍵設計亮點

• 無鎖擴容：MOVED 節點標記 + 多線程協助擴容 (helpTransfer)
• 細粒度同步：僅鎖定發生哈希碰撞的桶頭節點，非碰撞槽位 CAS 直插
• 樹化閾值 8：鏈表長度 ≥ 8 且數組長度 ≥ 64 才轉紅黑樹，防止退化攻擊
• get 完全無鎖：volatile 讀保證可見性，鏈表/樹遍歷無需同步

使用示例

ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
ExecutorService es = Executors.newCachedThreadPool();
for (int i = 0; i < 3; i++) es.execute(() -> {
    for (int j = 0; j < 10; j++) map.put(UUID.randomUUID().toString(), UUID.randomUUID().toString());
});
es.shutdown(); Thread.sleep(2000);
map.forEach((k, v) -> System.out.println(k + " -> " + v));

總結與最佳實踐

深入理解 JUC 源碼不僅能指導選型調優，更能培養無鎖編程、隊列同步、狀態機設計等核心並發思維。建議結合 AQS、ForkJoinPool、CompletableFuture 等進階組件持續深入探索。

參考資料：
1. ConcurrentHashMap 源碼分析 (JDK 8 版本)
2. Java 並發基礎總結