生產者消費者問題總結

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

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了。