【Java EE初阶七】多线程案例(生产者消费者模型)

1. 阻塞队列

        队列是先进先出的一种数据结构;

        阻塞队列,是基于队列,做了一些扩展,适用于多线程编程中;

阻塞队列特点如下:

        1、是线程安全的

        2、具有阻塞的特性

                2.1、当队列满了时,就不能往队列里放数据,就会阻塞等待,等队列中的数据出队列后,导致队列没满时,才能放数据。

                2.2、当当队列空了时,就不能从队列里拿数据,就会阻塞等待,等有数据进入队列后,导致队列不为空时,才能拿数据。

        由于阻塞队列的用处非常大,基于阻塞队列的功能,我们就可以实现多线程案例的第三种案例~ 生产消费者模型(其实描述的就是一种多线程编程的方法),引入生产者消费者模型(尤其是后端开发),生产者往队列中写入数据,消费者从队列中消费数据;

        阻塞队列总的来说就是由于前后执行顺序的线程由于一方面的速度过快,另外一方面的速度过慢,而导致整体的执行顺序出现不流畅的画面(快的线程为了使自己的产出能被另外一方面合理的消化),该方面线程不得不阻塞,等待另外一方面将产能消化之后,继续执行线程,制造产能;

2. 生产者消费者模型

        生产者消费者模型是一种很朴素的概念,描述的是一种多线程编程的方法。

 2.1 引入生产者消费者模型的意义

        2.1.1 解耦合

        引入该模型,就可以更好的做到“解耦合”(把代码的耦合程度,从高将到低-->就称为解耦合)

        在实际开发中,会涉及到 “分布式系统” ,服务器的整个功能不是由一个服务器实现的,而是由多个服务器组成,各自实现各自的一部分功能,再通过网络通信,把这些服务器联系起来,最终完成整个服务器的功能。典型分布式例子通过下图来进行简单的讲解:

        如上图所示,在该模型中入口服务器A与B、入口服务器A与C服务器的联系是密切相关的,请求要经过入口服务器A,才能传达给B、C服务器,即B、C服务器拿到想要的数据,再返回给入口服务器A,通过入口服务器A,再把响应传给客户端。

        但是如果请求突然骤升,这时超过入口服务器A接收请求的峰值,这时入口服务器A就挂了,入口服务器A挂了后,B、C服务器拿不到请求,也会挂掉,这就体现了入口服务器A和B、C服务器的耦合性比较高。

        当然如果B或C挂了的话,A大概率也会挂;

        当我们在入口服务器A和B、C服务器之间引入阻塞队列时,如下图所示:

        如上图所示,如果入口服务器A挂了,但是阻塞队列中还有请求的数据,至少不会因入口服务器挂A了,B、C服务器也挂了

        故此,入口服务器A和B、C、D服务器的耦合性也就降低了。

        上述描述的阻塞队列,并非是简单的数据结构,而是基于这个数据结构实现的服务器程序,且被部署到单独的主机上来;

2.1.2 削峰填谷

       如上图所示:当客户端这边的请求突然骤增时,入口服务器A一般来说是比较能抗压的,但是也是有极限的,这时我们引入阻塞队列,可以把这些请求数据都放进阻塞队列中,形成一个缓冲区,如此一来,即使外面的请求达到了峰值,也是由阻塞队列来承担,这样就形成了削峰填谷的效果。

        关于阻塞队列和消息队列的区别:

        阻塞队列:数据结构

        消息队列:基于阻塞队列实现服务器程序

3. 手敲代码模拟实现阻塞队列

3.1 了解阻塞队列

      java标准库提供了现成的阻塞队列这一数据结构,如下图所示:

        

        阻塞队列是基于队列扩展而来的,且在阻塞队列中,put是在具备阻塞功能的入队列操作,take方法是带阻塞功能的出队列操作,阻塞队列没有提供带有阻塞功能获取首元素的方法;

        java自带的阻塞队列的代码实现如下:

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;public class Main {public static void main(String[] args) throws InterruptedException {BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10);blockingQueue.put("smallye");String s1 = blockingQueue.take();System.out.println("第一个打印:s1 = " + s1);s1 = blockingQueue.take();System.out.println("第二个打印:s1 = " + s1);}
}

        结果如下:

        问题分析:

        主要是线程卡住了,当进行第二次出队列时,由于当前阻塞队列是空的,所以要等进行阻塞等待,当有元素入队列时,我们才能进行出队列操作。

