初始JavaEE篇——多线程(5):生产者-消费者模型、阻塞队列

在这里插入图片描述

找往期文章包括但不限于本期文章中不懂的知识点:

个人主页:我要学编程程(ಥ_ಥ)-CSDN博客

所属专栏:JavaEE

文章目录

  • 阻塞队列
  • 生产者—消费者模型
    • 生产者—消费者模型的优势:
    • 生产者—消费者模型的劣势:
  • Java标准库中的阻塞队列:
  • 模拟实现阻塞队列:

前面我们学习多线程的经典案例之一:饿汉模式与懒汉模式。两者的区别是创建类的实例的时机不同,前者是迫不及待的去创建类的实例,而后者是迫不得已去创建类的实例。这样就导致了前者在 get 方法中只有"读"操作,不会造成线程安全问题,而后者会出现线程安全问题。最后经过我们的不断深入探索并解决了其中的问题。首先是进行加锁操作,避免了修改操作原子性,其次是加了 if 判断语句,避免了不必要的加锁,从而导致的性能下降,最后,针对指令重排序的问题,在引用变量中加上了 volatile 关键字。如果想更加深入了解,可以点击下面的链接:饿汉模式、懒汉模式、指令重排序等

阻塞队列

现在我们来学习另外一个经典的案例:阻塞队列。
阻塞队列是属于队列的一种,但是和普通的队列相比,它具有以下的特性:
1、它具备线程安全的特点,即使在多线程的环境下,也是可以正常使用的。
2、阻塞特性:1)当队列为空时,如果再去队列中取元素的话,会发生阻塞,直至队列不为
空;2)当队列满了时,如果再去队列中插入元素的话,也会发生阻塞,直至队列不为满。

生产者—消费者模型

阻塞队列的主要应用场景是:“生产者—消费者模型”。那什么是生产者,什么又是消费者呢?简单理解就是,生产者与消费者之间是通过某种资源进行来进行交互的。生产者,就是生产这个资源的,而消费者,就是消耗这个资源的。
例如,在我们日常中,最常见的就是包饺子,包饺子需要擀面皮的人、包饺子的人、放面皮的布。(肉已经被绞肉机给搞好了) 这里就是一个经典的"生产者一消费者模型"。
生产者:擀面皮的人、消费者:包饺子的人、阻塞队列:放面皮的布。这里生产者与消费者进行交互的就是"面皮"这种资源。
我们在日常生活中,有两种包饺子的方式:
1、家里面几个人全部一起参与包饺子的全过程。即每个人都需要 擀面皮、包饺子。而擀面杖只有一个,那么当一个人在进行擀面皮时,另外几个人都得阻塞等待,当这个人把面皮给擀完之后,才会释放,这样下一个人才有机会去使用。这个擀面杖就是我们前面学习的锁。
上面的方式,我们会发现一个很大的缺陷:当其中一个人在生产面皮时,其余的人得阻塞等待,也就是有空闲时间。这对于计算机来说,简直就是浪费,因此下面这种方式更为合理。
2、一个人专门擀面皮,另外的人负责包饺子,这样就不会导致生产者或者消费者会出现空闲的情况(生产速度与消费速度是一致的)。

生产者—消费者模型的优势:

1、解耦合:
生产者与消费者避免了直接交互,而是通过阻塞队列来进行交互,这样有利于代码的解耦合,使得后期的维护成本变低。
2、削峰填谷:
当 生产者—消费者模型 应用于两个服务器时,就可以达到削峰填谷的效果。

在这里插入图片描述

因此为了避免上述的情况,我们需要对用一个阻塞队列来处理这种"突然的大量请求的情况"。例如,学校选课的时候,通常就会出现这样的情况。
解决方法:使用一个阻塞队列来充当缓冲的作用。当A服务器突然接收到有大量的请求时,这个阻塞队列便会接收这些请求,但是还是以平常的速度给到B服务器,这样B服务器还是会正常运行,这就是 “削峰”。当这个峰值过去之后,就是平常少量的请求,而阻塞队列这时候就会来处理在高峰期接收的请求,这样B服务器还是以平常的速度在处理请求,这就是"填谷"。阻塞队列通过降低高峰期的发送请求,而是在低谷期来处理,这样B服务器就是以一个平均的速度在处理请求。
注意:
1、阻塞队列通常可以接收并存放很多的请求。
2、高峰期比较短,所以阻塞队列一般不会出现满的情况。

生产者—消费者模型的劣势:

1、引入阻塞队列之后,整体的结构相交以前更为复杂了,同时也需要更多的机器进行部署,使生产环境的结构更复杂,同时管理起来也更为麻烦了。
2、效率也有一定的影响。之前是A服务器和B服务器直接进行交互,现在多了个阻塞队列,消息的传递所消耗的时间也变多了。

