【Java多线程案例】实现阻塞队列

1. 阻塞队列简介

1.1 阻塞队列概念

阻塞队列：是一种特殊的队列，具有队列"先进先出"的特性，同时相较于普通队列，阻塞队列是线程安全的，并且带有阻塞功能，表现形式如下：

  • 当队列满时，继续入队列就会阻塞，直到有其他线程从队列中取出元素
  • 当队列空时，继续出队列就会阻塞，直到有其他线程往队列中插入元素

基于阻塞队列我们可以实现生产者消费者模型，这在后端开发场景中是相当重要的！

1.2 生产者-消费者模型优势

基于阻塞队列实现的 生产者消费者模型 具有以下两大优势：

  1. 解耦合：

image.png
以搜狗搜索的服务器举例，用户输入搜索关键字 **美容，**客户端的请求到达搜狗的"入口服务器"时，会将请求转发到 广告服务器 和 大搜索服务器，此时广告服务器返回相关广告内容，大搜索服务器根据搜索算法匹配对应结果返回，如果按照这种方式通信，那么入口服务器需要编写两套代码分别同广告服务器和大搜索服务器进行交互，并且一个严重问题是如果其中广告服务器宕机了，会导致入口服务器无法正常工作进而影响大搜索服务器也无法正常工作！！
image.png
而引入阻塞队列后，入口服务器不需要知晓广告服务器和大搜索服务器的存在，只需要往阻塞队列中发送请求即可，而广告服务器和大搜索服务器也不需要知道入口服务器的存在，只需要从阻塞队列中取出请求处理完毕返回给阻塞队列即可，并且当其中大搜索服务器宕机时，不影响其他服务器以及入口服务器的正常运作！

  1. 削峰填谷：

image.png
如果没有阻塞队列，当遇到一些突发场景例如"双十一"大促等客户请求量激增的时候，入口服务器转发的请求量增多，压力就会变大，同理广告服务器和大搜索服务器处理过程复杂繁多，消耗的硬件资源就会激增，达到硬件瓶颈之后服务器就宕机了（直观现象就是客户端发送请求，服务器不会响应了）
image.png
而引入阻塞队列/消息队列之后，由于阻塞队列只负责存储相应的请求或者响应，无需额外的业务处理，因此抗压能力比广告服务器和大搜索服务器更强，当客户请求量激增的时候交由阻塞队列承受，而广告服务器和大搜索服务器只需要按照特定的速率进行读取并返回处理结果即可，就起到了 削峰填谷 的作用！

注意：此处的阻塞队列在现实场景中并不是一个单纯的数据结构，往往是一个基于阻塞队列的服务器程序，例如消息队列（MQ）

2. 标准库中的阻塞队列

2.1 基本介绍

Java标准库提供了现成的阻塞队列数据结构供开发者使用，即BlockingQueue接口
BlockingQueue：该接口具有以下实现类：

  1. ArrayBlockingQueue：基于数组实现的阻塞队列
  2. LinkedBlockingQueue：基于链表实现的阻塞队列
  3. PriorityBlockingQueue：带有优先级的阻塞队列

BlockingQueue方法：该接口具有以下常用方法

  1. 带有阻塞功能：
  • put：向队列中入元素，队列满则阻塞等待
  • take：向队列中取出元素，队列空则阻塞等待
  1. 不带有阻塞功能：
  • peek：返回队头元素（不取出）
  • poll：返回队头元素（取出）
  • offer：向队列中插入元素

2.2 代码示例

/**
 * 测试Java标准库提供的阻塞队列实现
 */
public class TestStandardBlockingQueue {

