JavaEE 初阶篇-生产者与消费者模型（线程通信）-Toy模板网

这篇具有很好参考价值的文章主要介绍了JavaEE 初阶篇-生产者与消费者模型（线程通信）。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

🔥博客主页：【小扳_-CSDN博客】
❤感谢大家点赞👍收藏⭐评论✍

JavaEE 初阶篇-生产者与消费者模型（线程通信）,JavaEE 初级篇,java,开发语言,java-ee,大数据

文章目录

1.0 生产者与消费者模型概述

2.0 在生产者与消费者模型中涉及的关键概念

2.1 缓冲区

2.2 生产者

2.3 消费者

2.4 同步机制

2.5 线程间通信

3.0 实现生产者与消费者模型的完整代码

1.0 生产者与消费者模型概述

消费者与生产者模型是计算机科学中一个经典的并发编程问题，描述了多个生产者和消费者之间如何共享有限缓冲区的情况。在该模型中，生产者负责生产物品并将其放入共享的缓冲区，而消费者则负责从缓冲区取出物品进行消费。生产者与消费者之间必须进行有效的同步和协调，以避免生产者在缓冲区满时继续生产物品，或消费者在缓冲区为空时尝试消费物品，从而导致竞争条件和数据不一致的问题。

2.0 在生产者与消费者模型中涉及的关键概念

缓冲区、生产者、消费者、同步机制和线程间通信。

2.1 缓冲区

用于存储生产者生产的物品，以便消费者可以从中取出。缓冲区通常是一个有限的队列或缓冲区，可以存储一定数量的物品。

实现缓冲区可以用到数组、链表实现。目前用的是循环数组实现缓冲区的功能。可以自定义数组大小，默认大小为 10 。

循环数组的实现思路，定义三个变量：当前存储的个数 size ，头队列的索引也是取出数据的索引：head 和尾队列的索引也是放入数据的索引处：tail 。

代码如下：

public class Desk {
    private final String[] arr;
    private int size = 0;
    private int head = 0;
    private int tail = 0;

    //有参构造器
    public Desk(int size) {
        this.arr = new String[size];
    }

    //无参构造器,默认大小为10
    public Desk(){
        this.arr = new String[10];
    }

}

定义了有参和无参两个构造器。将 size 、head 、tail 初始化都为 0 。

2.2 生产者

负责向缓冲区中生产物品并放入到其中。生产者在生产物品之前通常会检查缓冲区是否已满，如果已满则需要等待直到有空间可用。

实现生产者，就是实现一个 put 方法，先判断数组中的 size 与数组大小关系，若 size >= arr.length 时，先唤醒其他全部线程，然后当前线程则进入等待状态；若 size < arr.length 时，将数据放入到索引为 tail 处的数组位置，接着 tail++ ，tail 加完之后需要判断是否越界了，如果越界了，则需要进行将 tail 重新置为 0 。再来 size++ 操作，最后在再唤醒其他线程，当前线程也就可以进行等待状态了。

代码如下：

    //放入数据
    public synchronized void put(String data) throws InterruptedException {
        String name = Thread.currentThread().getName();
        String putData = name + ",放入一个数据:" + data;
        if ( ! (size >= arr.length) ){
            arr[tail] = putData;
            tail++;
            if (tail >= arr.length){
                tail = 0;
            }
            size++;
            System.out.println(name + "成功放入数据:" + data + 
                                      ",当前数据个数为:" + size + "个");
            Thread.sleep(1000);
            this.notifyAll();
            this.wait();
        }else {
            this.notifyAll();
            this.wait();
        }
    }

为了方便观察，用到了 Thread.sleep() 方法。

2.3 消费者

负责从缓冲区中取出物品并进行消费。消费者再消费物品之前常会检查缓冲区是否为空，如果为空则需要等待直到有物品可取。

消费者的实现也是一个 take() 方法，先判断 siez == 0 ，若成立，则先唤醒其他线程，当前线程则进入等待；若不成立，则获取数组中索引位置为 head 的数据，接着 head++ 处理，紧接着判断 head >= arr.length ，若成立，将 head = 0 处理。再接着 size-- ，最后唤醒其他线程，当前线程则进入等待状态。

