使用 BlockingQueue 輕鬆解決生產者-消費者問題

前言

在多執行緒程式設計中，生產者-消費者模式是經典的並發設計模式。傳統上我們會使用 wait() 和 notify() 來協調執行緒，但這容易出錯且程式碼複雜。Java 5 引入的 java.util.concurrent.BlockingQueue 介面提供了開箱即用的阻塞隊列實現，讓我們能以更簡潔、安全的方式解決這個問題。

為什麼選擇 BlockingQueue？

生產者-消費者問題的核心在於控制「生產」與「消費」的節奏：當隊列滿時生產者需等待，隊列空時消費者需等待。BlockingQueue 內建了 put()（阻塞放入）和 take()（阻塞取出）方法，自動處理執行緒的掛起與喚醒，開發者無需手動管理鎖與條件變數，極大降低了出錯風險。

實作範例

生產者

package cn.edu.hdu.chenpi.cpdemo.howtodoinjava;

import java.util.concurrent.BlockingQueue;

class Producer implements Runnable {
    private final BlockingQueue<Object> queue;

    Producer(BlockingQueue<Object> theQueue) {
        this.queue = theQueue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                Object resource = produceResource();
                queue.put(resource); // 隊列滿時自動阻塞
                System.out.println("生產資源 - 隊列大小: " + queue.size());
            }
        } catch (InterruptedException ex) {
            System.out.println("生產者中斷");
            Thread.currentThread().interrupt();
        }
    }

    private Object produceResource() throws InterruptedException {
        Thread.sleep(100); // 模擬生產耗時
        return new Object();
    }
}

消費者

package cn.edu.hdu.chenpi.cpdemo.howtodoinjava;

import java.util.concurrent.BlockingQueue;

class Consumer implements Runnable {
    private final BlockingQueue<Object> queue;

    Consumer(BlockingQueue<Object> theQueue) {
        this.queue = theQueue;
    }

    @Override
    public void run() {
        try {
            while (true) {
                Object resource = queue.take(); // 隊列空時自動阻塞
                System.out.println("消費資源 - 隊列大小: " + queue.size());
                consumeResource(resource);
            }
        } catch (InterruptedException ex) {
            System.out.println("消費者中斷");
            Thread.currentThread().interrupt();
        }
    }

    private void consumeResource(Object obj) throws InterruptedException {
        Thread.sleep(100); // 模擬消費耗時
        System.out.println("正在消費: " + obj);
    }
}

測試執行

package cn.edu.hdu.chenpi.cpdemo.howtodoinjava;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class ProducerConsumerExample {
    public static void main(String[] args) throws InterruptedException {
        int producerCount = 4;
        int consumerCount = 3;
        // 有界隊列，容量 20
        BlockingQueue<Object> queue = new LinkedBlockingQueue<>(20);

        for (int i = 0; i < producerCount; i++) {
            new Thread(new Producer(queue)).start();
        }
        for (int i = 0; i < consumerCount; i++) {
            new Thread(new Consumer(queue)).start();
        }

        // 運行 10 秒後優雅關閉
        Thread.sleep(10_000);
        System.exit(0);
    }
}

執行結果分析

程式輸出顯示隊列大小始終不超過 20，生產者與消費者交替運作，完美展示了阻塞隊列的流控效果：

生產資源 - 隊列大小: 1
消費資源 - 隊列大小: 0
生產資源 - 隊列大小: 1
...
生產資源 - 隊列大小: 20
消費資源 - 隊列大小: 19
...

總結

使用 BlockingQueue 解決生產者-消費者問題，程式碼清晰、線程安全且效能優異。實務上可根據需求選擇 ArrayBlockingQueue（固定大小陣列）、LinkedBlockingQueue（鏈表）或 PriorityBlockingQueue（優先級）等實現類。

原文連結：howtodoinjava.com