    private static BlockingQueue<Integer> queue = new ArrayBlockingQueue<>(10);
    public static void main(String[] args) {
        // 生产者
        Thread t1 = new Thread(() -> {
            int i = 0;
            while (true) {
                try {
                    queue.put(i);
                    System.out.println("生产数据：" + i);
                    i++;
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        // 消费者
        Thread t2 = new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(1000);
                    int ele = queue.take();
                    System.out.println("消费数据：" + ele);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
        });
        t1.start();
        t2.start();
    }
}

运行效果：
image.png
我们在主线程中创建了两个线程，其中t1线程作为生产者不断循环生产元素，而线程t2作为消费者每隔1s消费一个数据，所以我们很快看到当生产数据个数达到容量capacity时就会继续生产就会阻塞等待，直到消费者线程消费数据后才可以继续入队列，这样就实现了一个 生产者-消费者模型 ！

3. 自定义实现阻塞队列

首先我们需要明确实现一个阻塞队列需要哪些步骤？

  1. 首先我们需要实现一个普通队列
  2. 使用锁机制将普通队列变成线程安全的
  3. 通过特殊机制让该队列能够带有"阻塞"功能

3.1 实现普通队列

相信大家如果学过 数据结构与算法 相关课程，应该对队列这种数据结构的实现并不陌生！实现队列有基于数组的也有基于链表的，我们此处采用基于数组实现的，基于数组实现的循环队列也有以下两种方式：

  1. 腾出一个空间用来判断队列空或者满
  2. 使用额外的变量size用来记录当前元素的个数

我们使用第二种方式实现，实现代码如下：

/**
 * 自定义实现阻塞队列
 */
public class MyBlockingQueue {
    private int head = 0; // 头指针
    private int tail = 0; // 尾指针
    private int size = 0; // 当前元素个数
    private String[] array = null;
    private int capacity; // 容量

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
        this.array = new String[capacity];
    }

    /**
     * 入队列方法
     */
    public void put(String elem) {
        if (size == capacity) {
            // 队列已经满了
            return;
        }
        array[tail] = elem;
        tail++;
        if (tail >= capacity) {
            tail = 0;
        }
        size++;
    }

    /**
     * 出队列方法
     */
    public String take() {
        // 判断队列是否为空
        if (size == 0) {
            return null;
        }
        String topElem = array[head];
        head++;
        if (head >= capacity) {
            head = 0;
        }
        size--;
        return topElem;
    }

    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(3);
        queue.put("11");
        queue.put("22");
        queue.put("33");
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
    }
}

3.2 引入锁机制实现线程安全

引入synchronized关键字在原有队列实现的基础上实现线程安全，代码如下：

/**
 * 自定义实现阻塞队列
 */
public class MyBlockingQueue {
    private int head = 0; // 头指针
    private int tail = 0; // 尾指针
    private int size = 0; // 当前元素个数
    private String[] array = null;
    private int capacity; // 容量
    private Object locker = new Object(); // 锁对象

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
        this.array = new String[capacity];
    }

    /**
     * 入队列方法
     */
    public void put(String elem) {
        synchronized (locker) {
            if (size == capacity) {
                // 队列已经满了
                return;
            }
            array[tail] = elem;
            tail++;
            if (tail >= capacity) {
                tail = 0;
            }
            size++;
        }
    }

    /**
     * 出队列方法
     */
    public String take() {
        String topElem = "";
        synchronized (locker) {
            // 判断队列是否为空
            if (size == 0) {
                return null;
            }
            topElem = array[head];
            head++;
            if (head >= capacity) {
                head = 0;
            }
            size--;
        }
        return topElem;
    }

    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(3);
        queue.put("11");
        queue.put("22");
        queue.put("33");
        System.out.println(queue.take());
        System.out.println(queue.take());
        System.out.println(queue.take());
    }
}

我们在put、take等关键方法上将 多个线程修改同一个变量 部分的操作进行加锁处理，实现线程安全！

3.3 加入阻塞功能

在普通队列的实现中，如果队列满或者空我们直接使用return关键字返回，但是在多线程环境下我们希望实现阻塞等待的功能，这就可以使用Object类提供的wait/notify这组方法实现阻塞与唤醒机制了！我们就需要考虑阻塞与唤醒的时机了！
何时阻塞：这个问题非常简单，当队列满时入队列操作就应该阻塞等待，而当队列为空时出队列操作就需要阻塞等待
何时唤醒：想必大家都可以想到，对于入队列操作来说，只要队列不满就可以被唤醒，而对于出队列操作来说，队列不为空就可以被唤醒，因此，只要有线程调用take操作出队列，那么入队列的线程就可以被唤醒，而只要有线程调用put操作入队列，那么出队列的线程就可以被唤醒

/**
 * 自定义实现阻塞队列
 */
public class MyBlockingQueue {
    private int head = 0; // 头指针
    private int tail = 0; // 尾指针
    private int size = 0; // 当前元素个数
    private String[] array = null;
    private int capacity; // 容量
    private Object locker = new Object(); // 锁对象

    public MyBlockingQueue(int capacity) {
        this.capacity = capacity;
        this.array = new String[capacity];
    }

