生产者消费者模型（并发）

1 概述

在并发编程中使用生产者和消费者模式能够解决绝大多数并发问题。

该模式通过平衡生产进程和消费进程的工作能力来提高程序的整体处理数据的速度。

优点

  • 解耦
  • 支持并发
  • 支持忙闲不均

 

2 为什么要使用生产者和消费者模式

在进程世界里，生产者就是生产数据的进程，消费者就是消费数据的进程。

在多进程开发当中，如果生产者处理速度很快，而消费者处理速度很慢，那么生产者就必须等待消费者处理完，才能继续生产数据。

同样的道理，如果消费者的处理能力大于生产者，那么消费者就必须等待生产者。

为了解决这个问题于是引入了生产者和消费者模式。

3 什么是生产者消费者模式

生产者消费者模式是通过一个容器来解决生产者和消费者的强耦合问题。

生产者和消费者彼此之间不直接通讯，而通过阻塞队列来进行通讯，

所以生产者生产完数据之后不用等待消费者处理，直接扔给阻塞队列，

消费者不找生产者要数据，而是直接从阻塞队列里取，

4 阻塞队列就相当于一个缓冲区，平衡了生产者和消费者的处理能力。

5 如何实现

实现生产者消费者模型有两种方式：

  1. 采用 wait—notify 方式实现生产者消费者模型（注意这里需要加同步锁 synchronized）
  2. 采用 阻塞队列 方式实现生产者消费者模式

  （1）wait-notify 方式

这里设置了生产者生产速度大于消费者消费速度（通过 sleep() 方法实现）。

缓冲区 BufferArea.java：

public class BufferArea {

    // 当前资源数量的计数值
    private int currNum = 0;

    // 资源池中允许存放的资源数目
    private int maxSize = 10;

