Java面試——寫一個生產者與消費者

一、通過synchronize 中的 wait 和 notify 實現

---

【1】我們可以將生產者和消費者需要的方法寫在公共類中

package com.yintong.concurrent;

import java.util.LinkedList;

public class Concurrentcomm {
    //常量
    private static int MAX_VALUE = 10;
    //可以理解為緩存
    LinkedList<String> linkedList = new LinkedList<>();
    Object object = new Object();
    /*
     * 生產者方法
     */
    public void product() throws Exception {
        synchronized(linkedList) {
            while(MAX_VALUE == linkedList.size()) {
                System.out.println("倉庫已滿，【生產者】： 暫時不能執行生產任務!");
                linkedList.wait();
            }
            linkedList.push("   李四    ");
            System.out.println("【生產者】：生產了一個產品\t【現倉儲量為】:" + linkedList.size());
            linkedList.notifyAll();
        }
    }
    /*
     * 消費者方法
     */
    public void customer() throws Exception {
        /*
         * 根據jdk的void notifyAll()的描述，“解除那些在該對象上調用wait()方法的線程的阻塞狀態。該方法只能在同步方法或同步塊內部調用。
         * 如果當前線程不是對象所得持有者，
         * 該方法拋出一個java.lang.IllegalMonitorStateException 異常”
         * so我們使用同一把鎖
         */
        synchronized (linkedList) {
                        //多線程判斷中使用 while 不要使用 if 否則會出現虛假喚醒問題
            while(linkedList.size() == 0) {
                System.out.println("倉庫無貨，【消費者】： 暫時不能執行消費任務!");
                linkedList.wait();
            }
            linkedList.pop();
            System.out.println("【消費者】：消費了一個產品\t【現倉儲量為】:" + linkedList.size());
            linkedList.notifyAll();
        }
    }
}

【2】在 main 函數中調用生產者和消費者方法，並加限制即可

/**
 * @author zzx
 * @desc 生產者與消費者
 *
 */
public class Concurrent {
    //常量
    private static int MAX_VALUE = 100;

    public static void main(String[] args) {
        Concurrentcomm con = new Concurrentcomm();
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    for (int i = 0; i < MAX_VALUE; i++) {
                        Thread.sleep(0);
                        con.product();
                    }
                } catch (Exception e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }).start();
        // 消費者
        new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    Thread.sleep(10);
                    for (int i = 0; i < MAX_VALUE; i++) {
                        con.customer();
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

【3】簡單的生產者與消費者模式就完成了，可以看下運行的結果

 

二、通過 Lock 中的 await 與 signalAll 實現

---

【1】我們將公共的屬性和方法放在 Resouce 類中，在資源類中使用 Lock 中的 lock()進行加鎖，控制並發操作。使用 await()方法阻塞線程。使用 signalAll()喚醒線程。

/**
 * 通過 Lock 實現生產者與消費者
 * 資源類：將公共的資源放在一個單獨的類中，可以將其看做一個產品，自身就就有生產和消費的能力（方法）
 */
public class ProductAndConsumer {
    public static void main(String[] args) {
        Resouce resouce = new Resouce();
        //生產者
        new Thread(()->{
            for (int i=1;i<=5;i++) {
                resouce.product();
            }
        },String.valueOf("生產者")) .start();

        //消費者
        new Thread(()->{
            for (int i=1;i<=5;i++){
                resouce.consumer();
            }
        },String.valueOf("消費者")).start();
    }
}
//資源類
class Resouce {
    private int MAX_VALUE = 3;
    private int MIN_VALUE = 0;
    private int number = 0;
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();

    //生產者
    public void product(){
        try {
            lock.lock();
            //如果生產的數量大於最大值則阻塞
            while(number >= MAX_VALUE){
                condition.await();
            }
            number++;
            System.out.println("【生產者】：生產了一個產品\t【現倉儲量為】:" + number);
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }

    //消費者
    public void  consumer(){
        try {
            lock.lock();
            //如果消費的值=0則阻塞
            while(number <= MIN_VALUE){
                condition.await();
            }
            number--;
            System.out.println("【消費者】：消費了一個產品\t【現倉儲量為】:" + number);
            condition.signalAll();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
    }
}

【2】輸出結果展示：
  ​

三、synchronized 和 Lock 的區別

---

【1】原始構成：synchronized 是關鍵字屬於 JVM 層面。底層通過 monitorenter（進入）monitorexit（退出）實現。底層是通過 monitor 對象完成，其實 wait/notify 等方法也依賴於 monitor 對象，只有在同步塊或方法中才能調用 wait/notify 等方法。Lock 是具體類（java.util.concurrent.locks.Lock）是 API 層面的鎖。
【2】使用方法：synchronized 不需要用戶手動釋放鎖，當 synchronized 代碼執行完后，系統會自動釋放鎖。ReentrantLock 則需要用戶手動釋放鎖，若未主動釋放鎖，就可能導致出現死鎖的現象。
【3】等待是否中斷：synchronized 不可中斷，除非拋出異常或者正常運行完成。ReentrantLock 可中斷，1）、設置超時時間 tryLock(long timeout,TimeUnit unit) 2）、lockInterruptibly() 放在代碼塊中，調用 interrupt() 方法可中斷。
【4】加鎖是否公平：synchronized 非公平鎖。ReentrantLock 兩者都可以，默認是非公平鎖，構造方法可以傳入 boolean 值，true 為公平鎖，false 為非公平鎖。
【5】鎖綁定多個條件 Condition：synchronized 沒有。ReentrantLock 用來實現分組喚醒需要喚醒的線程們，可以精確喚醒，而不是像 synchronized 要么隨機喚醒一個線程要么喚醒全部線程。

四、通過阻塞隊列實現生產者與消費者

---

【1】通過blockQueue 中的 put/take 方法實現生產者與消費者，具體實現如下：當生產者使用put 生產到指定的隊列大小3時，就會阻塞當前線程。這是消費者線程會通過 take 方法消費隊列中的消息。當隊列中沒有消息時，會阻塞，直到有消息消費。

public class BlockProductConsumer {
    public static void main(String[] args) {
        MyResouce resouce = new MyResouce(new ArrayBlockingQueue(3));
        //生產者線程
        new Thread(()->{
            for(int i=1;i<=10;i++){
                resouce.product();
            }
        },"生產者").start();

        //消費者線程
        new Thread(()->{
            for(int i=1;i<=10;i++){
                try {
                    resouce.consumer();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        },"消費者").start();

        try {
            TimeUnit.SECONDS.sleep(1);
            resouce.stop();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}


/**
 * 公共資源類
 */
class MyResouce{
    //標記 while 無限循環
    private volatile boolean FLAG = true;
    //隊列中存入的數值
    private AtomicInteger atomicInteger = new AtomicInteger();
    //組合一個阻塞隊列，通過構造器傳入
    private BlockingQueue blockingQueue;
    public MyResouce(BlockingQueue blockingQueue) {
        this.blockingQueue = blockingQueue;
    }

    //生產者
    public void product(){
        try {
            while (FLAG){
                blockingQueue.put(String.valueOf(atomicInteger.incrementAndGet()));
                System.out.println("生產者生產第"+blockingQueue.size()+"個產品");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    //消費者
    public void consumer() throws InterruptedException {
        while (FLAG){
            blockingQueue.take();
            System.out.println("消費者消費第"+(blockingQueue.size()+1)+"個產品");
        }
    }

    public void stop(){
        FLAG = false;
        System.out.println("========================");
    }
}

【2】效果展示：
  ​​