摘要：創(chuàng)建一個(gè)阻塞隊(duì)列生產(chǎn)者生產(chǎn)，目前總共有消費(fèi)者消費(fèi)，目前總共有原文鏈接更多教程

生產(chǎn)者和消費(fèi)者問題是線程模型中老生常談的問題，也是面試中經(jīng)常遇到的問題。光在Java中的實(shí)現(xiàn)方式多達(dá)數(shù)十種，更不用說加上其他語言的實(shí)現(xiàn)方式了。那么我們?cè)撊绾螌W(xué)習(xí)呢？

本文會(huì)通過精講wait()和notify()方法實(shí)現(xiàn)生產(chǎn)者-消費(fèi)者模型，來學(xué)習(xí)生產(chǎn)者和消費(fèi)者問題的原理。

目的是當(dāng)你理解了最簡(jiǎn)單實(shí)現(xiàn)原理，再看其他的實(shí)現(xiàn)，無非使用了更高級(jí)的機(jī)制（例如鎖、信號(hào)量、管道等等）來照貓畫虎的實(shí)現(xiàn)這個(gè)原理，萬變不離其宗，它們的原理都是一樣的。

本文也會(huì)列出一部分其他的實(shí)現(xiàn)方式代碼。千萬不要嘗試去背誦所有實(shí)現(xiàn)代碼，只有掌握了實(shí)現(xiàn)原理才能遇到問題的時(shí)候游刃有余。

啥是生產(chǎn)者-消費(fèi)者模型：

生產(chǎn)者和消費(fèi)者在同一時(shí)間段內(nèi)共用同一個(gè)存儲(chǔ)空間，生產(chǎn)者往存儲(chǔ)空間中添加產(chǎn)品，消費(fèi)者從存儲(chǔ)空間中取走產(chǎn)品，當(dāng)存儲(chǔ)空間為空時(shí)，消費(fèi)者阻塞，當(dāng)存儲(chǔ)空間滿時(shí)，生產(chǎn)者阻塞。

現(xiàn)實(shí)生活中的例子：12306搶購(gòu)火車票、淘寶購(gòu)買商品、倉(cāng)庫(kù)管理等。

分步的實(shí)現(xiàn)我們的模型

public class Test1 {
    private static Integer count = 0;  //代表生產(chǎn)的商品數(shù)量
    private static final Integer FULL = 10;  //代表商品最多多少個(gè)（也就是緩沖區(qū)大?。?    private static final Object LOCK = new Object(); //鎖對(duì)象  ----分析1

    public static void main(String[] args) {
        for (int i = 0; i < 5; i++) {   //創(chuàng)造一堆生產(chǎn)者和消費(fèi)者模擬真實(shí)環(huán)境
            new Thread(new Producer()).start();
        }
        for (int i = 0; i < 5; i++) {
            new Thread(new Consumer()).start();
        }
    }

    static class Producer implements Runnable { //代表生產(chǎn)者
        @Override
        public void run() {
           
        }
    }

    static class Consumer implements Runnable {  //代表消費(fèi)者
        @Override
        public void run() {
          
        }
    }
}

分析1.在main函數(shù)中創(chuàng)建了5個(gè)消費(fèi)者線程任務(wù)和5個(gè)生產(chǎn)者線程任務(wù)，當(dāng)這10個(gè)線程同時(shí)運(yùn)行時(shí)，需要保證生產(chǎn)者和消費(fèi)者所公用的緩沖區(qū)是同步被改變的，就是說不同線程訪問緩沖區(qū)的數(shù)據(jù)不能發(fā)生錯(cuò)亂。這里就是用一個(gè)鎖來保證緩沖區(qū)每次只有一個(gè)線程訪問

接下來看下生產(chǎn)者和消費(fèi)者的實(shí)現(xiàn)：

static class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {  //一次多生產(chǎn)幾個(gè)商品
                try {
                    Thread.sleep(3000);  //模擬真實(shí)環(huán)境，讓生產(chǎn)的慢一點(diǎn)，間隔3秒
                } catch (Exception e) {
                    e.printStackTrace();
                }

