当前位置：首页 > news >正文

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

news 来源：原创 2024/5/17 11:07:22

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

文章目录

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];}//无参构造器,默认大小为10public 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);//生产者线程1Thread thread1 = new Thread(()->{try {while (true) {desk.put("华为电脑");}} catch (InterruptedException e) {throw new RuntimeException(e);}},"生产者1");thread1.start();//生产者线程2Thread thread2 = new Thread(()->{try {while (true) {desk.put("小米su7");}} catch (InterruptedException e) {throw new RuntimeException(e);}},"生产者2");thread2.start();//生产者线程3Thread thread3 = new Thread(()->{try {while (true) {desk.put("大疆无人机");}} catch (InterruptedException e) {throw new RuntimeException(e);}},"生产者3");thread3.start();//消费者1Thread thread4 = new Thread(()->{try {while (true) {desk.take();}} catch (InterruptedException e) {throw new RuntimeException(e);}},"消费者1");thread4.start();//消费者2Thread 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];}//无参构造器,默认大小为10public 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();}}
}

一部分的运行结果：