    /**
     * 入队列方法
     */
    public void put(String elem) {
        synchronized (locker) {
            while (size == capacity) {
                // 队列已经满了（进行阻塞）
                try {
                    locker.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            array[tail] = elem;
            tail++;
            if (tail >= capacity) {
                tail = 0;
            }
            size++;
            locker.notifyAll();
        }
    }

    /**
     * 出队列方法
     */
    public String take() {
        String topElem = "";
        synchronized (locker) {
            // 判断队列是否为空
            while (size == 0) {
                try {
                    locker.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            topElem = array[head];
            head++;
            if (head >= capacity) {
                head = 0;
            }
            size--;
            locker.notifyAll();
        }
        return topElem;
    }

    public static void main(String[] args) {
        MyBlockingQueue queue = new MyBlockingQueue(10);
        // 生产者
        Thread producer = new Thread(() -> {
            int i = 0;
            while (true) {
                queue.put(i + "");
                System.out.println("生产元素：" + i);
                i++;
            }
        });
        // 消费者
        Thread consumer = new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                String elem = queue.take();
                System.out.println("消费元素" + elem);
            }
        });
        producer.start();
        consumer.start();
    }
}

我们使用wait/notify这组操作实现了阻塞/唤醒功能，并且满足必须使用在synchronized关键字内部的使用条件，这里有一个注意点

为什么我们将if判断条件改成了while循环呢？？？这是需要考虑清楚的！

image.png
如图所示：一开始由于队列满所以生产者1进入阻塞状态，释放锁，然后生产者2也进入阻塞状态释放锁，此时消费者消费一个元素后唤醒生产者1，然后生产者1生产一个元素后（记住此时队列已满）继续唤醒，但是此时唤醒的恰恰是 生产者2 ，生产者2继续执行生产元素，于是就出现问题，我们总结一下出现问题的原因：

  1. notifyAll是随机唤醒，无法指定唤醒线程，因此可能出现生产者唤醒生产者，消费者唤醒消费者的情况
  2. if判定条件一经执行就无法继续判定，所以生产者2被唤醒后没有再次判断当前队列是否满

于是我们的应对策略就是使用while循环，当线程被唤醒使重新判断，如果队列仍满，入队列操作继续阻塞，而队列仍空，出队列操作继续阻塞！Java标准也推荐我们使用 while 关键字和 wait 关键字一起使用！
image.png

4. 应用场景（实现生产者消费者模型）

我们继续基于我们自定义实现的阻塞队列再来实现 生产者-消费者模型
代码示例（主函数）：

public static void main(String[] args) {
    MyBlockingQueue queue = new MyBlockingQueue(10);
    // 生产者
    Thread producer = new Thread(() -> {
        int i = 0;
        while (true) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            queue.put(i + "");
            System.out.println("生产元素：" + i);
            i++;
        }
    });
    // 消费者
    Thread consumer = new Thread(() -> {
        while (true) {
            String elem = queue.take();
            System.out.println("消费元素" + elem);
        }
    });
    producer.start();
    consumer.start();
}

运行效果：
image.png
此时我们创建两个两个线程，producer作为生产者线程每隔1s生产一个元素，consumer作为消费者线程不断消费元素，此时我们看到的就是消费者消费很快，当阻塞队列空时就进入阻塞状态，直到生产者线程生产元素后才被唤醒继续执行！此时我们真正模拟实现了 阻塞队列 这样的数据结构！

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

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

相关文章

【PWN · heap | Arbitrary Alloc】2015_9447ctf_search-engine

【PWN · heap | Arbitrary Alloc】2015_9447ctf_search-engine

和【PWN heap | House Of Spirit】2014_hack.lu_oreo-CSDN博客略有区别&#xff0c;但都是通过malloc一块fake_chunk到指定区域&#xff0c;获得对该区域的写权限 目录 零、简单介绍 一、题目分析 1.主要功能 2.index_sentence(): 增添一条语句到“库”中 3.search_word(…
阅读更多...
Leecode之分割链表

Leecode之分割链表

一.题目及剖析 https://leetcode.cn/problems/partition-list-lcci/description/ 二.思路引入 就是将其分成大小两个链表,以x为分界线进行分堆,最后再将两链表合并 三.代码引入 /*** Definition for singly-linked list.* struct ListNode {* int val;* struct Lis…
阅读更多...
Netty中的常用组件(三)

