数据结构与算法——Java实现 31.阻塞队列

news2025/1/17 5:56:30

                        —— 24.10.8

一、问题提出

目前队列存在的问题

1.很多场景要求分离生产者、消费者两个角色、它们需要由不同的线程来担当，而之前的实现根本没有考虑线程安全问题

2.poll方法，队列为空，那么在之前的实现里会返回null，如果就是硬要拿到一个元素呢?以现在的实现只能不断循环尝试

3.offer方法，队列为满，那么在之前的实现里会返回false，如果就是硬要塞入一个元素呢?以现在的实现只能不断循环尝试

4.指令交错，多个线程会造成混乱效果

二、解决方法

 为解决线程不安全问题，需要给线程加锁，使线程局部阻塞

用条件变量让 poll 或 offer 线程进入等待 状态，而不是不断循环尝试，让CPU空转

三、单锁实现

Java中两种锁的选择

① synchronized：关键字，功能少

② ReentrantLock：可重入锁，功能丰富

lock() 加锁

unlock() 解锁

lockInterruptibly() 加锁（可在阻塞时打断，提前唤醒）

offer方法实现

if判断

问题

从 tailWaits 中唤醒的线程，会与新来的 offer 的线程争抢锁，谁能抢到是不一定的，如果后者先抢到，就会导致条件又发生变化

这种情况称之为虚假唤醒，唤醒后应该重新检查条件，看是不是得重新进入等待

    public void offer(String e) throws InterruptedException {
        // 加锁，可重入锁阻塞时可打断方法（可被强制唤醒）
        lock.lockInterruptibly();
        try {
            // 判断是否为满
            if (isFull()){
                // 队列满时，使offer线程阻塞,直到poll线程取走后，有位置时再恢复运行
                // tail.signal() 唤醒线程
                tailWaits.await();
            }
            array[tail] = e;
            if (++tail == array.length){
                tail = 0;
            }
            size++;
        }finally {
            // 解锁
            lock.unlock();
        }
    }
