JAVA多线程-生产者与消费者模型

概述

生产者-消费者模型是多线程编程中的一个经典模型，主要描述的是生产者和消费者在同一时间段内共用同一块存储空间（通常也称为缓冲区，类似于仓库的概念），工作时，生产者向缓冲区去存放数据，而消费者则从缓冲区中取走数据。

特点

  1. 生产者在生产的时候，消费者不能消费，需要等待。
  2. 消费者在消费的时候，生产者不能生产，需要等待。
  3. 当缓冲区已满时，生产者暂停生产，自动阻塞；此时需要通知消费者去消费。
  4. 当缓冲区为空时，消费者暂停消费，自动阻塞；此时需要通知生产者去生产。

解决生产者/消费者问题的方法

  1. 采用某种机制维护生产者和消费者之间的同步。比如：
<1>. 上述特点1和特点2需要通过同步来解决，比如synchronize，Lock等
<2>. 上述特点3和特点4则需要涉及到线程间的通信了，生产者线程生产数据放入缓冲区后，通知消费者
线程取出数据;消费者线程取出数据后，通知生产者生产数据，比如可以利用wait()/notify()机制来实现。
  1. 在生产者和消费者之间建立一个管道

以上第一种方式比较常用，第二种方式见得不多。

实现方式

  1. wait()/notify()方法
  2. await()/signal()方法
  3. BlockingQueue阻塞队列
  4. PipedInputStream / PipedOutputStream 管道流

以上前三个是同步机制实现，第四个是管道方式，目前不太清楚。

1. wait()/notify()方法

wait()方法：当缓冲区已满/空时，生产者/消费者线程停止自己的执行，并且释放锁资源，使得自己处于等待（阻塞）状态，让其他线程执行。

notify()方法：当生产者/消费者向缓冲区中放入/取出一个产品时，向其他等待的线程发出可执行的通知，同时释放锁资源，使得自己处于等待状态。

Storage.java仓库类：

package com.feizi.java.concurrency.model.one;

import java.util.LinkedList;

/**
 * wait()方法：当缓冲区已满/空时，生产者/消费者线程停止自己的执行，并且释放锁资源，使得自己处于等待（阻塞）状态，让其他线程执行。
 * notify()方法：当生产者/消费者向缓冲区中放入/取出一个产品时，向其他等待的线程发出可执行的通知，同时释放锁资源，使得自己处于等待状态。
 * Created by feizi on 2018/5/28.
 */
public class Storage {
    //仓库最大存储量
    private static final int MAX_COUNT = 10;

    //仓库存储的载体
    private LinkedList<Object> list = new LinkedList<>();

