生產者消費者模式通過協調生產者和消費者對共享緩沖區的訪問，實現多線程協作。1. 使用wait()/notifyall()機制：當緩沖區滿時生產者等待，空時消費者等待，通過notifyall()喚醒線程避免死鎖；2. 選擇合適的阻塞隊列：如arrayblockingqueue（有界隊列適合穩定場景）、linkedblockingqueue（適合速度差異大場景）、priorityblockingqueue（優先級處理）、delayqueue（延遲任務）和synchronousqueue（傳遞性場景）；3. 其他實現方式：包括使用blockingqueue簡化代碼、reentrantlock與condition提供更靈活控制。不同方法適用于不同需求，blockingqueue適合簡單實現，reentrantlock適合復雜控制，而wait/notify是基礎理解方式。
生產者消費者模式是一種經典的多線程協作模式，它巧妙地平衡了生產速度和消費速度，避免了資源浪費和數據丟失。在Java中，我們可以利用wait()和notify()/notifyAll()方法來實現這一模式。簡單來說，生產者負責生產數據并放入共享緩沖區，消費者負責從緩沖區取出數據進行消費。
解決方案
實現生產者消費者模式的核心在于協調生產者和消費者對共享緩沖區的訪問。以下是一個簡單的Java實現：
立即學習“Java免費學習筆記（深入）”；
import java.util.LinkedList; import java.util.Queue; public class ProducerConsumer { private static final int CAPACITY = 5; private final Queue<Integer> buffer = new LinkedList<>(); private final Object lock = new Object(); class Producer implements Runnable { @Override public void run() { int i = 0; while (true) { synchronized (lock) { try { while (buffer.size() == CAPACITY) { System.out.println("Buffer is full, Producer is waiting"); lock.wait(); // 緩沖區滿，生產者等待 } buffer.offer(i); System.out.println("Produced: " + i); i++; lock.notifyAll(); // 通知消費者 Thread.sleep((long) (Math.random() * 100)); // 模擬生產時間 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } } } class Consumer implements Runnable { @Override public void run() { while (true) { synchronized (lock) { try { while (buffer.isEmpty()) { System.out.println("Buffer is empty, Consumer is waiting"); lock.wait(); // 緩沖區空，消費者等待 } int value = buffer.poll(); System.out.println("Consumed: " + value); lock.notifyAll(); // 通知生產者 Thread.sleep((long) (Math.random() * 200)); // 模擬消費時間 } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } } } public static void main(String[] args) { ProducerConsumer pc = new ProducerConsumer(); new Thread(pc.new Producer()).start(); new Thread(pc.new Consumer()).start(); new Thread(pc.new Consumer()).start(); // 多個消費者 } }
這段代碼中，buffer 是共享緩沖區，lock 是用于同步的鎖對象。Producer 和 Consumer 分別是生產者和消費者線程。關鍵在于 wait() 和 notifyAll() 的使用：
- wait()：當緩沖區滿（生產者）或空（消費者）時，線程調用 wait() 方法釋放鎖并進入等待狀態。
- notifyAll()：當生產者生產了新的數據或消費者消費了數據后，調用 notifyAll() 方法喚醒所有等待的線程，讓它們重新競爭鎖。
為什么使用 notifyAll() 而不是 notify()？
雖然 notify() 也能喚醒一個等待的線程，但在生產者消費者模式中，使用 notifyAll() 更安全。 考慮以下情況：如果只有一個消費者線程在等待，notify() 可以喚醒它。但是，如果有多個消費者線程都在等待，而 notify() 喚醒的恰好也是一個消費者線程，那么這個線程可能會發現緩沖區仍然是空的，然后再次進入等待狀態。這樣就可能導致死鎖。notifyAll() 確保所有等待的線程都有機會被喚醒并檢查條件，從而避免死鎖。
如何選擇合適的阻塞隊列？
Java提供了多種阻塞隊列，例如 ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue、DelayQueue、SynchronousQueue 等。選擇合適的阻塞隊列取決于具體的應用場景：
- ArrayBlockingQueue: 基于數組實現的有界阻塞隊列，需要指定容量。適合于生產者和消費者速度相對穩定的場景。
- LinkedBlockingQueue: 基于鏈表實現的阻塞隊列，可以選擇有界或無界。無界隊列可能導致內存溢出，需要謹慎使用。適合于生產者和消費者速度差異較大的場景。
- PriorityBlockingQueue: 具有優先級的無界阻塞隊列。元素必須實現 Comparable 接口。適合于需要按照優先級處理任務的場景。
- DelayQueue: 延遲隊列，元素需要在指定的延遲時間后才能被消費。元素必須實現 Delayed 接口。適合于實現定時任務的場景。
- SynchronousQueue: 同步隊列，每個插入操作必須等待一個相應的移除操作，反之亦然。可以看作是一個容量為 0 的阻塞隊列。適合于傳遞性場景，例如線程池。
選擇阻塞隊列時，需要考慮以下因素：
- 是否有界: 有界隊列可以防止內存溢出，但可能導致生產者阻塞。無界隊列則沒有這個限制，但需要注意內存消耗。
- 是否需要優先級: 如果需要按照優先級處理任務，則需要選擇 PriorityBlockingQueue。
- 是否需要延遲: 如果需要在指定的延遲時間后才能消費元素，則需要選擇 DelayQueue。
- 并發性能: 不同阻塞隊列的并發性能可能有所不同，需要根據實際情況進行選擇。
除了wait/notify，還有哪些實現生產者消費者模式的方式？
除了 wait()/notify() 機制，Java還提供了其他方式來實現生產者消費者模式，例如：
-
BlockingQueue: Java并發包 java.util.concurrent 提供的阻塞隊列接口。它簡化了生產者消費者模式的實現，無需手動管理鎖和等待/通知機制。例如上面的代碼可以改為使用 ArrayBlockingQueue。
import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; public class ProducerConsumerBlockingQueue { private static final int CAPACITY = 5; private final BlockingQueue<Integer> buffer = new ArrayBlockingQueue<>(CAPACITY); class Producer implements Runnable { @Override public void run() { int i = 0; while (true) { try { buffer.put(i); // 阻塞直到隊列有空間 System.out.println("Produced: " + i); i++; Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } } class Consumer implements Runnable { @Override public void run() { while (true) { try { int value = buffer.take(); // 阻塞直到隊列有元素 System.out.println("Consumed: " + value); Thread.sleep((long) (Math.random() * 200)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } } public static void main(String[] args) { ProducerConsumerBlockingQueue pc = new ProducerConsumerBlockingQueue(); new Thread(pc.new Producer()).start(); new Thread(pc.new Consumer()).start(); new Thread(pc.new Consumer()).start(); } }使用 BlockingQueue 可以大大簡化代碼，并提高可讀性。put() 方法在隊列滿時會阻塞，take() 方法在隊列空時會阻塞，無需手動處理等待和通知。
-
ReentrantLock 和 Condition: ReentrantLock 提供了比 synchronized 更靈活的鎖機制，Condition 則提供了比 wait()/notify() 更精細的線程等待和通知機制。 可以實現更復雜的同步邏輯。
import java.util.LinkedList; import java.util.Queue; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class ProducerConsumerReentrantLock { private static final int CAPACITY = 5; private final Queue<Integer> buffer = new LinkedList<>(); private final Lock lock = new ReentrantLock(); private final Condition notFull = lock.newCondition(); private final Condition notEmpty = lock.newCondition(); class Producer implements Runnable { @Override public void run() { int i = 0; while (true) { lock.lock(); try { while (buffer.size() == CAPACITY) { System.out.println("Buffer is full, Producer is waiting"); notFull.await(); // 緩沖區滿，生產者等待 } buffer.offer(i); System.out.println("Produced: " + i); i++; notEmpty.signalAll(); // 通知消費者 Thread.sleep((long) (Math.random() * 100)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { lock.unlock(); } } } } class Consumer implements Runnable { @Override public void run() { while (true) { lock.lock(); try { while (buffer.isEmpty()) { System.out.println("Buffer is empty, Consumer is waiting"); notEmpty.await(); // 緩沖區空，消費者等待 } int value = buffer.poll(); System.out.println("Consumed: " + value); notFull.signalAll(); // 通知生產者 Thread.sleep((long) (Math.random() * 200)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } finally { lock.unlock(); } } } } public static void main(String[] args) { ProducerConsumerReentrantLock pc = new ProducerConsumerReentrantLock(); new Thread(pc.new Producer()).start(); new Thread(pc.new Consumer()).start(); new Thread(pc.new Consumer()).start(); } }在這個例子中，notFull 和 notEmpty 是兩個 Condition 對象，分別用于生產者和消費者的等待和通知。await() 方法類似于 wait()，signalAll() 方法類似于 notifyAll()。 使用 ReentrantLock 和 Condition 可以實現更細粒度的控制，例如可以只喚醒特定的生產者或消費者。
選擇哪種方式取決于具體的需求。 如果只是簡單的生產者消費者模式，使用 BlockingQueue 是最簡單和推薦的方式。 如果需要更靈活的控制，可以使用 ReentrantLock 和 Condition。 而 wait()/notify() 機制則是最基礎的方式，理解它可以幫助我們更好地理解并發編程的原理。
以上就是Java中如何實現生產者消費者模式 詳解w<a