import java.util.Arrays;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class TestThreadUnsafe {
    private final String[] array = new String[10];
    // 尾指针
    private int tail = 0;
    // 元素个数
    private int size = 0;
    // 创建一个可重入锁对象
    ReentrantLock lock = new ReentrantLock();
    // 条件变量对象（集合线程）
    Condition tailWaits = lock.newCondition();

    public void offer(String e) throws InterruptedException {
        // 加锁，可重入锁阻塞时可打断方法（可被强制唤醒）
        lock.lockInterruptibly();
        try {
            // 判断是否为满
            if (isFull()){
                // 队列满时，使offer线程阻塞,直到poll线程取走后，有位置时再恢复运行
                // tail.signal() 唤醒线程
                tailWaits.await();
            }
            array[tail] = e;
            if (++tail == array.length){
                tail = 0;
            }
            size++;
        }finally {
            // 解锁
            lock.unlock();
        }
    }

    public void poll(String e) throws InterruptedException {

    }

    @Override
    public String toString() {
        return Arrays.toString(array);
    }

    private boolean isFull(){
        return size == array.length;
    }

    private boolean isEmpty(){
        return size == 0;
    }

    public static void main(String[] args) throws InterruptedException{
        TestThreadUnsafe queue = new TestThreadUnsafe();
        for (int i = 0; i < 10; i++) {
            queue.offer("e"+i);
        }
        new Thread(()->{
            try {
                System.out.println(Thread.currentThread().getName()+"添加元素之前");
                queue.offer("e10");
                System.out.println(Thread.currentThread().getName()+"添加元素成功");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        },"t1").start();

        new Thread(()->{
            System.out.println("开始唤醒");
            try{
                queue.lock.lockInterruptibly();
                queue.tailWaits.signal();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }finally {
                queue.lock.unlock();
            }
        },"t2").start();
    }
}

while判断 

解决了虚假唤醒的问题

    @Override
    public void offer(E e) throws InterruptedException {    // poll 等待队列非空
        lock.lockInterruptibly();
        try{
            while (isFull()){
                // 放在条件变量等待
                tailWaits.await();
            }
            array[tail] = e;
            if (++tail == array.length){
                tail = 0;
            }
            size++;
            // 唤醒等待线程
            headWaits.signal();
        }finally {
            lock.unlock();
        }
    }
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BlockingQueue1<E> implements BlockingQueue<E> {
    private final E[] array;
    private int head;
    private int tail;
    private int size;

    // 根据容量创造一个数组
    public BlockingQueue1(int capacity) {
        array = (E[]) new Object[capacity];
    }

    // 加可重入锁
    private ReentrantLock lock = new ReentrantLock();
    // 配合poll方法条件变量，在队列头部删除
    private Condition headWaits = lock.newCondition();
    // 配合offer方法条件变量，在队列尾部加入
    private Condition tailWaits = lock.newCondition();

    // 判空
    private boolean isEmpty(){
        return head == tail;
    }

    // 判满
    private boolean isFull(){
        return size == array.length;
    }

    @Override
    public String toString() {
        return "array=" + Arrays.toString(array);
    }

    @Override
    public void offer(E e) throws InterruptedException {    // poll 等待队列非空
        lock.lockInterruptibly();
        try{
            while (isFull()){
                // 放在条件变量等待
                tailWaits.await();
            }
            array[tail] = e;
            if (++tail == array.length){
                tail = 0;
            }
            size++;
            // 唤醒等待线程
            headWaits.signal();
        }finally {
            lock.unlock();
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        lock.lockInterruptibly();
        try{
            // 将毫秒时间转换为纳秒时间
            long t = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()){
                if (t<=0){
                    return false;
                }
                // 最多等待多少纳秒
                tailWaits.awaitNanos(t);
            }
            array[tail] = e;
            if (++tail == array.length){
                tail = 0;
            }
            size++;
            // 唤醒等待线程
            headWaits.signal();
            return true;
        }finally {
            lock.unlock();
        }
    }

    @Override
    public E poll() throws InterruptedException {
        lock.lockInterruptibly();
        try {
            while (isEmpty()){
                headWaits.await();
            }
            E e = array[head];
            array[head] = null;
            if (++head == array.length){
                head = 0;
            }
            size--;
            tailWaits.signal();
            return e;
        }finally {
            lock.unlock();
        }
    }
}

 main函数

public class TestBlockingQueue1 {
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue1<String> queue = new BlockingQueue1<>(3);

        Thread t1 = new Thread(() -> {
            try {
                System.out.println(System.currentTimeMillis()+" begin ");
                queue.offer("任务1");
                System.out.println(queue);
                queue.offer("任务2");
                System.out.println(queue);
                queue.offer("任务3");
                System.out.println(queue);
                queue.offer("任务4",5000);
                System.out.println(queue);
                System.out.println(System.currentTimeMillis() + " end ");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        },"生产者");
        t1.start();

        Thread.sleep(2000);
        queue.poll();
    }
}

 

四、双锁实现

单锁问题：

单锁实现的缺陷：两个线程用了同一把锁，一个执行时，另一个就需阻塞，而offer方法添加元素和poll方法取走元素使用了同一把锁，这样两个线程不能同时执行，两方法相互阻塞

解决方法：

offer方法主要操作尾指针，poll方法主要操作头指针，将offer方法和poll方法分别添加一个锁，用两把锁分别保护头指针和尾指针，从而分别保护offer和poll两个方法

代码实现

import java.util.Arrays;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

public class BlockingQueue2<E> implements BlockingQueue<E> {
    private final E[] array;
    private int head;
    private int tail;
    private int size;

    // tailLock给offer方法入队用,Condition分别创建两个等待的条件变量
    private ReentrantLock tailLock = new ReentrantLock();
    private Condition notEmpty = tailLock.newCondition();
    // headLock给poll方法出队用
    private ReentrantLock headLock = new ReentrantLock();
    private Condition notFull = headLock.newCondition();

    public BlockingQueue2(int capacity) {
        this.array = (E[]) new Object[capacity];
    }

    // 判空
    private boolean isEmpty(){
        return size == 0;
    }

    // 判满
    private boolean isFull(){
        return size == array.length;
    }

    @Override
    public String toString() {
        return "array=" + Arrays.toString(array);
    }

    @Override
    public void offer(E e) throws InterruptedException {
        // 加锁
        tailLock.lockInterruptibly();
        try{
            while (isFull()) {
                notEmpty.await();
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            size++;
        }finally {
            tailLock.unlock();
        }
    }

    @Override
    public boolean offer(E e, long timeout) throws InterruptedException {
        return false;
    }

    @Override
    public E poll() throws InterruptedException {
        // 加锁
        headLock.lockInterruptibly();
        try{
            while (isEmpty()){
                notEmpty.await();
            }
            E e = array[head];
            array[head] = null;
            if (++head == array.length) {
                head = 0;
            }
            size--;
            return e;
        }finally {
            headLock.unlock();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue2<String> queue = new BlockingQueue2<>(3);
        queue.offer("任务1");

        new Thread(()->{
            try {
                queue.offer("任务2");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        },"offer").start();

        new Thread(()->{
            try {
                System.out.println(queue.poll());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        },"poll").start();
    }
}

size自增/自减问题

size的自增自减不能保障安全，size自增自减在多个线程同时执行时可能遇到冲突

解决方法

用原子变量AtomicInteger类型保证安全

getAndIncrement 自增方法，能保证线程安全

getAndDecrement 自减方法，能保证线程安全

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

Spring Boot 进阶-深入SpringBoot的数据校验原理

Spring Boot 进阶-深入SpringBoot的数据校验原理

在之前的分析中我们知道要想了解一个场景启动器的原理就必须要找到它对应的自动配置类。下面我们就来探索一下数据校验spring-boot-starter-validation场景启动器的原理吧？ ValidationAutoConfiguration 配置类 首先我们来看在这个配置类上都有哪些条件注解，并且这些条件注解…
阅读更多...
鸿蒙开发实战项目【硅谷租房】--- 项目介绍

鸿蒙开发实战项目【硅谷租房】--- 项目介绍

目录 一、简述 二、项目资料 2.1 UI设计稿 2.2 服务器 2.3 Apifox接口JSON文件 使用 Apifox 测试接口 一、简述 这是一个基于 鸿蒙 API12 开发的移动端租房 App&#xff0c;用户可以使用该应用搜索租房列表、查看房屋详情、预约租房等。 该项目的tabbar包含五部分&…
阅读更多...
OpenTelemetry 演示与 OpenTelemetry 的 Elastic 分发

OpenTelemetry 演示与 OpenTelemetry 的 Elastic 分发

作者&#xff1a;来自 Elastic Roger Coll 了解 Elastic 如何致力于支持用户使用 OpenTelemetry。探索我们对 OpenTelemetry Demo 的公开部署&#xff0c;并了解 Elastic 的解决方案如何增强你的可观察性体验。 最近&#xff0c;Elastic 为各种 OpenTelemetry 组件引入了 Elas…
阅读更多...
谨防网络诈骗，天上不会掉馅儿饼

谨防网络诈骗，天上不会掉馅儿饼

“网络诈骗现已是国际性问题。” 面对网络诈骗&#xff0c;风险管理工具要选且必须选。网络诈骗日益猖獗&#xff0c;当前背景下&#xff0c;IP风险画像也开始成为了防范网络诈骗的重要工具之一。 近期&#xff0c;一起利用AI换脸拟声技术实施的诈骗案件引起了广泛关注。 一名…
阅读更多...
C# 获取可执行文件目录

C# 获取可执行文件目录

---------------------------------------------------------------------------
阅读更多...
OCP迎来新版本，让OceanBase的运维管理更高效

OCP迎来新版本，让OceanBase的运维管理更高效

近期&#xff0c;OceanBase的OCP发布了新版本&#xff0c;全面支持 OceanBase 内核 4.3.2 及更低版本。新版本针对基础运维、性能监控、运维配置、外部集成等多个方面实现了 20余项的优化及强化措施&#xff0c;增强产品的易用性和稳定性&#xff0c;从而帮助用户更加高效地管理…
阅读更多...
前端 接口扁平化转换树数据的方法并添加序号

前端 接口扁平化转换树数据的方法并添加序号

最终实现图 . 1. 接口扁平化数据转树型结构数据 &#xff08;递归&#xff09; // 转换树行数据的方法 export const transTree (list, pidKey, idKey id) > {// 最终生成的树行结构const treeData []// 对传入进来的 数据进行遍历&#xff0c;查找对应的子级list.for…
阅读更多...
tauri开发Mac电脑Safari浏览器一个很奇怪的问题：在 input 输入框输入的是全小写英文字母，会自动将首字母转换为大写解决办法

tauri开发Mac电脑Safari浏览器一个很奇怪的问题：在 input 输入框输入的是全小写英文字母，会自动将首字母转换为大写解决办法

问题原因 在 Mac 系统中默认使用 Safari 的内核 WKWebView 作为渲染引擎&#xff0c;而 Safari 浏览器的一些 “人性化” 机制&#xff1a;如果输入框中输入的是全小写英文&#xff0c;会自动将首字母转换为大写。 解决办法 我只需要禁止这个默认的行为&#xff0c;即可解决这…
阅读更多...
循环神经网络-RNN

循环神经网络-RNN

文章目录 前言一、RNN介绍1.基本结构2.隐藏态特点3.输出计算 二、RNN循环由来与局限三、RNN延申 前言 因为传统神经网络无法训练出具有顺序的数据且模型搭建时没有考虑数据上下之间的关系。所以我们提出了循环神经网络。 一、RNN介绍 循环神经网络&#xff08;Recurrent Neu…
阅读更多...
LabVIEW惯性导航系统仿真平台

LabVIEW惯性导航系统仿真平台

LabVIEW开发捷联惯性导航系统仿真平台&#xff0c;采用模块化设计&#xff0c;利用LabVIEW的图形化编程特性&#xff0c;提高了系统仿真的效率和精度&#xff0c;同时具备良好的可扩展性和用户交互性。 项目背景 当前&#xff0c;惯性导航系统&#xff08;INS&#xff09;的研…
阅读更多...
EMC-LISN是什么

EMC-LISN是什么

LISN&#xff0c;Line Impedance Stabilization Network的缩写&#xff0c;即线路阻抗稳定网络 LISN主要是用于EMI测试中传导发射&#xff08;CE&#xff09;的 我们希望只测试到待测产品传导出来的干扰&#xff0c;能够不测试到电源输入本身上面的干扰。 测试场地不同&…
阅读更多...
如何避免PuTTY的连接超时

如何避免PuTTY的连接超时

问题&#xff1a;使用PuTTY默认创建的SSH连接&#xff0c;过一会就会提示“Remote side unexpectedly closed network connection" 解决方法&#xff1a; 要防止PuTTY会话由于空闲而断开连接&#xff0c;可以通过启用keep-alives功能&#xff0c;使PuTTY定期向远程主机发…
阅读更多...
【实时计算 Flink】检查点和快照超时的诊断方法与调优策略

【实时计算 Flink】检查点和快照超时的诊断方法与调优策略

Flink的状态管理是一个复杂而关键的领域&#xff0c;涉及到作业的性能、稳定性和资源利用等多个方面。通过对状态生成机制和优化策略地深入理解与正确应用&#xff0c;结合实时计算Flink版提供的产品能力&#xff0c;可以帮您有效地优化Flink作业以应对大规模状态作业带来的挑战…
阅读更多...
PigGo的安装配置

PigGo的安装配置

TyporaPigGo的使用 1、下载PigGo 1.1、先打开Typora的偏好设置 1.2 下载PigGo app 1.3 下载Setup-xxx.exe版本 1.4安装成功后打开PigGo 2、配置PigGo 1、安装gitee-uploader插件 2、 PicGo设置 ​ ​ 3、图床设置 3.1配置SM.MS 3.1.1 登录sm,生成秘钥 如果没有注册&…
阅读更多...
期权懂|期权交易涨跌幅限制会随时调整吗？

期权懂|期权交易涨跌幅限制会随时调整吗？

本期让我懂 你就懂的期权懂带大家来了解&#xff0c;期权交易涨跌幅限制会随时调整吗&#xff1f;有兴趣的朋友可以看一下。期权小懂每日分享期权知识&#xff0c;帮助期权新手及时有效地掌握即市趋势与新资讯&#xff01; 期权交易涨跌幅限制会随时调整吗&#xff1f; 涨跌幅…
阅读更多...
天玑 9400 基本确认：4大升级，一代“冰龙”来了

天玑 9400 基本确认：4大升级，一代“冰龙”来了

去年&#xff0c;天玑9300 破釜沉舟&#xff0c;打破了A系不可击败的神话。但今年&#xff0c;对安卓阵营来说&#xff0c;才是扬眉吐气的时刻。 因为芯片人才的流失&#xff0c;果子已经雄风不再。即使是 4nm 工艺打3nm工艺&#xff0c;天玑 9300 的 GPU效能&#xff0c;也压…
阅读更多...
机器学习笔记-1

机器学习笔记-1

文章目录 前言一、How to find a function二、Define Loss from Training Data三、Optimization总结 前言 机器学习&#xff08;Machine Learning, ML&#xff09;是一门让计算机通过数据来自动学习和改进的技术。它的核心理念是通过分析大量的历史数据来找到其中的规律&#…
阅读更多...
YOLOv10改进策略【注意力机制篇】| CVPR2024 CAA上下文锚点注意力机制

YOLOv10改进策略【注意力机制篇】| CVPR2024 CAA上下文锚点注意力机制

一、本文介绍 本文记录的是基于CAA注意力模块的YOLOv10目标检测改进方法研究。在远程遥感图像或其他大尺度变化的图像中目标检测任务中&#xff0c;为准确提取其长距离上下文信息&#xff0c;需要解决大目标尺度变化和多样上下文信息时的不足的问题。CAA能够有效捕捉长距离依赖…
阅读更多...
Spark算子使用-Map,FlatMap,Filter,diatinct,groupBy,sortBy

Spark算子使用-Map,FlatMap,Filter,diatinct,groupBy,sortBy

目录 Map算子使用 FlatMap算子使用 Filter算子使用-数据过滤 Distinct算子使用-数据去重 groupBy算子使用-数据分组 sortBy算子使用-数据排序 Map算子使用 # map算子主要使用长场景&#xff0c;一个转化rdd中每个元素的数据类型&#xff0c;拼接rdd中的元素数据&#xf…
阅读更多...
NUKE 15有哪些新的改进功能？影视后期特效合成NUKE 15 安装包分享 【Mac/win】

NUKE 15有哪些新的改进功能？影视后期特效合成NUKE 15 安装包分享 【Mac/win】

Nuke 15是一款由英国The Foundry公司开发的专业的合成软件&#xff0c;被广泛用于电影、电视和广告制作中的后期合成和特效制作。 Nuke 15拥有强大的功能和灵活性&#xff0c;可以帮助用户处理各种复杂的合成任务&#xff0c;包括图像修复、色彩校正以及粒子特效等。它具备高效…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023