Netty中的常用组件(三)

ChannelPipeline 基于Netty的网路应用程序中根据业务需求会使用Netty已经提供的Channelhandler 或者自行开发ChannelHandler&#xff0c;这些ChannelHandler都放在ChannelPipeline中统一 管理&#xff0c;事件就会在ChannelPipeline中流动&#xff0c;并被其中一个或者多个Chan…
阅读更多...
【前端高频面试题--TypeScript篇】

【前端高频面试题--TypeScript篇】

&#x1f680; 作者 &#xff1a;“码上有前” &#x1f680; 文章简介 &#xff1a;前端高频面试题 &#x1f680; 欢迎小伙伴们 点赞&#x1f44d;、收藏⭐、留言&#x1f4ac; 高频前端面试题--Vue3.0篇 什么是TypeScript&#xff1f;TypeScript数据类型TypeScript中命名空…
阅读更多...
用python写个类

用python写个类

对类的简单概述 面向过程编程,是按照预定的顺序一步步执行,面向对象编程则是将相似的对象归为一类,定义它们的属性和行为,例如人有姓名,性别,年龄等属性,有吃饭,睡觉,学习等行为.这便是类,类是对象的模版,对象是类的实例,例如人是一个抽象的类,而张三是一个具体的对象. 学会用…
阅读更多...
机器学习系列——（十九）层次聚类

机器学习系列——（十九）层次聚类

引言 在机器学习和数据挖掘领域&#xff0c;聚类算法是一种重要的无监督学习方法&#xff0c;它试图将数据集中的样本分组&#xff0c;使得同一组内的样本相似度高&#xff0c;不同组间的样本相似度低。层次聚类&#xff08;Hierarchical Clustering&#xff09;是聚类算法中的…
阅读更多...
Blazor Wasm Gitee 码云登录

Blazor Wasm Gitee 码云登录

目录: OpenID 与 OAuth2 基础知识Blazor wasm Google 登录Blazor wasm Gitee 码云登录Blazor SSR/WASM IDS/OIDC 单点登录授权实例1-建立和配置IDS身份验证服务Blazor SSR/WASM IDS/OIDC 单点登录授权实例2-登录信息组件wasmBlazor SSR/WASM IDS/OIDC 单点登录授权实例3-服务端…
阅读更多...
分享66个表单按钮，总有一款适合您

分享66个表单按钮，总有一款适合您

分享66个表单按钮&#xff0c;总有一款适合您 66个表单按钮下载链接&#xff1a;https://pan.baidu.com/s/19lOG5sxI2Uy3KBIscffHRw?pwd8888 提取码&#xff1a;8888 Python采集代码下载链接&#xff1a;采集代码.zip - 蓝奏云 学习知识费力气&#xff0c;收集整理更不…
阅读更多...
VUE学习——数组变化侦测

VUE学习——数组变化侦测

官方文档 变更方法&#xff1a; 使用之后&#xff0c;ui可以直接发生改变。改变原数组 替换数组&#xff1a; 使用之后需要接受重新赋值&#xff0c;不然ui不发生改变。不改变原数组
阅读更多...
【开源】JAVA+Vue+SpringBoot实现实验室耗材管理系统

【开源】JAVA+Vue+SpringBoot实现实验室耗材管理系统

目录 一、摘要1.1 项目介绍1.2 项目录屏 二、功能模块2.1 耗材档案模块2.2 耗材入库模块2.3 耗材出库模块2.4 耗材申请模块2.5 耗材审核模块 三、系统展示四、核心代码4.1 查询耗材品类4.2 查询资产出库清单4.3 资产出库4.4 查询入库单4.5 资产入库 五、免责说明 一、摘要 1.1…
阅读更多...
股票均线的使用方法和实战技术，看涨看空的均线形态与案例教学

股票均线的使用方法和实战技术，看涨看空的均线形态与案例教学

一、教程描述 本套教程讲解了14种均线的特殊形态&#xff0c;通过直观图形以及大量案例的教学&#xff0c;将深奥、繁琐的均线变得生动与具体&#xff0c;广大投资者在认真学习以后&#xff0c;可以学会均线的使用方法&#xff0c;掌握最强的均线应用实战技术。本套教程不仅适…
阅读更多...
动态规划的一个初步学习

动态规划的一个初步学习

