您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 程式設計入門

深入解析 Java BlockingQueue 核心原理與實戰效能測試

編輯：Java 程式設計入門

前言：為何選擇 BlockingQueue？

在維護一個 Java 專案時，團隊內部常討論 Java 與 C++ 的效能優劣。有資深開發者認為 Java 標準庫由大師級人物編寫，穩定性極高，且在線程間傳遞訊息時，直接使用 BlockingQueue 即足夠應對大多數場景。為了驗證這一論點，我決定撰寫一個基準測試程式，實測其吞吐量表現，並深入剖析其底層原始碼實現機制。

實戰效能測試：十萬條資料傳遞耗時

測試環境：單生產者、單消費者模式，使用 LinkedBlockingQueue 作為緩衝區。生產者以批次 10,000 條的速度持續投放資料，消費者輪詢取出並統計耗時。

核心測試代碼

// 主測試類
import java.util.concurrent.*;

public class BlockingQueueBenchmark {
    public static void main(String[] args) {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>();
        Thread consumer = new Thread(new DataConsumer(queue));
        consumer.start();

        // 生產者主迴路
        while (true) {
            try {
                for (int i = 0; i < 10000; i++) {
                    queue.offer(i); // 非阻塞放入
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

// 消費者實現
class DataConsumer implements Runnable {
    private final BlockingQueue<Integer> queue;

    DataConsumer(BlockingQueue<Integer> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        long start = System.currentTimeMillis();
        int count = 0;
        while (true) {
            Integer val = queue.poll(); // 非阻塞取出
            if (val != null) count++;
            if (count == 100_000) {
                long end = System.currentTimeMillis();
                System.err.println("總耗時: " + (end - start) + " ms");
                break;
            }
        }
    }
}

測試結果分析

在測試機器上，傳遞 10 萬條整型資料僅耗時 約 50 毫秒，展現了極高的吞吐能力。這得益於 LinkedBlockingQueue 採用的**雙鎖機制**（Put Lock / Take Lock）與**條件變量**，實現了生產與消費操作的並行化。

核心原始碼深度解析

以 JDK 8 源碼為例，重點剖析 offer() 與 poll() 兩個非阻塞核心方法。

1. offer(E e) —— 非阻塞入隊

public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final AtomicInteger count = this.count;
    // 快速失敗：隊列已滿
    if (count.get() == capacity) return false;

    int c = -1;
    final ReentrantLock putLock = this.putLock;
    putLock.lock();
    try {
        // 雙重檢查
        if (count.get() < capacity) {
            insert(e); // 入隊操作
            c = count.getAndIncrement();
            // 若隊列未滿，喚醒其他生產者
            if (c + 1 < capacity) notFull.signal();
        }
    } finally {
        putLock.unlock();
    }
    // 若入隊前隊列為空，喚醒消費者
    if (c == 0) signalNotEmpty();
    return c >= 0;
}

2. poll() —— 非阻塞出隊

public E poll() {
    final AtomicInteger count = this.count;
    // 快速失敗：隊列為空
    if (count.get() == 0) return null;

    E x = null;
    int c = -1;
    final ReentrantLock takeLock = this.takeLock;
    takeLock.lock();
    try {
        if (count.get() > 0) {
            x = extract(); // 出隊操作
            c = count.getAndDecrement();
            // 若出隊後仍有元素，喚醒其他消費者
            if (c > 1) notEmpty.signal();
        }
    } finally {
        takeLock.unlock();
    }
    // 若出隊前隊列已滿，喚醒生產者
    if (c == capacity) signalNotFull();
    return x;
}