    /**
     * 生产产品-同步方法
     * @param producer
     */
    public synchronized void produce(String producer){
        while (list.size() == MAX_COUNT){
            System.out.println("【仓库已满】，" + producer + "：暂时不能执行生产任务...");
            try {
                //仓库已满，暂不生产，生产阻塞
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //生产产品
        list.add(new Object());
        System.out.println( producer + "：=====》生产了一个产品\t，【仓库储量】为：" + list.size());
        notifyAll();
    }

    /**
     * 消费产品-同步方法
     * @param consumer
     */
    public synchronized void consume(String consumer){
        while (list.size() == 0){
            System.out.println("【仓库已空】， " + consumer + "：暂不消费...");
            try {
                //仓库已空，暂不消费，消费阻塞
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        //消费产品
        list.remove();
        System.out.println(consumer + "：================》消费了一个产品\t，【仓库储量】为：" + list.size());
        notifyAll();
    }

    /**
     * 生产产品-同步块
     * @param producer
     */
    public void produce1(String producer){
        synchronized (list){
            while (list.size() == MAX_COUNT){
                try {
                    System.out.println("【仓库已满】，" + producer + "：暂时不能执行生产任务...");
                    //仓库已满，暂不生产，生产阻塞
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //生产产品
            list.add(new Object());
            System.out.println( producer + "：=====》生产了一个产品\t，【仓库储量】为：" + list.size());
            list.notifyAll();
        }
    }

    /**
     * 消费产品-同步块
     * @param consumer
     */
    public void consume1(String consumer){
        synchronized (list){
            while (list.size() == 0){
                try {
                    System.out.println("【仓库已空】， " + consumer + "：暂不消费...");
                    //仓库已空，暂不消费，消费阻塞
                    list.wait();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //消费产品
            list.remove();
            System.out.println(consumer + "：================》消费了一个产品\t，【仓库储量】为：" + list.size());
            list.notifyAll();
        }
    }
}

Producer.java生产者类：

package com.feizi.java.concurrency.model.one;

import java.util.concurrent.TimeUnit;

/**
 * Created by feizi on 2018/5/28.
 */
public class Producer implements Runnable {
    /*生产者名称*/
    private String name;
    /*仓库*/
    private Storage storage;

    public Producer(String name, Storage storage) {
        this.name = name;
        this.storage = storage;
    }

    @Override
    public void run() {
        while (true){
            try {
                storage.produce(name);
//                storage.produce1(name);
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Consumer.java消费者类：

package com.feizi.java.concurrency.model.one;

import java.util.concurrent.TimeUnit;

/**
 * Created by feizi on 2018/5/28.
 */
public class Consumer implements Runnable {
    /*消费者名称*/
    private String name;
    /*仓库*/
    private Storage storage;

    public Consumer(String name, Storage storage) {
        this.name = name;
        this.storage = storage;
    }

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.MILLISECONDS.sleep(2500);
                storage.consume(name);
//                storage.consume1(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

TestOne.java测试类：

package com.feizi.java.concurrency.model.one;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by feizi on 2018/5/28.
 */
public class TestOne {
    public static void main(String[] args) {
        Storage storage = new Storage();
        /*for (int i = 0; i < 2; i++){
            new Thread(new Producer("生产者【" + i + "】", storage)).start();
        }

        for (int i = 0; i < 5; i++){
            new Thread(new Consumer("消费者【" + i + "】" , storage)).start();
        }*/

        ExecutorService threadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 2; i++){
            threadPool.execute(new Producer("生产者【" + i + "】", storage));
        }

        for (int i = 0; i < 5; i++){
            threadPool.execute(new Consumer("消费者【" + i + "】" , storage));
        }

        threadPool.shutdown();
    }
}

控制台输出结果：

生产者【1】：=====》生产了一个产品    ，【仓库储量】为：1
生产者【0】：=====》生产了一个产品    ，【仓库储量】为：2
生产者【1】：=====》生产了一个产品    ，【仓库储量】为：3
生产者【0】：=====》生产了一个产品    ，【仓库储量】为：4
消费者【0】：================》消费了一个产品 ，【仓库储量】为：3
消费者【5】：================》消费了一个产品 ，【仓库储量】为：2
消费者【3】：================》消费了一个产品 ，【仓库储量】为：1
消费者【4】：================》消费了一个产品 ，【仓库储量】为：0
【仓库已空】， 消费者【1】：暂不消费...
【仓库已空】， 消费者【2】：暂不消费...
生产者【0】：=====》生产了一个产品    ，【仓库储量】为：1
消费者【2】：================》消费了一个产品 ，【仓库储量】为：0
【仓库已空】， 消费者【1】：暂不消费...
生产者【1】：=====》生产了一个产品    ，【仓库储量】为：1
消费者【1】：================》消费了一个产品 ，【仓库储量】为：0

2.await() / signal()方法

await()和signal()的功能基本上和wait()/notify()类似，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更强的灵活性，通过在Lock对象上调用newCondition()方法，将条件变量和一个锁对象进行绑定，进而控制并发程序访问竞争资源的安全。

Storage.java仓库类：

package com.feizi.java.concurrency.model.two;

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

/**
 * await()和signal()的功能基本上和wait()/notify()类似，完全可以取代它们，但是它们和新引入的锁定机制Lock直接挂钩，具有更强的灵活性，
 * 通过在Lock对象上调用newCondition()方法，将条件变量和一个锁对象进行绑定，进而控制并发程序访问竞争资源的安全
 * Created by feizi on 2018/5/28.
 */
public class Storage {
    //仓库最大存储量
    private static final int MAX_COUNT = 10;

    //仓库存储的载体
    private LinkedList<Object> list = new LinkedList<>();

    //可重入锁
    private final Lock lock = new ReentrantLock();

    //仓库满的条件变量
    private final Condition full = lock.newCondition();

    //仓库空的条件变量
    private final Condition empty = lock.newCondition();

    /**
     * 生产产品
     * @param producer
     */
    public void produce(String producer){
        try {
            //获得锁
            lock.lock();
            while (list.size() == MAX_COUNT){
                try {
                    System.out.println("仓库已满，【" + producer + "】：暂时不能执行生产任务...");
                    //由于仓库已满，暂不生产，生产阻塞
                    full.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //生产产品
            list.add(new Object());

            System.out.println("===>【" + producer + "】：生产了一个产品，\t【仓库储量】为：" + list.size());

            empty.signalAll();
        } finally {
            //释放锁
            lock.unlock();
        }
    }

    /**
     * 消费产品
     * @param consumer
     */
    public void consume(String consumer){
        try {
            //获得锁
            lock.lock();

            while (list.size() == 0){
                try {
                    System.out.println("仓库已空，【" + consumer + "】：暂时不能执行消费任务...");
                    //由于仓库已空，暂不消费，消费阻塞
                    empty.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

            //消费产品
            list.remove();

            System.out.println("=============>【" + consumer + "】：消费了一个产品，\t【仓库储量】为：" + list.size());
            full.signalAll();
        }finally {
            //释放锁
            lock.unlock();
        }
    }
}

Producer.java生产者类：

package com.feizi.java.concurrency.model.two;

import java.util.concurrent.TimeUnit;

/**
 * Created by feizi on 2018/5/28.
 */
public class Producer implements Runnable {
    private String name;

    private Storage storage;

    public Producer(String name, Storage storage) {
        this.name = name;
        this.storage = storage;
    }

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.SECONDS.sleep(1);
                storage.produce(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Consumer.java消费者类：

package com.feizi.java.concurrency.model.two;

import java.util.concurrent.TimeUnit;

/**
 * Created by feizi on 2018/5/28.
 */
public class Consumer implements Runnable {
    private String name;

    private Storage storage;

    public Consumer(String name, Storage storage) {
        this.name = name;
        this.storage = storage;
    }

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.SECONDS.sleep(2);
                storage.consume(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

TestTwo.java测试类：

package com.feizi.java.concurrency.model.two;


import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by feizi on 2018/5/28.
 */
public class TestTwo {
    public static void main(String[] args) {
        Storage storage = new Storage();

        ExecutorService threadPool = Executors.newCachedThreadPool();
        for (int i = 0; i < 2; i++){
            threadPool.execute(new Producer("生产者" + i, storage));
        }

        for (int i = 0; i < 4; i++){
            threadPool.execute(new Consumer("消费者" + i, storage));
        }

        threadPool.shutdown();
    }
}

控制台输出结果：

===>【生产者3】：生产了一个产品， 【仓库储量】为：1
===>【生产者0】：生产了一个产品， 【仓库储量】为：2
===>【生产者4】：生产了一个产品， 【仓库储量】为：3
===>【生产者2】：生产了一个产品， 【仓库储量】为：4
===>【生产者1】：生产了一个产品， 【仓库储量】为：5
===>【生产者1】：生产了一个产品， 【仓库储量】为：6
=============>【消费者0】：消费了一个产品，   【仓库储量】为：5
===>【生产者3】：生产了一个产品， 【仓库储量】为：6
===>【生产者4】：生产了一个产品， 【仓库储量】为：7
===>【生产者2】：生产了一个产品， 【仓库储量】为：8
===>【生产者0】：生产了一个产品， 【仓库储量】为：9
=============>【消费者1】：消费了一个产品，   【仓库储量】为：8
===>【生产者1】：生产了一个产品， 【仓库储量】为：9
===>【生产者4】：生产了一个产品， 【仓库储量】为：10
仓库已满，【生产者0】：暂时不能执行生产任务...
仓库已满，【生产者3】：暂时不能执行生产任务...

3.BlockingQueue阻塞队列

BlockingQueue是一个已经在内部实现了同步的阻塞队列(由链表结构组成的有界阻塞队列)，实现方式采用的是await()/signal()方法，可以在生成对象时指定容量大小，用于阻塞操作的是put()和take()方法。

put()方法：类似于前面的生产者线程，容量达到最大时，自动阻塞

take()方法：类似于前面的消费者线程，容量为0时，启动阻塞

Storage.java仓库类：

package com.feizi.java.concurrency.model.three;

import java.util.concurrent.LinkedBlockingQueue;

/**
 * BlockingQueue是一个已经在内部实现了同步的阻塞队列，实现方式采用的是await()/signal()方法，
 * 可以在生成对象时指定容量大小，用于阻塞操作的是put()和take()方法
 * put()方法：类似于前面的生产者线程，容量达到最大时，自动阻塞
 * take()方法：类似于前面的消费者线程，容量为0时，启动阻塞
 * Created by feizi on 2018/5/28.
 */
public class Storage {
    //仓库最大存储量
    private static final int MAX_COUNT = 10;

    //仓库存储的载体
    private LinkedBlockingQueue<Object> list = new LinkedBlockingQueue<>(10);

    /**
     * 生产产品
     * @param producer
     */
    public void produce(String producer){
        //如果仓库已满
        if(list.size() == MAX_COUNT){
            System.out.println("仓库已满，【" + producer + "】：暂时不能执行生产任务...");
        }

        try {
            //生产产品
            list.put(new Object());
            System.out.println("===>【" + producer + "】：生产了一个产品，\t【仓库储量】为：" + list.size());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 消费产品
     * @param consumer
     */
    public void consume(String consumer){
        //如果仓库已空
        if (list.size() == 0){
            System.out.println("仓库已空，【" + consumer + "】：暂时不能执行消费任务...");
        }

        try {
            //消费产品
            list.take();
            System.out.println("================>【" + consumer + "】：消费了一个产品，\t【仓库储量】为：" + list.size());
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}

Producer.java生产者类：

package com.feizi.java.concurrency.model.three;

import java.util.concurrent.TimeUnit;

/**
 * Created by feizi on 2018/5/28.
 */
public class Producer implements Runnable {
    private String name;
    private Storage storage;

    public Producer(String name, Storage storage) {
        this.name = name;
        this.storage = storage;
    }

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.SECONDS.sleep(1);
                storage.produce(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

Consumer.java消费者类：

package com.feizi.java.concurrency.model.three;

import java.util.concurrent.TimeUnit;

/**
 * Created by feizi on 2018/5/28.
 */
public class Consumer implements Runnable {
    private String name;
    private Storage storage;

    public Consumer(String name, Storage storage) {
        this.name = name;
        this.storage = storage;
    }

    @Override
    public void run() {
        while (true){
            try {
                TimeUnit.SECONDS.sleep(2);
                storage.consume(name);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

TestThree.java测试类：

package com.feizi.java.concurrency.model.three;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * Created by feizi on 2018/5/28.
 */
public class TestThree {
    public static void main(String[] args) {
        Storage storage = new Storage();

        ExecutorService threadPool = Executors.newCachedThreadPool();

        for (int i = 0; i < 3; i++){
            threadPool.execute(new Producer("生产者" + i, storage));
        }

        for (int j = 0; j < 5; j++){
            threadPool.execute(new Consumer("消费者" + j, storage));
        }
    }
}

控制台输出结果：

===>【生产者2】：生产了一个产品， 【仓库储量】为：1
===>【生产者1】：生产了一个产品， 【仓库储量】为：2
===>【生产者0】：生产了一个产品， 【仓库储量】为：3
===>【生产者1】：生产了一个产品， 【仓库储量】为：4
===>【生产者2】：生产了一个产品， 【仓库储量】为：5
===>【生产者0】：生产了一个产品， 【仓库储量】为：6
================>【消费者4】：消费了一个产品，    【仓库储量】为：5
================>【消费者0】：消费了一个产品，    【仓库储量】为：4
================>【消费者1】：消费了一个产品，    【仓库储量】为：3
================>【消费者2】：消费了一个产品，    【仓库储量】为：2
================>【消费者3】：消费了一个产品，    【仓库储量】为：1
===>【生产者1】：生产了一个产品， 【仓库储量】为：2
===>【生产者0】：生产了一个产品， 【仓库储量】为：3
===>【生产者2】：生产了一个产品， 【仓库储量】为：4
===>【生产者2】：生产了一个产品， 【仓库储量】为：5
===>【生产者1】：生产了一个产品， 【仓库储量】为：6
===>【生产者0】：生产了一个产品， 【仓库储量】为：7
================>【消费者4】：消费了一个产品，    【仓库储量】为：6
================>【消费者0】：消费了一个产品，    【仓库储量】为：5

原文参考

感谢以下作者的分享，受益匪浅~~

  1. https://www.cnblogs.com/Ming8006/p/7243858.html
  2. https://www.cnblogs.com/xrq730/p/4855663.html
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 206,311评论 6赞 481
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 88,339评论 2赞 382
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 152,671评论 0赞 342
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 55,252评论 1赞 279
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 64,253评论 5赞 371
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 49,031评论 1赞 285
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,340评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,973评论 0赞 259
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 43,466评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,937评论 2赞 323
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 38,039评论 1赞 333
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,701评论 4赞 323
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,254评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 30,259评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,485评论 1赞 262
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,497评论 2赞 354
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,786评论 2赞 345

推荐阅读更多精彩内容