    /**
     * 从资源池中取走资源
     */
    public synchronized void get() {
        if (currNum > 0) {
            currNum--;
            System.out.println("Cosumer_" + Thread.currentThread().getName() + "消耗一件资源，" + "当前缓冲区有" + currNum + "个");
            // 通知生产者生产资源
            notifyAll();
        } else {
            try {
                // 如果没有资源，则 Cosumer_ 进入等待状态
                System.out.println("Cosumer_" + Thread.currentThread().getName() + ": 当前缓冲区资源不足，进入等待状态");
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 向缓冲区中添加资源
     */
    public synchronized void put() {
        // 若当前缓冲区内的资源计数小于最大 size 数，才加
        if (currNum < maxSize) {
            currNum++;
            System.out.println(Thread.currentThread().getName() + "生产一件资源，当前资源池有" + currNum + "个");

            // 通知等待的消费者
            notifyAll();
        } else {
            // 若当前缓冲区的资源计数大于最大 size 数，则等待
            try {
                System.out.println(Thread.currentThread().getName() + "线程进入等待 << 当前缓冲区的资源计数大于最大 size 数");
                // 生产者进入等待状态，并释放锁
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

生产者 Producer.java：

public class Producer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Producer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Producer_" + getName());
    }

    @Override
    public void run() {
        // 不断的生产资源
        while (true) {
            sleepSomeTime();
            mBufferArea.put();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消费者 Consumer：

public class Consumer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Consumer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Consumer_" + getName());
    }

    @Override
    public void run() {
        // 不断的取出资源
        while (true) {
            sleepSomeTime();
            mBufferArea.get();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

测试 Test.java：

public class Test {

    public static void main(String[] args) {
        BlockQueueBufferArea bufferArea = new BlockQueueBufferArea();

        Consumer consumer1 = new Consumer(bufferArea);
        Consumer consumer2 = new Consumer(bufferArea);
        Consumer consumer3 = new Consumer(bufferArea);

        Producer producer1 = new Producer(bufferArea);
        Producer producer2 = new Producer(bufferArea);
        Producer producer3 = new Producer(bufferArea);

        consumer1.start();
        consumer2.start();
        consumer3.start();

        producer1.start();
        producer2.start();
        producer3.start();

    }

}

打印结果如下

ProducerThread-5生产一件资源，当前资源池有1个
ProducerThread-4生产一件资源，当前资源池有2个
ProducerThread-3生产一件资源，当前资源池有3个
ProducerThread-5生产一件资源，当前资源池有4个
ProducerThread-4生产一件资源，当前资源池有5个
ProducerThread-3生产一件资源，当前资源池有6个
ProducerThread-5生产一件资源，当前资源池有7个
ProducerThread-4生产一件资源，当前资源池有8个
ProducerThread-3生产一件资源，当前资源池有9个
ProducerThread-3生产一件资源，当前资源池有10个
ProducerThread-4线程进入等待 << 当前缓冲区的资源计数大于最大 size 数
ProducerThread-5线程进入等待 << 当前缓冲区的资源计数大于最大 size 数
ProducerThread-3线程进入等待 << 当前缓冲区的资源计数大于最大 size 数

>> 注释：3个生产者线程生产满了10个（maxSize）产品，然后就都进入了等待

Cosumer_Consumer_Thread-0消耗一件资源，当前缓冲区有9个
Cosumer_Consumer_Thread-1消耗一件资源，当前缓冲区有8个
Cosumer_Consumer_Thread-2消耗一件资源，当前缓冲区有7个

>> 注释：3个消费者消费了3个产品

ProducerThread-3生产一件资源，当前资源池有8个
ProducerThread-5生产一件资源，当前资源池有9个
ProducerThread-4生产一件资源，当前资源池有10个

>> 注释：生产者立马又生产3个

...

>> 然后一直循环往复这个过程

（2） 阻塞队列方式

    阻塞队列的特点：

  • 当队列元素已满的时候，阻塞插入操作
  • 当队列元素为空的时候，阻塞获取操作

    不同的阻塞队列：

    ArrayBlockingQueue 与 LinkedBlockingQueue 都是支持 FIFO (先进先出)，但是 LinkedBlockingQueue 是无界的，而        ArrayBlockingQueue 是有界的。

这里我们采用无界阻塞队列来演示生产者消费者模式。

演示

还是设置生产者生产速度大于消费者消费速度（通过 sleep() 方法实现）

缓冲区 BlockQueueBufferArea.java：

public class BlockQueueBufferArea {

    BlockingQueue<Integer> mProductPoll = new LinkedBlockingQueue(10);

    public void  put() {
        try {
            System.out.println(Thread.currentThread().getName() + "产品池被放入了一个资源");
            mProductPoll.put(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    public void get() {
        try {
            System.out.println(Thread.currentThread().getName() + "产品池被取走了一个资源");
            mProductPoll.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

生产者 Producer.java：

public class Producer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Producer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Producer_" + getName());
    }

    @Override
    public void run() {
        // 不断的生产资源
        while (true) {
            sleepSomeTime();
            mBufferArea.put();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(200);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

消费者 Consumer.java：

public class Consumer extends Thread {

    private BlockQueueBufferArea mBufferArea;

    public Consumer(BlockQueueBufferArea bufferArea) {
        this.mBufferArea = bufferArea;
        setName("Consumer_" + getName());
    }

    @Override
    public void run() {
        // 不断的取出资源
        while (true) {
            sleepSomeTime();
            mBufferArea.get();
        }
    }

    private void sleepSomeTime() {
        try {
            Thread.sleep(3000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

}

测试 Test.java：

public class Test {

    public static void main(String[] args) {
        BlockQueueBufferArea bufferArea = new BlockQueueBufferArea();

        Consumer consumer1 = new Consumer(bufferArea);
        Consumer consumer2 = new Consumer(bufferArea);
        Consumer consumer3 = new Consumer(bufferArea);

        Producer producer1 = new Producer(bufferArea);
        Producer producer2 = new Producer(bufferArea);
        Producer producer3 = new Producer(bufferArea);

        consumer1.start();
        consumer2.start();
        consumer3.start();

        producer1.start();
        producer2.start();
        producer3.start();

    }

}

打印结果如下：

Producer_Thread-5产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源
Consumer_Thread-0产品池被取走了一个资源
Consumer_Thread-1产品池被取走了一个资源
Consumer_Thread-2产品池被取走了一个资源
Producer_Thread-4产品池被放入了一个资源
Producer_Thread-5产品池被放入了一个资源
Producer_Thread-3产品池被放入了一个资源

参考链接

https://blog.csdn.net/MONKEY_D_MENG/article/details/6251879

 https://www.jianshu.com/p/9f6b7fb891dd