3.2 实现阻塞队列

        我们尝试实现一个阻塞队列,要求达到与标准库中的队列有着类似的效果;

        步骤如下:

        1、先实现普通队列

        2、再加上线程安全

        3、再加上阻塞功能

3.2.1 先实现普通队列

        代码如下所示:

// 为了简单, 不写作泛型的形式. 考虑存储的元素就是单纯的 String
class MyBlockingQueue {private String elems[] = null;private int head = 0;//记录头结点private  int tail = 0;//记录尾结点private int size = 0;//队列元素个数//构造方法,定义队列的容量大小public MyBlockingQueue(int capacity) {this.elems = new String[capacity];}//入队列public void put(String elem) {//判断容量满了没,满了就不能入队列,要阻塞等待if(size >= this.elems.length) {//阻塞等待,先不写,先实现普通功能的队列return;}//入队列elems[tail] = elem;tail++;//因为是循环队列,所以要判断尾巴有没有超过容量大小下标,超过了就要从0开始了if(tail > elems.length) {tail = 0;}//队列元素要++size++;}//出队列public String take() {String elem = null;//要判断队列是不是空的,空就不能出队列了,要阻塞等待if(size == 0) {//阻塞等待,因为是先实现普通队列的功能,所以后面再补充return null;}elem = elems[head];head++;//因为是循环队列,所以要判断头结点有没有超过容量大小下标,超过了就要0开始了if(head >= elems.length) {head = 0;}//出队列后,队列元素要--size--;return elem;}
}

        测试代码及结果如下:

public class Main {public static void main(String[] args) {MyBlockingQueue blockingQueue = new MyBlockingQueue(10);blockingQueue.put("smallye");String s1 = blockingQueue.take();System.out.println("第一个打印:s1 = " + s1);}
}

3.2.2 再加上线程安全

        对于不线程安全的代码我们要进行加锁操作,首先针对的就是写操作,该部分的代码块肯定是要加锁的,因为多线程同时执行写操作,会导致线程不安全,如下图所示:

        下面,我们讨论一下这两个代码要不要加锁,以take为例,如下图所示:

        当前代码里面的队列为空,但是依旧执行出队列的逻辑,所以我们判断条件也应该加锁;

        以put为例,如下图所示:

        当前代码里面的队列已经满了,但是依旧执行入队列的逻辑;

        修改后代码如下:

class MyBlockingQueue {Object locker = new Object();private String elems[] = null;private int head = 0;//记录头结点private  int tail = 0;//记录尾结点private int size = 0;//队列元素个数//构造方法,定义队列的容量大小public MyBlockingQueue(int capacity) {this.elems = new String[capacity];}//入队列public void put(String elem) {synchronized (locker) {//判断容量满了没,满了就不能入队列,要阻塞等待if(size >= this.elems.length) {//阻塞等待,先不写,先实现普通功能的队列return;}//因为这些都是写操作,也有读操作,多线程并发执行时,写操作是线程不安全的,要把这些打包成一个原子,加锁synchronized (locker) {//入队列elems[tail] = elem;tail++;//因为是循环队列,所以要判断尾巴有没有超过容量大小下标,超过了就要从0开始了if(tail > elems.length) {tail = 0;}//队列元素要++size++;}}}//出队列public String take() {String elem = null;//因为这些都是写操作,也有读操作,多线程并发执行时,写操作是线程不安全的,要把这些打包成一个原子,加锁synchronized (locker) {//要判断队列是不是空的,空就不能出队列了,要阻塞等待if(size == 0) {//阻塞等待,因为是先实现普通队列的功能,所以后面再补充return null;}elem = elems[head];head++;//因为是循环队列,所以要判断头结点有没有超过容量大小下标,超过了就要0开始了if(head >= elems.length) {head = 0;}//出队列后,队列元素要--size--;return elem;}}
}