Java标准库中的阻塞队列:

Java的标准库中提供的阻塞队列是:BlockingQueue。

在这里插入图片描述

我们在日常的开发中,主要就是使用:ArrayBlockingQueue、LinkedBlockingQueue、PriorityBlockingQueue。下面是使用的示例:

在这里插入图片描述

public class Test {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(100);
        // 在多线程中,队列的插入方法要用put。因为其带有阻塞功能,且线程安全
        queue.put(1);
        queue.put(2);
        queue.put(3);
        queue.put(4);
        // 同样多线程中的删除也要用take
        int n = queue.size();
        for (int i = 0; i < n; i++) {
            System.out.print(queue.take()+" "); // 1 2 3 4
        }
    }
}

我们现在可以去看一下阻塞功能。

public class Test {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(1);
        // 在队列为空的情况下,尝试去取元素
        System.out.println(queue.take()); // 由于是单线程,因此会一直阻塞,即死等。

        queue.put(1);
        // 在队列为满足,尝试去插入新元素
        queue.put(2);
    }
}

同样下面去尝试插入新元素时,也是会发生阻塞等待的,也是死等的情况。

public class Test {
    public static void main(String[] args) {
        int n = 1;
        System.out.println("队列的总容量:"+n);
        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(n);
        System.out.println("队列此时的容量:"+queue.size());

        Thread t1 = new Thread(()-> {
            try {
                System.out.println("阻塞队列为空,尝试取出元素,等待其他的线程插入元素");
                queue.take();
                System.out.println("成功取出元素~");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        Thread t2 = new Thread(()->{
            try {
                for (int i = 0; i < 3; i++) {
                    Thread.sleep(1000); // 确保t1线程先执行到take方法,并放慢让我们观察
                    System.out.println("正在尝试插入第" + (i + 1) + "个元素");
                    queue.put(i);
                    System.out.println("插入成功~");
                }
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        t1.start();
        t2.start();
    }
}

运行结果:
在这里插入图片描述

模拟实现阻塞队列:

要求:我们主要是实现队列的 put方法、take方法、size方法即可。

思路:put、take方法都需要保证线程安全和阻塞的特性。
线程安全,我们直接对代码进行加锁操作即可;
阻塞特性:当队列为满时,要阻塞到其他线程使用掉其中的对头元素,即得等待其他线程调用take方法来唤醒当前因队列满而造成的阻塞,这也就需要用到我们前面学习的wait 和 notify 方法。

阻塞队列代码:

public class MyBlockingQueue {
    // 基于数组去模拟实现
    private static int[] array = null;
    private static int usedSize = 0; // 元素个数
    private int head = 0; // 头指针
    private int tail = 0; // 尾指针

    public MyBlockingQueue() {
        array = new int[10];
    }

    public MyBlockingQueue(int capacity) {
        if (capacity < 0) {
            throw new RuntimeException();
        } else if (capacity >= Integer.MAX_VALUE) {
            array = new int[Integer.MAX_VALUE];
        } else {
            array = new int[capacity];
        }
    }

    // put方法
    public void put(int x) throws InterruptedException {
        synchronized (this) { // 保证线程安全
            while (usedSize >= array.length) { // 满足阻塞队列的特性
                // 阻塞等待
                this.wait(); // 等待其他线程取出元素,使队列不为满
            }
            array[tail] = x;
            tail = (tail+1) % array.length; // 这里是采用循环队列的方式
            usedSize++;
            this.notify(); // 唤醒空的阻塞
        }
    }

    public int take() throws InterruptedException {
        synchronized (this) { // 保证线程安全
            while (usedSize <= 0) { // 满足阻塞队列的特性
                // 阻塞等待
                this.wait(); // 等待其他线程插入元素,使队列不为空
            }
            int ans = array[head];
            head = (head+1) % array.length;
            usedSize--;
            this.notify(); // 唤醒满的阻塞
            return ans;
        }
    }

    public int size() {
        return usedSize;
    }
}

测试代码:

public class Test {
    public static void main(String[] args) throws InterruptedException {
        int n = 1;
        MyBlockingQueue queue = new MyBlockingQueue(n);
        System.out.println("队列的总容量为:"+n);
        System.out.println("队列当前的容量为:"+queue.size());
        Thread t1 = new Thread(()->{
            try {
                Thread.sleep(500);
                System.out.println("队列为空,阻塞等待别的线程插入数据");
                queue.take();
                System.out.println("取出元素~");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });

        Thread t2 = new Thread(()->{
            for (int i = 0; i < 3; i++) {
                try {
                    Thread.sleep(1000);
                    System.out.println("正在尝试插入第"+(i+1)+"个元素");
                    queue.put(i);
                    System.out.println("插入成功~");
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });

        t1.start();
        t2.start();
    }
}

运行结果:

在这里插入图片描述

注意:
1、我们使用加锁操作,是为了避免出现下面这种情况:一个线程在修改,另一个线程在读取,把数据修改之后,可能会造成另一个线程执行有误,因此我们得对代码进行加锁操作,是同一时刻只能有一个线程去进行修改操作(读取操作是不会影响数据的),因此对于修改操作的代码,都得处于 synchronized 代码块中,而上述 put、take 方法的大部分代码都是修改操作,因此我们就将整个代码逻辑都置于 synchronized 代码块中了。

2、put、take 方法中之所以将判断阻塞的条件放到 while 循环中,是因为可能会出现下面这样的情况:有三个线程都是处于put方法的阻塞状态,而这时新来了一个执行take方法的线程,其会随机唤醒三个线程中的一个,当三个线程中,某个线程执行完 notify 方法之后,也会随机唤醒剩下的两个线程,但是此时这个唤醒操作不符合要求,因为我们是希望将处于take方法的阻塞线程所唤醒,因此这个是错误唤醒,所以我们要用 while 循环去再次线程判断到底是不是因为正常唤醒而被唤醒的。

好啦!本期 初始JavaEE篇——多线程(4):生产者-消费者模型、阻塞队列 的学习之旅 就到此结束啦!

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

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

Redis常见面试题(二)

Redis性能优化 Redis性能测试 阿里Redis性能优化 使用批量操作减少网络传输 Redis命令执行步骤&#xff1a;1、发送命令&#xff1b;2、命令排队&#xff1b;3、命令执行&#xff1b;4、返回结果。其中 1 与 4 消耗时间 --> Round Trip Time&#xff08;RTT&#xff0c;…

Scala学习记录,List

List是一个不可变&#xff08;immutable&#xff09;的序列。特点&#xff1a;数据是有序的 前面学习的Set&#xff0c;Map数据是无序的&#xff1b;Array是有序的&#xff0c;Array数组物理空间上是连续的 List可变不可变&#xff1a; list中不可变的列表是不能修改的 list…

【从零开始的LeetCode-算法】1456. 定长子串中元音的最大数目

给你字符串 s 和整数 k 。 请返回字符串 s 中长度为 k 的单个子字符串中可能包含的最大元音字母数。 英文中的 元音字母 为&#xff08;a, e, i, o, u&#xff09;。 示例 1&#xff1a; 输入&#xff1a;s "abciiidef", k 3 输出&#xff1a;3 解释&#xff1a…

0-基于图的组合优化算法学习(NeurIPS 2017)(未完)

文章目录 Abstract1 Introduction2 图上的贪婪算法的通用表述3 表示:图嵌入3.1 Structure2Vec3.2 参数化 Q ^ ( h ( S ) , v ; Θ ) \widehat{Q}(h(S), v; \Theta) Q ​(h(S),v;Θ)4 Training: Q-learningAbstract 为NP-hard组合优化问题设计好的启发式或近似算法通常需要大…

RK3568平台开发系列讲解(设备树篇)设备树(device Tree)的由来

🚀返回专栏总目录 文章目录 一、设备树的由来二、设备树的组成沉淀、分享、成长,让自己和他人都能有所收获!😄 一、设备树的由来 首先不得不提到Linus的一封重要的邮件:(硬件解耦)(可以复用的代码) Gaah. Guys, this whole ARM thing is a f*cking pain in the ass.…

基于C++深度优先遍历迷宫

c实现的深度优先遍历迷宫&#xff0c;迷宫大小为20*20&#xff0c;代码简练清楚&#xff0c;内涵关键注释。代码与网上都不一样。 深度优先遍历迷宫&#xff0c;核心思想是借助一个栈&#xff0c;站在一个节点上时&#xff0c;将它附近可以走的节点存在栈中&#xff0c;再按顺…

QML项目实战:自定义CheckBox

目录 一.添加模块 import QtQuick.Controls 1.2 import QtQuick.Controls.Styles 1.4 import QtGraphicalEffects 1.15 二.自定义CheckBox 1.CheckBox设置 2.勾选框设置 3.标签部分 4. 状态变化处理 5.文本设置 三.效果 1.当enabled为true 2.当enabled为true 3.当…

天命人开店日记之门店经营调研(下)

在调研前拟定了一些想要去了解的信息&#xff0c;包括&#xff1a;月销量、净利润、用户购买的主要担忧、与电商平台的竞争差异等关键内容&#xff0c;然而当自己去实地考察线下门店时&#xff0c;确发现实际情况与自己的预期相差非常大。大大出乎预料的包括三方面&#xff1a;…

【昇腾】Linux系统常见命令

文章目录 查看操作系统信息查看EulerOS内核版本 查看root下的内容查看/etc目录下的内容sh: yum: command not foundValueError: zero-size array to reduction operation minimum which has no identityAttributeError: torch_npu._C._NPUDeviceProperties object has no attri…

立体视觉的核心技术:视差计算与图像校正详解

立体视觉的核心技术&#xff1a;视差计算与图像校正详解 在立体视觉中&#xff0c;通过双目相机&#xff08;即左右两台相机&#xff09;的不同视角捕获的图像&#xff0c;结合几何关系&#xff0c;我们可以推算出场景中物体的深度。本文将深入讲解如何基于视差&#xff08;di…

11.6-11.7重大专业能力测试(换皮c++考试)全攻略(两天速通版)

relations的vector存储的就是Relation类型的数据&#xff0c;并不是指针&#xff0c;所以relations[i]访问Relation的成员就是直接用.&#xff0c; 但是joins的JoinSql里面存的是指针&#xff0c;并不是实际的数据&#xff0c;所以应当用->来访问其中的成员 结构体当中的Sq…

Go语言结构体、方法与接口

文章目录 一、结构体构造函数Go语言中的构造函数语法 二、结构体方法和接收器无参数和返回值值类型接收者指针类型接收者方法继承方法重写 三、结构体比较结构体比较要求结构体比较符号 四、接口声明接口定义接口特点接口格式标准格式接口的实现&#xff1a;空接口error接口 五…

用Puppeteer点击与数据爬取:实现动态网页交互

用Puppeteer与代理IP抓取51job招聘信息&#xff1a;动态网页交互与数据分析 引言 在数据采集领域&#xff0c;传统的静态网页爬虫方式难以应对动态加载的网页内容。动态网页通常依赖JavaScript加载数据&#xff0c;用户需要与页面交互才能触发内容显示。因此&#xff0c;我们…

Sophos | 网络安全

在 SophosLabs 和 SophosAI 的威胁情报、人工智能和机器学习的支持下&#xff0c;Sophos 提供广泛的高级产品和服务组合&#xff0c;以保护用户、网络和端点免受勒索软件、恶意软件、漏洞利用、网络钓鱼和各种其他网络攻击。Sophos 提供单一的集成式基于云的管理控制台 Sophos …

盘点RPA在政务领域落地应用

数字政府是数字经济的中坚力量&#xff0c;以强有力的“抓手”带动着各行各业的数字化转型以及新技术的应用与普及。近两年&#xff0c;以RPA为代表的数字技术在政务实践中的表现受到了很高的关注&#xff0c;RPA数字员工在各地相关政务部门悄然上岗&#xff0c;有效助力政府信…

mysql5.7安装SSL报错解决(2),总结

Caused by: java.io.EOFException: SSL peer shut down incorrectly 在java里面连接mysql5.7.17数据库&#xff0c;报以上错误&#xff0c; 将数据库升级到mysql5.7.44就可以了。 这两天处理java连接mysql的问题&#xff0c;报了各种错误&#xff0c;总结一下就是openssl和mysq…

前端基础-html-注册界面

&#xff08;200粉啦&#xff0c;感谢大家的关注~ 一起加油吧~&#xff09; 浅浅分享下作业&#xff0c;大佬轻喷~ 网页最终效果&#xff1a; 详细代码&#xff1a; ​ <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8"…

论文 | Teaching Algorithmic Reasoning via In-context Learning

这篇论文《通过上下文学习教授算法推理》探讨了如何通过上下文学习&#xff08;In-context Learning, ICL&#xff09;有效训练大型语言模型&#xff08;LLMs&#xff09;以进行算法推理。以下是从多个角度对这项工作的详细解读&#xff1a; 1. 问题陈述与研究动机 算法推理的…

Json 类型与多值索引 — OceanBase 4.3.2 AP 功能体验

本文来自 2024年OceanBase技术征文大赛——“让技术被看见 | OceanBase 布道师计划”的用户征文。也欢迎更多的技术爱好者参与征文&#xff0c;赢取万元大奖。和我们一起&#xff0c;用文字让代码跳动起来&#xff01; 参与2024年OceanBase技术征文大赛>> MySQL在5.7.8…

FPAGA学习~问题记录

1.Error: concurrent assignmentto a non-netstart is not permitted&#xff08;错误&#xff1a;不允许并发分配到非网络‘start’&#xff09; 原因&#xff1a;wire 或reg 类型不匹配引起的&#xff0c;assign与wrie搭配使用&#xff0c;而reg一般在always、initial语句块中…