您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Java 線程通信機制詳解：傳統方式、Condition 與 BlockingQueue 實戰

編輯：Java 綜合教學

Java 線程通信核心概念

線程通信是確保多線程協調運行的關鍵機制，通常在實現線程同步時才需要引入。Java 提供了三種主流的線程通信方案，分別是基於 Object 的傳統方法、基於 Lock 的 Condition 接口，以及基於阻塞隊列 BlockingQueue 的生產者-消費者模型。

一、傳統線程通信：wait/notify/notifyAll

Object 類提供了三個核心方法：

wait()：導致當前線程等待，並釋放同步監視器的鎖，直到被 notify() 或 notifyAll() 喚醒。
notify()：喚醒在此同步監視器上等待的單個線程（若有多個則任意選擇）。
notifyAll()：喚醒所有等待線程，這些線程將競爭鎖以繼續執行。

這三個方法必須由同步監視器對象調用，分為兩種場景：

1. 同步方法場景

同步監視器為 this，可直接調用三個方法。下例演示兩個線程輪流對共享數據遞增：

public class SyncMethodThreadCommunication {
    static class DataWrap {
        int data = 0;
        boolean flag = false;
        public synchronized void addThreadA() {
            if (flag) { try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } }
            data++;
            System.out.println(Thread.currentThread().getName() + " " + data);
            flag = true;
            notify();
        }
        public synchronized void addThreadB() {
            if (!flag) { try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } }
            data++;
            System.out.println(Thread.currentThread().getName() + " " + data);
            flag = false;
            notify();
        }
    }
    // 線程類省略...
    public static void main(String[] args) {
        DataWrap dataWrap = new DataWrap();
        new ThreadA(dataWrap).start();
        new ThreadB(dataWrap).start();
    }
}

2. 同步代碼塊場景

需顯式使用監視器對象（如 dataWrap）調用三個方法：

public class SyncBlockThreadComminication {
    static class DataWrap { boolean flag; int data; }
    static class ThreadA extends Thread {
        DataWrap dataWrap;
        @Override public void run() {
            for (int i = 0; i < 10; i++) {
                synchronized (dataWrap) {
                    if (dataWrap.flag) { try { dataWrap.wait(); } catch (InterruptedException e) { e.printStackTrace(); } }
                    dataWrap.data++;
                    System.out.println(getName() + " " + dataWrap.data);
                    dataWrap.flag = true;
                    dataWrap.notify();
                }
            }
        }
    }
    // ThreadB 與 main 方法類似...
}

二、使用 Condition 控制線程通信

當使用 Lock（如 ReentrantLock）實現同步時，配套使用 Condition 協調線程。Condition 的 await()、signal()、signalAll() 分別對應傳統的 wait()、notify()、notifyAll()。

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class SyncLockThreadCommunication {
    static class DataWrap {
        int data; boolean flag;
        private final Lock lock = new ReentrantLock();
        private final Condition condition = lock.newCondition();
        public void addThreadA() {
            lock.lock();
            try {
                if (flag) { try { condition.await(); } catch (InterruptedException e) { e.printStackTrace(); } }
                data++;
                System.out.println(Thread.currentThread().getName() + " " + data);
                flag = true;
                condition.signal();
            } finally { lock.unlock(); }
        }
        // addThreadB 類似...
    }
    // 線程類與 main 方法省略...
}

三、使用 BlockingQueue 實現生產者-消費者模式

BlockingQueue 是為線程通信設計的隊列接口，核心特性：

生產者調用 put(E e) 時，隊列滿則阻塞；
消費者調用 take() 時，隊列空則阻塞。

常見實現類：

實現類	特點
ArrayBlockingQueue	基於數組，有界隊列
LinkedBlockingQueue	基於鏈表，可選有界/無界
PriorityBlockingQueue	優先級隊列，元素需實現 Comparable
SynchronousQueue	同步隊列，存取必須交替進行
DelayQueue	延遲隊列，元素需實現 Delayed 接口

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueThreadComminication {
    static class DataWrap { int data; }
    static class ThreadA extends Thread { // 消費者
        private BlockingQueue blockingQueue;
        @Override public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    DataWrap dw = blockingQueue.take();
                    dw.data++;
                    System.out.println(getName() + " " + dw.data);
                    Thread.sleep(1000);
                } catch (InterruptedException e) { e.printStackTrace(); }
            }
        }
    }
    static class ThreadB extends Thread { // 生產者
        private BlockingQueue blockingQueue;
        private DataWrap dataWrap;
        @Override public void run() {
            for (int i = 0; i < 100; i++) {
                try {
                    dataWrap.data++;
                    System.out.println(getName() + " " + dataWrap.data);
                    blockingQueue.put(dataWrap);
                    Thread.sleep(1000);
                } catch (InterruptedException e) { e.printStackTrace(); }
            }
        }
    }
    public static void main(String[] args) {
        DataWrap dataWrap = new DataWrap();
        BlockingQueue queue = new ArrayBlockingQueue<>(1);
        new ThreadA(queue, "Consumer").start();
        new ThreadB(queue, dataWrap, "Producer").start();
    }
}