3.2.3 再加上阻塞功能

        我们给put要加上阻塞功能,就要在这条件判断上加上wait,我们用locker的对象给他wait,而且wait必须要在synchronized内使用,这里的locker正好能对应上;当这个队列满时,就阻塞等待,等take方法拿走一个数据时,才被唤醒,加上阻塞功能后的代码如下:

class MyBlockingQueue {Object locker = new Object();private String elems[] = null;private int head = 0;//记录头结点private  int tail = 0;//记录尾结点private int size = 0;//队列元素个数//构造方法,定义队列的容量大小public MyBlockingQueue(int capacity) {this.elems = new String[capacity];}//入队列public void put(String elem) throws InterruptedException {synchronized (locker) {//判断容量满了没,满了就不能入队列,要阻塞等待if (size >= this.elems.length) {//阻塞等待,先不写,先实现普通功能的队列synchronized (locker) {locker.wait();}}//因为这些都是写操作,也有读操作,多线程并发执行时,写操作是线程不安全的,要把这些打包成一个原子,加锁synchronized (locker) {//入队列elems[tail] = elem;tail++;//因为是循环队列,所以要判断尾巴有没有超过容量大小下标,超过了就要从0开始了if(tail > elems.length) {tail = 0;}//队列元素要++size++;locker.notify();}}}//出队列public String take() throws InterruptedException {String elem = null;//因为这些都是写操作,也有读操作,多线程并发执行时,写操作是线程不安全的,要把这些打包成一个原子,加锁synchronized (locker) {//要判断队列是不是空的,空就不能出队列了,要阻塞等待if (size == 0) {//阻塞等待,因为是先实现普通队列的功能,所以后面再补充synchronized (locker) {locker.wait();}}elem = elems[head];head++;//因为是循环队列,所以要判断头结点有没有超过容量大小下标,超过了就要0开始了if(head >= elems.length) {head = 0;}//出队列后,队列元素要--size--;locker.notify();return elem;}}
}

        当我们进行阻塞wait时,一定要在适当的条件下notify,如下图所示:

        代码讲解:

        当put时,队列满了时就要阻塞等待,等take队列后,就会唤醒put操作,接着put就能入队列了;

        如果队列不满也不空时,每次put和take都会notify一次,其实不会有影响,因为就算没有其他线程在等待,唤醒也没有事,不会对程序造成啥影响。而且我们的代码,一定是要么满,要么空,要么不满也不空。

        但是,如果有两个线程同时put,现在队列是满的,A线程先阻塞,B线程也阻塞,这时有第三个线程take一次,把A线程的wait唤醒了,等A执行到下面的notify,A线程里put的notify就会唤醒B线程里的wait,但是因为A线程put了,和第三个线程的take一取一放抵消了,此时队列还是满的;因为A线程里的put把B线程里的wait唤醒了,这时已经是满了的队列还往里放元素,就造成了线程安全问题。

        解决方案:把条件判断if换成while循环语句,不是只判断一次,当有其他线程把wait唤醒后,还要再判断一次这个队列是不是满的或者是空的,如果不是满的或者不是空的,才释放这个wait,不然就要继续wait,如此该问题也就解决了。

最终代码:
 