啥叫动态规划 在我们写很多的题目时&#xff0c;常常可以用暴力枚举来写&#xff0c;缺点就是速度太慢了。如果我们用一个数组或者哈希表&#xff08;虽然我还没学过哈希表&#xff09;将之前暴力枚举的数据储存起来&#xff0c;当再一次枚举到这个数字的时候就直接调用数组或…
阅读更多...
使用vue-client-only 解决组件不兼容SSR问题

使用vue-client-only 解决组件不兼容SSR问题

目录 前言 一、解决方案 1.基于Nuxt 框架的SSR应用 2.基于vue2框架的应用 3.基于vue3框架的应用 二、总结 往期回顾 前言 最近在我的单页面SSR应用上开发JSON编辑器功能&#xff0c;在引入组件后直接客户端跳转OK&#xff0c;但是在直接加载服务端渲染的时候一直报这…
阅读更多...
并发容器（Map、List、Set）实战及其原理

并发容器（Map、List、Set）实战及其原理

目录 JUC包下的并发容器 CopyOnWriteArrayList 应用场景 CopyOnWriteArrayList使用 CopyOnWriteArrayList原理 CopyOnWriteArrayList 的缺陷 扩展知识&#xff1a;迭代器的 fail-fast 与 fail-safe 机制 ConcurrentHashMap 应用场景 ConcurrentHashMap使用 数…
阅读更多...
阿里云幻兽帕鲁服务器免费搭建解决方法，白嫖阿里云

阿里云幻兽帕鲁服务器免费搭建解决方法，白嫖阿里云

阿里云幻兽帕鲁服务器免费搭建方案&#xff0c;先在阿里云高校计划「云工开物」活动领取学生专享300元无门槛代金券&#xff0c;幻兽帕鲁专用服务器4核16G配置26元1个月、149元半年&#xff0c;直接使用这个无门槛300元代金券抵扣即可免费搭建幻兽帕鲁服务器。阿里云服务器网al…
阅读更多...
2.8日学习打卡----初学RabbitMQ（三）

2.8日学习打卡----初学RabbitMQ（三）

2.8日学习打卡 一.springboot整合RabbitMQ 之前我们使用原生JAVA操作RabbitMQ较为繁琐&#xff0c;接下来我们使用 SpringBoot整合RabbitMQ&#xff0c;简化代码编写 创建SpringBoot项目&#xff0c;引入RabbitMQ起步依赖 <!-- RabbitMQ起步依赖 --> <dependency&g…
阅读更多...
小游戏和GUI编程(3) | 基于 SFML 的字符阵

小游戏和GUI编程(3) | 基于 SFML 的字符阵

小游戏和GUI编程(3) | 基于 SFML 的字符阵 1. 简介 使用 EasyX 图形库时&#xff0c; 官方第一个例子是字符阵。 EasyX 不开源&#xff0c; 也不能跨平台&#xff0c; API 陈旧&#xff0c; API 是 C 而不是 C。 现在使用 SFML 来实现字符阵&#xff0c; 克服 EasyX 的这些问…
阅读更多...
OCP使用CLI创建和构建应用

OCP使用CLI创建和构建应用

文章目录 环境登录创建project赋予查看权限部署第一个image创建route检查pod扩展应用 部署一个Python应用连接数据库创建secret加载数据并显示国家公园地图 清理参考 环境 RHEL 9.3Red Hat OpenShift Local 2.32 登录 通过 crc console --credentials 可以查看登录信息&…
阅读更多...
动态内存管理（下）

动态内存管理（下）

1.常见的动态内存的错误 我们在学习动态内存的时候&#xff0c;常出现的一些错误我们来看一下。 1.对NULL指针的解引用操作 例如我们在使用malloc或者calloc开辟动态空间的时候&#xff0c;有时候没有判断是否开辟成功而直接对齐的返回指针进行解引用&#xff0c;此时如果开…
阅读更多...
[论文总结] 深度学习在农业领域应用论文笔记12

[论文总结] 深度学习在农业领域应用论文笔记12

文章目录 1. 3D-ZeF: A 3D Zebrafish Tracking Benchmark Dataset (CVPR, 2020)摘要背景相关研究所提出的数据集方法和结果个人总结 2. Automated flower classification over a large number of classes (Computer Vision, Graphics & Image Processing, 2008)摘要背景分割…
阅读更多...
最新文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发 尧图建网站