代码如下：

    //取出数据
    public synchronized void take() throws InterruptedException {
        String name = Thread.currentThread().getName();
        if ( !(size == 0)){
            String ret = arr[head];
            head++;
            if (head >= arr.length){
                head = 0;
            }
            size--;
            System.out.println(name + "读取到了:" + ret + 
                                      ",当前还剩数据个数为:" + size + "个");
            Thread.sleep(1000);
            this.notifyAll();
            this.wait();
        }else {
            this.notifyAll();
            this.wait();
        }
    }

同样，这里也用到了 Thread.sleep() 方法，主要是为了方便观察。

2.4 同步机制

用于实现生产者与消费者之间的同步协调。常用的同步机制包括互斥锁、条件变量、信号变量等，以确保生产者和消费者之间的操作发生竞争条件。

实现中就是用到了 synchronized() 这个关键字。这确保了在多线程环境下，同一时刻只有一个线程可以访问 put() 和 take() 方法中的关键代码块，从而保证了线程安全性。

2.5 线程间通信

生产者与消费者通常运行再不同的线程中，它们之间需要通过线程间通信机制进行协作。常用的线程间通信方式包括 wait-notify 机制等。

在循环中调用 wait() 方法，以避免虚假唤醒问题。在同步块中调用 notifyAll() 方法，以确保线程安全性。

3.0 实现生产者与消费者模型的完整代码

public class ProducerConsumer {
    public static void main(String[] args) {
        Desk desk = new Desk(1);
            //生产者线程1
            Thread thread1 = new Thread(()->{
                try {
                    while (true) {
                        desk.put("华为电脑");
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            },"生产者1");
            thread1.start();
            //生产者线程2
            Thread thread2 = new Thread(()->{
                try {
                    while (true) {
                        desk.put("小米su7");
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            },"生产者2");
            thread2.start();

            //生产者线程3
            Thread thread3 = new Thread(()->{
                try {
                    while (true) {
                        desk.put("大疆无人机");
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            },"生产者3");
            thread3.start();

            //消费者1
            Thread thread4 = new Thread(()->{
                try {
                    while (true) {
                        desk.take();
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            },"消费者1");
            thread4.start();

            //消费者2
            Thread thread5 = new Thread(()->{
                try {
                    while (true) {
                        desk.take();
                    }
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            },"消费者2");
            thread5.start();
    }
}

public class Desk {
    private final String[] arr;
    private int size = 0;
    private int head = 0;
    private int tail = 0;

    //有参构造器
    public Desk(int size) {
        this.arr = new String[size];
    }

    //无参构造器,默认大小为10
    public Desk(){
        this.arr = new String[10];
    }

    //放入数据
    public synchronized void put(String data) throws InterruptedException {
        String name = Thread.currentThread().getName();
        String putData = name + ",放入一个数据:" + data;
        if ( ! (size >= arr.length) ){
            arr[tail] = putData;
            tail++;
            if (tail >= arr.length){
                tail = 0;
            }
            size++;
            System.out.println(name + "成功放入数据:" + data + ",当前数据个数为:" + size + "个");
            Thread.sleep(1000);
            this.notifyAll();
            this.wait();
        }else {
            this.notifyAll();
            this.wait();
        }
    }
    //取出数据
    public synchronized void take() throws InterruptedException {
        String name = Thread.currentThread().getName();
        if ( !(size == 0)){
            String ret = arr[head];
            head++;
            if (head >= arr.length){
                head = 0;
            }
            size--;
            System.out.println(name + "读取到了:" + ret + ",当前还剩数据个数为:" + size + "个");
            Thread.sleep(1000);
            this.notifyAll();
            this.wait();
        }else {
            this.notifyAll();
            this.wait();
        }
    }
}

一部分的运行结果：

JavaEE 初阶篇-生产者与消费者模型（线程通信）,JavaEE 初级篇,java,开发语言,java-ee,大数据