class MyBlockingQueue {Object locker = new Object();private String elems[] = null;private int head = 0;//记录头结点private  int tail = 0;//记录尾结点private int size = 0;//队列元素个数//构造方法,定义队列的容量大小public MyBlockingQueue(int capacity) {this.elems = new String[capacity];}//入队列public void put(String elem) throws InterruptedException {synchronized (locker) {//判断容量满了没,满了就不能入队列,要阻塞等待while (size >= this.elems.length) {//阻塞等待,先不写,先实现普通功能的队列synchronized (locker) {locker.wait();}}//因为这些都是写操作,也有读操作,多线程并发执行时,写操作是线程不安全的,要把这些打包成一个原子,加锁synchronized (locker) {//入队列elems[tail] = elem;tail++;//因为是循环队列,所以要判断尾巴有没有超过容量大小下标,超过了就要从0开始了if(tail > elems.length) {tail = 0;}//队列元素要++size++;locker.notify();}}}//出队列public String take() throws InterruptedException {String elem = null;//因为这些都是写操作,也有读操作,多线程并发执行时,写操作是线程不安全的,要把这些打包成一个原子,加锁synchronized (locker) {//要判断队列是不是空的,空就不能出队列了,要阻塞等待while (size == 0) {//阻塞等待,因为是先实现普通队列的功能,所以后面再补充synchronized (locker) {locker.wait();}}elem = elems[head];head++;//因为是循环队列,所以要判断头结点有没有超过容量大小下标,超过了就要0开始了if(head >= elems.length) {head = 0;}//出队列后,队列元素要--size--;locker.notify();return elem;}}
}

        在实际开发中,生产者消费者模型,往往是多个生产者,多个消费者;这里的生产者和消费者往往不仅仅是一个线程,也可能是一个独立的服务器,甚至是一组服务器程序。生产者消费者模型,最核心的部分还是阻塞队列,可以使用synchronized和wait / notify 达到线程安全与阻塞。

3.3 实现生产者消费者模型

        代码如下:

package thread;// 为了简单, 不写作泛型的形式. 考虑存储的元素就是单纯的 String
class MyBlockingQueue {private String[] elems = null;private int head = 0;private int tail = 0;private int size = 0;// 准备锁对象, 如果使用 this 也可以.private Object locker = new Object();public MyBlockingQueue(int capacity) {elems = new String[capacity];}public void put(String elem) throws InterruptedException {// 锁加到这里和加到方法上本质一样的. 加到方法上是给 this 加锁. 此处是给 locker 加锁.synchronized (locker) {while (size >= elems.length) {// 队列满了.// 后续需要让这个代码能够阻塞.locker.wait();}// 新的元素要放到 tail 指向的位置上elems[tail] = elem;tail++;if (tail >= elems.length) {tail = 0;}size++;// 入队列成功之后唤醒locker.notify();}}public String take() throws InterruptedException {String elem = null;synchronized (locker) {while (size == 0) {// 队列空了.// 后续也需要让这个代码阻塞locker.wait();}// 取出 head 位置的元素并返回elem = elems[head];head++;if (head >= elems.length) {head = 0;}// 这个代码不要遗漏.size--;// 元素出队列成功之后, 加上唤醒locker.notify();}return elem;}
}public class ThreadDemo28 {public static void main(String[] args) throws InterruptedException {MyBlockingQueue queue = new MyBlockingQueue(1000);// 生产者Thread t1 = new Thread(() -> {int n = 1;while (true) {try {queue.put(n + "");System.out.println("生产元素 " + n);n++;} catch (InterruptedException e) {throw new RuntimeException(e);}}});// 消费者Thread t2 = new Thread(() -> {while (true) {try {String n = queue.take();System.out.println("消费元素 " + n);Thread.sleep(500);} catch (InterruptedException e) {throw new RuntimeException(e);}}});t1.start();t2.start();}
}

        结果如下:

        如图所示,生产者消费者模型大抵是生产一个,消费一个,主要是生产之后消费者再消费;

ps:关于阻塞队列和生产着消费者模型的内容就到这里了,如果对你有帮助的话就请一键三连哦!!!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.hqwc.cn/news/327536.html

如若内容造成侵权/违法违规/事实不符,请联系编程知识网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

druid Communications link failure报错处理

现象 日志报错&#xff1a;com.mysql.jdbc.exceptions.jdbc4.CommunicationsException: Communications link failure 原因 从数据库连接池拿到了已经关闭的连接&#xff0c;导致报错。druid有定时任务进行空闲连接的检测和回收&#xff0c;当连接时长超过mysql的连接超时时间…

YOLOv8改进 | 2023主干篇 | FasterNeT跑起来的主干网络( 提高FPS和检测效率)

一、本文介绍 本文给大家带来的改进机制是FasterNet网络,将其用来替换我们的特征提取网络,其旨在提高计算速度而不牺牲准确性,特别是在视觉任务中。它通过一种称为部分卷积(PConv)的新技术来减少冗余计算和内存访问。这种方法使得FasterNet在多种设备上运行速度比其他网络…

【Flink精讲】Flink数据延迟处理

面试题&#xff1a;Flink数据延迟怎么处理&#xff1f; 将迟到数据直接丢弃【默认方案】将迟到数据收集起来另外处理&#xff08;旁路输出&#xff09;重新激活已经关闭的窗口并重新计算以修正结果&#xff08;Lateness&#xff09; Flink数据延迟处理方案 用一个案例说明三…

C# Attribute特性实战(1):Swtich判断优化

文章目录 前言简单Switch问题无参Swtich方法声明Swtich Attribute声明带有Swtich特性方法主方法结果 有参Switch修改代码修改运行过程运行结果 总结 前言 在经过前面两章内容的讲解&#xff0c;我们已经简单了解了如何使用特性和反射。我们这里解决一个简单的案例 C#高级语法 …

摄像头视频录制程序使用教程(Win10)

摄像头视频录制程序-Win10 &#x1f957;介绍&#x1f35b;使用说明&#x1f6a9;config.json 说明&#x1f6a9;启动&#x1f6a9;关闭&#x1f6a9;什么时候开始录制&#xff1f;&#x1f6a9;什么时候触发录制&#xff1f;&#x1f6a9;调参 &#x1f957;介绍 检测画面变化…

前端Web系统架构设计

文章目录 1.目录结构定义2. 路由封装2.1 API路由定义2.2 组件路由定义 3. Axios请求开发4. 环境变量封装5. storage模块封装(sessionStorage, localStorage)6. 公共函数封装(日期,金额,权限..)7. 通用交互定义(删除二次确认,类别,面包屑...)8. 接口全貌概览 1.目录结构定义 2. …

30、共空间模式CSP与白化矩阵

CSP算法和PCA降维都涉及到了白化&#xff0c;那白化的目的和作用到底是啥呢&#xff1f; 矩阵白化目的&#xff1a; 对于任意一个矩阵X&#xff0c;对其求协方差&#xff0c;得到的协方差矩阵cov(X)并不一定是一个单位阵。 下面介绍几个线代矩阵的几个概念&#xff1a; 1、…

计算机网络——OSI参考模型

1. OSI模型的基本概念 1.1 定义 OSI&#xff08;开放式系统互联&#xff09;模型是一个用于理解和标准化电信系统或计算机网络功能的概念性的框架&#xff0c;用于描述和标准化不同计算机系统或网络设备间通信的功能。 1.2 OSI模型的性质 这个模型由国际标准化组织&#xff08…

CMake入门教程【核心篇】定义C++宏定义(add_compile_definitions)

😈「CSDN主页」:传送门 😈「Bilibil首页」:传送门 😈「本文的内容」:CMake入门教程 😈「动动你的小手」:点赞👍收藏⭐️评论📝 文章目录 1.基本用法2.定义单个宏3.定义多个宏4.条件定义宏5.使用预定义变量6.使用 generator 表达式

7款实用的SQLite数据库可视化管理工具

前言 俗话说得好“工欲善其事&#xff0c;必先利其器”&#xff0c;合理的选择和使用可视化的管理工具可以降低技术入门和使用门槛。今天推荐7款实用的SQLite数据库可视化管理工具(GUI)&#xff0c;帮助大家更好的管理SQLite数据库。 什么是SQLite&#xff1f; SQLite是一个…

每日一题——LeetCode1089.复写0

方法一 splice&#xff1a; 通过数组的slice方法&#xff0c;碰到 0就在后面加一个0&#xff0c;最后截取原数组的长度&#xff0c;舍弃后面部分。 但这样做是违反了题目的要求&#xff0c;不要在超过该数组长度的位置写入元素。 var duplicateZeros function(arr) {var le…

超简单的详细教程:如何为一个GitHub开源项目做出贡献!

仓库&#xff1a;Ai-trainee/GPT-Prompts-Hub 让我们通过一个具体的例子&#xff0c;详细了解如何从克隆一个GitHub仓库开始&#xff0c;一步步地贡献到一个项目。以下是详细步骤&#xff0c;包括所需的代码和说明&#xff1a; 首先我们Fork想要贡献的项目&#xff0c;然后请看…