Java 多線程系列（六）：阻塞隊列簡化生產者-消費者模式

Java 多線程系列（六）：阻塞隊列簡化生產者-消費者模式

本文通過經典的「生產者-消費者」模型，對比傳統 wait/notify 手動同步與 BlockingQueue 內置同步的實現差異，展示如何利用 JDK 並發包簡化多線程協作代碼。

傳統實現：手動鎖 + Condition

使用 ReentrantLock 配合 Condition 實現緩衝區同步：

public class ConsumerProducer {
    private static Buffer buffer = new Buffer();

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.execute(new ProducerTask());
        executor.execute(new ConsumerTask());
        executor.shutdown();
    }

    static class ProducerTask implements Runnable {
        @Override
        public void run() {
            try {
                int i = 1;
                while (true) {
                    System.out.println("Producer writes " + i);
                    buffer.write(i++);
                    Thread.sleep((int)(Math.random() * 10000));
                }
            } catch (InterruptedException e) { e.printStackTrace(); }
        }
    }

    static class ConsumerTask implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    System.out.println("\t\t\tConsumer reads " + buffer.read());
                    Thread.sleep((int)(Math.random() * 10000));
                }
            } catch (InterruptedException e) { e.printStackTrace(); }
        }
    }

    static class Buffer {
        private static final int CAPACITY = 1;
        private final LinkedList<Integer> queue = new LinkedList<>();
        private final Lock lock = new ReentrantLock();
        private final Condition notEmpty = lock.newCondition();
        private final Condition notFull = lock.newCondition();

        public void write(int value) {
            lock.lock();
            try {
                while (queue.size() == CAPACITY) {
                    System.out.println("Wait for notFull condition");
                    notFull.await();
                }
                queue.offer(value);
                notEmpty.signal();
            } catch (Exception e) { e.printStackTrace(); }
            finally { lock.unlock(); }
        }

        public int read() {
            int value = 0;
            lock.lock();
            try {
                while (queue.isEmpty()) {
                    System.out.println("\t\t\tWait for notEmpty condition");
                    notEmpty.await();
                }
                value = queue.remove();
                notFull.signal();
            } catch (Exception e) { e.printStackTrace(); }
            finally { lock.unlock(); }
            return value;
        }
    }
}

進階實現：ArrayBlockingQueue 內置同步

ArrayBlockingQueue 是有界阻塞隊列，內部已實現線程安全的 put/take，無需手動編寫同步邏輯：

public class ConsumerProducerUsingBlockingQueue {
    // 容量為 2 的有界隊列
    private static ArrayBlockingQueue<Integer> buffer = new ArrayBlockingQueue<>(2);

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        executor.execute(new ProducerTask());
        executor.execute(new ConsumerTask());
        executor.shutdown();
    }

    static class ProducerTask implements Runnable {
        @Override
        public void run() {
            try {
                int i = 1;
                while (true) {
                    System.out.println("Producer writes " + i);
                    buffer.put(i++); // 隊列滿時自動阻塞
                    Thread.sleep((int)(Math.random() * 10000));
                }
            } catch (InterruptedException e) { e.printStackTrace(); }
        }
    }

    static class ConsumerTask implements Runnable {
        @Override
        public void run() {
            try {
                while (true) {
                    System.out.println("\t\t\tConsumer reads " + buffer.take()); // 隊列空時自動阻塞
                    Thread.sleep((int)(Math.random() * 10000));
                }
            } catch (InterruptedException e) { e.printStackTrace(); }
        }
    }
}

三大核心阻塞隊列對比

核心方法速查

最佳實踐總結

1. 優先使用阻塞隊列：減少手動同步代碼，降低死鎖、活鎖風險。
2. 合理設置容量：有界隊列防止 OOM，容量過小導致生產者頻繁阻塞，需壓測調優。
3. 選擇合適類型：高併發用 LinkedBlockingQueue，需優先級用 PriorityBlockingQueue，固定緩衝用 ArrayBlockingQueue。
4. 避免在鎖中調用阻塞方法：隊列操作已包含同步，外層無需再加鎖。