                synchronized (LOCK) {  //線程同步  
                    while (count.equals(FULL)) {  //當(dāng)緩沖區(qū)滿了  
                        try {
                            LOCK.wait();   //讓線程等待  ----分析1
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count++;  //緩沖區(qū)不滿時(shí)繼續(xù)生產(chǎn)商品，商品加一
                    System.out.println(Thread.currentThread().getName() + "生產(chǎn)者生產(chǎn)，目前總共有" + count);
                    LOCK.notifyAll(); //喚醒等待的消費(fèi)者
                }
            }
        }
    }

    static class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                synchronized (LOCK) {  
                    while (count == 0) {  //當(dāng)沒有商品時(shí)，需要等待生產(chǎn)者生產(chǎn)商品
                        try {
                            LOCK.wait();   //----分析 2
                        } catch (Exception e) {
                        }
                    }
                    count--; //商品被消耗，商品減一
                    System.out.println(Thread.currentThread().getName() + "消費(fèi)者消費(fèi)，目前總共有" + count);
                    LOCK.notifyAll();  //商品被消耗后，通知等待的生產(chǎn)者
                }
            }
        }
    }

分析：
1.當(dāng)緩沖區(qū)滿了的時(shí)候，需要阻止生產(chǎn)者繼續(xù)生產(chǎn)商品
2.當(dāng)緩沖區(qū)為空，沒有商品時(shí)，需要阻止消費(fèi)者繼續(xù)消費(fèi)商品

相信代碼分析和詳細(xì)的注釋，你已經(jīng)能很好的理解這個(gè)生產(chǎn)者-消費(fèi)者模型的原理了。接下來貼出其他的幾種實(shí)現(xiàn)代碼。

使用鎖實(shí)現(xiàn)：

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

public class Test1 {
    private static Integer count = 0;
    private static final Integer FULL = 10;
    //創(chuàng)建一個(gè)鎖對(duì)象
    private Lock lock = new ReentrantLock();
    //創(chuàng)建兩個(gè)條件變量，一個(gè)為緩沖區(qū)非滿，一個(gè)為緩沖區(qū)非空
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
    public static void main(String[] args) {
        Test1 test1 = new Test1();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                //獲取鎖
                lock.lock();
                try {
                    while (count == FULL) {
                        try {
                            notFull.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生產(chǎn)者生產(chǎn)，目前總共有" + count);
                    //喚醒消費(fèi)者
                    notEmpty.signal();
                } finally {
                    //釋放鎖
                    lock.unlock();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                lock.lock();
                try {
                    while (count == 0) {
                        try {
                            notEmpty.await();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消費(fèi)者消費(fèi)，目前總共有" + count);
                    notFull.signal();
                } finally {
                    lock.unlock();
                }
            }
        }
    }
}

使用阻塞隊(duì)列：
當(dāng)隊(duì)列滿了或空了的時(shí)候進(jìn)行入隊(duì)列操作都會(huì)被阻塞。

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

public class Test1 {
    private static Integer count = 0;
    //創(chuàng)建一個(gè)阻塞隊(duì)列
    final BlockingQueue blockingQueue = new ArrayBlockingQueue<>(10);
    public static void main(String[] args) {
        Test1 test1 = new Test1();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
        new Thread(test1.new Producer()).start();
        new Thread(test1.new Consumer()).start();
    }
    class Producer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                try {
                    blockingQueue.put(1);
                    count++;
                    System.out.println(Thread.currentThread().getName()
                            + "生產(chǎn)者生產(chǎn)，目前總共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    class Consumer implements Runnable {
        @Override
        public void run() {
            for (int i = 0; i < 10; i++) {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
                try {
                    blockingQueue.take();
                    count--;
                    System.out.println(Thread.currentThread().getName()
                            + "消費(fèi)者消費(fèi)，目前總共有" + count);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}