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

Posted 张起灵-小哥

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了Java——多线程高并发系列之生产者-消费者模式相关的知识,希望对你有一定的参考价值。

文章目录:

写在前面

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集合中还有两个元素呢。

 

以上是关于Java——多线程高并发系列之生产者-消费者模式的主要内容,如果未能解决你的问题,请参考以下文章

java 多线程并发系列之 生产者消费者模式的两种实现

Java——多线程高并发系列之Condition接口中的await()signal()signAll()方法

Java——多线程高并发系列之Condition接口中的await()signal()signAll()方法

Java并发程序设计设计模式与并发之生产者-消费者模式

Java多线程之并发协作生产者消费者设计模式

Java多线程之并发协作生产者消费者设计模式