Java中生產者消費者問題總結

生產者-消費者算是并發編程中常見的問題。依靠緩沖區我們可以實現生產者與消費者之間的解耦。生產者只管往緩沖區里面放東西，消費者只管往緩沖區里面拿東西。這樣我們避免生產者想要交付數據給消費者，但消費者此時還無法接受數據這樣的情況發生。

wait notify

這個問題其實就是線程間的通訊，所以要注意的是不能同時讀寫。生產者在緩沖區滿的時候不生產，等待；消費者在緩沖區為空的時候不消費，等待。比較經典的做法是wait和notify。

生產者線程執行15次set操作

public class Producer implements Runnable{ private Channel channel; public Producer(Channel channel) { this.channel = channel; } @Override public void run() { for(int i=0;i<15;i++){ channel.set(Thread.currentThread().getName()+' '+i); } }}

消費者線程執行10次get操作

public class Consumer implements Runnable { private Channel channel; public Consumer(Channel channel) { this.channel = channel; } @Override public void run() { for(int i=0;i<10;i++){ System.out.println('Consumer '+Thread.currentThread().getName()+' get '+channel.get()); } }}

現在定義Channel類,并創建兩個生產者線程和三個消費者線程

public class Channel { private List<String> buffer=new ArrayList<>(); private final int MAX_SIZE=10; public synchronized String get(){ while (buffer.size()==0){//不要用if，醒來了也要再次判斷 try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } } String str=buffer.remove(0); notifyAll(); return str; } public synchronized void set(String str){ while (buffer.size()==MAX_SIZE){ try { wait(); } catch (InterruptedException e) { e.printStackTrace(); } } buffer.add(str); notifyAll(); } public static void main(String[] args) { Channel channel=new Channel(); Producer producer=new Producer(channel); Consumer consumer=new Consumer(channel); for(int i=0;i<2;i++){ new Thread(producer).start(); } for (int i=0;i<3;i++){ new Thread(consumer).start(); } }}

使用notifyAll而不是notify的原因是，notify有可能出現多次喚醒同類的情況，造成“假死”。我們可以使用Condition來實現更精確的喚醒。

Condition

將上面代碼中的Channel類修改一下即可

public class Channel { private List<String> buffer=new ArrayList<>(); private final int MAX_SIZE=10; private Lock lock=new ReentrantLock(); private Condition producer=lock.newCondition(); private Condition consumer=lock.newCondition(); public String get(){ String str=null; try { lock.lock(); while (buffer.size()==0){ consumer.await(); } str=buffer.remove(0); producer.signalAll(); }catch (InterruptedException e) { e.printStackTrace(); }finally { lock.unlock(); } return str; } public void set(String str){ try { lock.lock(); while (buffer.size()==MAX_SIZE){ producer.await(); } buffer.add(str); consumer.signalAll(); }catch (InterruptedException e) { e.printStackTrace(); }finally { lock.unlock(); } }}

雙緩沖與Exchanger

當同步的花銷非常大時，我們可以采用雙緩沖區的辦法。雙緩沖的一個好處就在于：因為生產者和消費者各自擁有一個緩沖區，所以他們不會同時對同一個緩沖區進行操作，那么我們就不需要為讀寫操作加鎖，用空間換了時間。在Java中可以通過Exchanger來交換兩個線程之間的數據結構。

public class Producer implements Runnable{ private List<String> buffer; private Exchanger<List<String>> exchanger; public Producer(List<String> buffer, Exchanger<List<String>> exchanger){ this.buffer=buffer; this.exchanger=exchanger; } @Override public void run() { for(int i=0;i<10;i++){ for (int j=0;j<10;j++) buffer.add('Thrad '+Thread.currentThread().getName()+' : '+i+' '+j); try { buffer=exchanger.exchange(buffer); } catch (InterruptedException e) { e.printStackTrace(); } } }}public class Consumer implements Runnable { private Exchanger<List<String>> exchanger; private List<String> buffer; public Consumer(List<String> buffer,Exchanger<List<String>> exchanger) { this.exchanger = exchanger; this.buffer = buffer; } @Override public void run() { for(int i=0;i<10;i++){ try { buffer=exchanger.exchange(buffer); } catch (InterruptedException e) { e.printStackTrace(); } for(int j=0;j<10;j++){ String message=buffer.get(0); System.out.println(message); buffer.remove(0); } } }}public class Main { public static void main(String[] args) { List<String> buffer1=new ArrayList<>(); List<String> buffer2=new ArrayList<>(); Exchanger<List<String>> exchanger=new Exchanger<>(); Producer producer=new Producer(buffer1,exchanger); Consumer consumer=new Consumer(buffer2,exchanger); Thread t1=new Thread(producer); Thread t2=new Thread(consumer); t1.start(); t2.start(); }}

BlockingQueue

我們可以使用更為方便安全的阻塞式集合來實現生產消費者模型。

這類集合具有的特點是：當集合已滿或者是為空的時候，被調用的方法不會立即執行，該方法將被阻塞，直到可以成功執行為止。

public class Channel { private BlockingQueue<String> blockingQueue=new ArrayBlockingQueue<>(10); public String get(){ String str=null; try { str=blockingQueue.take(); } catch (InterruptedException e) { e.printStackTrace(); } return str; } public void set(String str){ try { blockingQueue.put(str); } catch (InterruptedException e) { e.printStackTrace(); } }}

這次的Channel類是不是比之前的簡潔了許多，有了BlockingQueue我們就不用再去寫wait和notify了。

到此這篇關于Java中生產者消費者問題總結的文章就介紹到這了,更多相關Java生產者消費者內容請搜索好吧啦網以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持好吧啦網！