Java——多线程高并发系列之生产者-消费者模式

Posted 2021-09-11 张起灵-小哥

文章目录：

写在前面

Demo

MyStack模拟栈

ProducerThread

ConsumerThread

Test

测试结果1

测试结果2

测试结果3

---

写在前面

在 Java 中，负责生产数据的模块是生产者，负责使用数据的模块是消费者。生产者消费者解决数据的平衡问题，即先有数据然后才能使用，没有数据时，消费者需要等待。

下面的案例中，我是这样做的：使生产者把数据存储到 List 集合中，消费者从 List 集合中取数据，使用 List 集合模拟栈。

---

Demo

MyStack模拟栈

这个类中，定义了一个list集合，以及设定了它的最大长度为3。

• 模拟入栈：向list集合中添加元素，每添加一个元素，就 this.notifyAll() 唤醒所有等待的线程：（情况一、唤醒了某个生产线程，就继续向list集合中生产新的元素；情况二：唤醒了某个消费线程，它可以消费已经生产出来的元素，如果再唤醒某个消费线程，它就必须进入等待状态了，因为list集合中已经没有元素了）。如果此时list集合中的元素数量>=MAX，那么就不能再生产了，此时就 this.wait() 进入等待状态。
• 模拟出栈：从list集合中取元素（remove方法移除元素），每移除一个元素，就 this.notifyAll() 唤醒所有等待的线程：（情况一、唤醒了某个消费线程，此时它需要进入等待状态，因为在它被唤醒之前，list集合中的元素已经被上一个消费线程消费完了；情况二、唤醒了某个生产线程，此时list集合中的元素数量<=MAX，所以可以正常生产）。如果此时list集合中的元素数量=0，那就不能再消费了，此时就 this.wait() 进入等待状态。

package com.szh.producerandconsumer;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * 该类用来模拟栈
 */
public class MyStack {

    //定义一个list集合模拟栈
    private List<String> list=new ArrayList<>();

    //栈的最大容量
    private static final int MAX=3;

    //定义一个方法模拟入栈操作
    public synchronized void push() {
        //当栈的容量达到最大（数据已满）,就进入等待状态
        while (list.size() >= MAX) {
            System.out.println(Thread.currentThread().getName() + " begin wait...");
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        String data="data--- " + new Random().nextInt();
        list.add(data);
        System.out.println( Thread.currentThread().getName() + "添加了数据: " + data);
        //当多个生产者多个消费者时,使用 notify() 可能会出现假死的情况
        //所以这里使用notifyAll()
        this.notifyAll();
    }

    //定义一个方法模拟出栈操作
    public synchronized void pop() {
        //如果栈中没有数据，就进入等待状态
        while (list.size() == 0) {
            System.out.println(Thread.currentThread().getName() + " begin wait...");
            try {
                this.wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println( Thread.currentThread().getName() + "出栈数据: " + list.remove(0) );
        this.notifyAll();
    }
}

ProducerThread

package com.szh.producerandconsumer;

/**
 *
 */
public class ProducerThread extends Thread {

    private MyStack myStack;

    public ProducerThread(MyStack myStack) {
        this.myStack=myStack;
    }

    @Override
    public void run() {
        myStack.push();
    }
}

ConsumerThread

package com.szh.producerandconsumer;

/**
 *
 */
public class ConsumerThread extends Thread {

    private MyStack myStack;

    public ConsumerThread(MyStack myStack) {
        this.myStack=myStack;
    }

    @Override
    public void run() {
        myStack.pop();
    }
}

Test

package com.szh.producerandconsumer;

/**
 *
 */
public class Test {

    public static void main(String[] args) {

        MyStack myStack = new MyStack();

        ProducerThread p1 = new ProducerThread(myStack);
        ProducerThread p2 = new ProducerThread(myStack);
        ProducerThread p3 = new ProducerThread(myStack);
        ConsumerThread c1 = new ConsumerThread(myStack);
        ConsumerThread c2 = new ConsumerThread(myStack);
        ConsumerThread c3 = new ConsumerThread(myStack);

        p1.setName("生产者 1 号");
        p2.setName("生产者 2 号");
        p3.setName("生产者 3 号");
        c1.setName("消费者 1 号");
        c2.setName("消费者 2 号");
        c3.setName("消费者 3 号");

        p2.start();
        p3.start();
        c1.start();
        c2.start();
        c3.start();
        p1.start();
    }
}

测试结果1

这种情况就是正常的，生产一个、消费一个。所以就不再解释了。

测试结果2

这种情况是说：首先生产者1号抢到了CPU执行权，它生产了一个，此时list集合的size=1；notifyAll()之后消费者3号抢到了CPU执行权，此时list集合中有元素，所以它可以正常的消费一个，此时list集合的size=0；notifyAll()之后又有两个消费者线程分别抢到了CPU执行权，而此时list集合size已经是0，不能消费了，所以这两个消费者线程都只能进入wait()等待状态；wait()释放锁对象之后，又有两个生产者线程分别抢到了CPU执行权，那么它们俩一看，list集合中size=0，这肯定要生产啊，于是就生产了两个元素，此时list集合的size=2；notifyAll()之后，因为消费者1号、2号还在等待呢，所以它们两个现在可以消费了，也就是运行结果图的最后两行。

测试结果3

这种情况是说：首先生产者1号抢到了CPU执行权，于是它可以向list集合中生产一个元素，此时list集合的size=1；notifyAll()之后，有一个消费者2号抢到了CPU执行权，因为list集合中有一个元素，所以它可以正常的消费，此时lits集合的size=0；notifyAll()之后，又有一个消费者线程抢到了CPU执行权，那list集合中没有元素了，你肯定不能再消费了，所以这个消费者线程需要wait去等待；wait()释放锁对象之后，连续两个生产者线程抢到了CPU执行权，因为list集合的size=0，所以它们俩都可以向list集合中生产元素，此时list集合的size=2；notifyAll()之后，之前等待的那两个消费者此时就可以正常的消费了，因为list集合中还有两个元素呢。