Java阻塞队列

news2024/10/6 22:33:58

什么阻塞队列

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作支持阻塞的插入和移除方法。

  1. 支持阻塞的插入方法:意思是当队列满时,队列会阻塞插入元素的线程,直到队列不满。
  2. 支持阻塞的移除方法:意思是在队列为空时,获取元素的线程会等待队列变为非空。

阻塞队列常用于生产者和消费者的场景,生产者是向队列里添加元素的线程,消费者是从队列里取元素的线程。阻塞队列就是生产者用来存放元素、消费者用来获取元素的容器。

四组API

image-20230731214338293

  • 抛出异常:当队列满时,如果再往队列里插入元素,会抛出IllegalStateException(“Queue full”)异常。当队列空时,从队列里获取元素会抛出NoSuchElementException异常。
  • 返回特殊值:当往队列插入元素时,会返回元素是否插入成功,成功返回true。如果是移除方法,则是从队列里取出一个元素,如果没有则返回null。
  • 一直阻塞:当阻塞队列满时,如果生产者线程往队列里put元素,队列会一直阻塞生产者线程,直到队列可用或者响应中断退出。当队列空时,如果消费者线程从队列里take元素,队列会阻塞住消费者线程,直到队列不为空。
  • 超时退出:当阻塞队列满时,如果生产者线程往队列里插入元素,队列会阻塞生产者线程一段时间,如果超过了指定的时间,生产者线程就会退出。

注意:如果是无界阻塞队列,队列不可能会出现满的情况,所以使用put或offer方法永远不会被阻塞,而且使用offer方法时,该方法永远返回true。

Java中7种阻塞队列

image-20230801093001063

ArrayBlockingQueue:一个由数组结构组成的有界阻塞队列。

LinkedBlockingQueue:一个由链表结构组成的有界阻塞队列。

PriorityBlockingQueue:一个支持优先级排序的无界阻塞队列。

DelayQueue:一个使用优先级队列实现的无界阻塞队列。

SynchronousQueue:一个不存储元素的阻塞队列。

LinkedTransferQueue:一个由链表结构组成的无界阻塞队列。

LinkedBlockingDeque:一个由链表结构组成的双向阻塞队列。

1.ArrayBlockingQueue

ArrayBlockingQueue是一个用数组实现的有界阻塞队列。

默认情况下不保证线程公平的访问队列,所谓公平访问队列是指阻塞的线程,可以按照阻塞的先后顺序访问队列,即先阻塞线程先访问队列非公平性是对先等待的线程是非公平的,当队列可用时,阻塞的线程都可以争夺访问队列的资格,有可能先阻塞的线程最后才访问队列为了保证公平性,通常会降低吞吐量。

下面是公平阻塞队列:

ArrayBlockingQueue fairQueue = new  ArrayBlockingQueue(1000, true);

public ArrayBlockingQueue(int capacity, boolean fair) {
        if (capacity <= 0)
            throw new IllegalArgumentException();
        this.items = new Object[capacity];
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
}

访问者的公平性是使用可重入锁实现的。

2.LinkedBlockingQueue

LinkedBlockingQueue是一个用链表实现的有界阻塞队列。此队列的默认和最大长度为Integer.MAX_VALUE。此队列按照先进先出的原则对元素进行排序。

3.PriorityBlockingQueue

PriorityBlockingQueue是一个支持优先级的无界阻塞队列。默认情况下元素采取自然顺序升序排列。也可以自定义类实现compareTo()方法来指定元素排序规则,或者初始化PriorityBlockingQueue时,指定构造参数Comparator来对元素进行排序。需要注意的是不能保证同优先级元素的顺序。

4.DelayQueue *

DelayQueue是一个支持延时获取元素的无界阻塞队列队列使用PriorityQueue来实现队列中的元素必须实现Delayed接口,在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。

image-20230801093608736

可以将DelayQueue运用在以下应用场景:

缓存系统的设计:可以用DelayQueue保存缓存元素的有效期,使用一个线程循环查询DelayQueue,一旦能从DelayQueue中获取元素时,表示缓存有效期到了。

定时任务调度:使用DelayQueue保存当天将会执行的任务和执行时间,一旦从DelayQueue中获取到任务就开始执行,比如TimerQueue就是使用DelayQueue实现的。

(1)如何实现Delayed接口

DelayQueue队列的元素必须实现Delayed接口。

public interface Delayed extends Comparable<Delayed> {
    long getDelay(TimeUnit unit);
}

参考ScheduledThreadPoolExecutor里ScheduledFutureTask类的实现,一共有三步。

第一步:在对象创建的时候,初始化基本数据。使用time记录当前对象延迟到什么时候可以使用,使用sequenceNumber来标识元素在队列中的先后顺序

       private static final AtomicLong sequencer = new AtomicLong();
	   ScheduledFutureTask(Runnable r, V result, long ns, long period) {
            super(r, result);
            this.time = ns;
            this.period = period;
            this.sequenceNumber = sequencer.getAndIncrement();
        }

第二步:实现getDelay方法,该方法返回当前元素还需要延时多长时间,单位是纳秒

        public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), NANOSECONDS);
        }

延迟时间参数的单位是纳秒,自己设计的时候最好使用纳秒,实现getDelay()方法时可以指定任意单位,一旦以秒或分作为单位,而延时时间又精确不到纳秒会出现误差。

第三步:实现compareTo方法来指定元素的顺序。

下述代码让延时时间最长的放在队列的末尾:

        public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                return 0;
            if (other instanceof ScheduledFutureTask) {
                ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
                long diff = time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
                else if (sequenceNumber < x.sequenceNumber)
                    return -1;
                else
                    return 1;
            }
            long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }

(2)如何实现延时阻塞队列

当消费者从队列里获取元素时,如果元素没有达到延时时间,就阻塞当前线程。

take方法:

long delay = first.getDelay(TimeUnit.NANOSECONDS);
if (delay <= 0)
    return q.poll();
else if (leader != null)
        available.await();
else {
    Thread thisThread = Thread.currentThread();
	leader = thisThread;
        try {
                available.awaitNanos(delay);
            } finally {
                if (leader == thisThread)
                leader = null;
            }
    }

变量leader是一个等待获取队列头部元素的线程。

如果leader不等于空,表示已经有线程在等待获取队列的头元素。使用await()方法让当前线程等待信号。

如果leader等于空,则把当前线程设置成leader,并使用awaitNanos()方法让当前线程等待接收信号或等待delay时间。

5.SynchronousQueue

SynchronousQueue是一个不存储元素的阻塞队列。每一个put操作必须等待一个take操作,否则不能继续添加元素。

它支持公平访问队列。默认情况下线程采用非公平性策略访问队列。构造设置为true,则等待的线程会采用先进先出的顺序访问队列。

image-20230731225333708

队列本身并不存储任何元素,非常适合传递性场景(负责把生产者线程处理的数据直接传递给消费者线程)。SynchronousQueue的吞吐量高于LinkedBlockingQueue和ArrayBlockingQueue。

6.LinkedTransferQueue

LinkedTransferQueue是一个由链表结构组成的无界阻塞TransferQueue队列。相对于其他阻塞队列,LinkedTransferQueue多了tryTransfer和transfer方法

(1)transfer方法

如果当前有消费者正在等待接收元素(消费者使用take()方法或带时间限制的poll()方法时),transfer方法可以把生产者传入的元素立刻transfer(传输)给消费者。

如果没有消费者在等待接收元素,transfer方法会将元素存放在队列的tail节点,并等到该元素被消费者消费了才返回。

transfer方法的关键代码:

// ...
Node pred = tryAppend(s, haveData);
return awaitMatch(s, pred, e, (how == TIMED), nanos);

第一行试图把存放当前元素的s节点作为tail节点。

第二行让CPU自旋等待消费者消费元素。因为自旋会消耗CPU,所以自旋一定的次数后使用Thread.yield()方法来暂停当前正在执行的线程,并执行其他线程。

(2)tryTransfer方法

tryTransfer方法是用来试探生产者传入的元素是否能直接传给消费者。 如果没有消费者等待接收元素,则返回false。和transfer方法的区别是tryTransfer方法无论消费者是否接收,方法立即返回,而transfer方法是必须等到消费者消费了才返回

带有时间限制的tryTransfer(E e,long timeout,TimeUnit unit)方法,试图把生产者传入的元素直接传给消费者,但是如果没有消费者消费该元素则等待指定的时间再返回,如果超时还没消费元素,则返回false,如果在超时时间内消费了元素,则返回true。

7.LinkedBlockingDeque

LinkedBlockingDeque是一个由链表结构组成的双向阻塞队列。所谓双向队列指的是可以从队列的两端插入和移出元素。

双向队列多了一个操作队列的入口,在多线程同时入队时,也就减少了一半的竞争。

LinkedBlockingDeque多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法,以First单词结尾的方法,表示插入、获取(peek)或移除双端队列的第一个元素。以Last单词结尾的方法,表示插入、获取或移除双端队列的最后一个元素。另外,插入方法add等同于addLast,移除方法remove等效于removeFirst。

在初始化LinkedBlockingDeque时可以设置容量防止其过度膨胀。另外,双向阻塞队列可以运用在“工作窃取”模式中。

阻塞队列的实现原理

如果队列是空的,消费者会一直等待,当生产者添加元素时,消费者是如何知道当前队列有元素的呢?

使用通知模式实现。所谓通知模式,就是当生产者往满的队列里添加元素时会阻塞住生产者,当消费者消费了一个队列中的元素后,会通知生产者当前队列可用。

以ArrayBlockingQueue为例,ArrayBlockingQueue使用了Condition来实现(await、signal方法):

// 队列两种状态,空或者满 两个Condition对象 等待通知
private final Condition notFull;
private final Condition notEmpty;
public ArrayBlockingQueue(int capacity, boolean fair) {
        // ... 
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }
public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}
private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

调用Condition#await方法,会将同步队列中的头节点获取锁的线程(释放锁)转移到等待队列中。

当往队列里插入一个元素时,如果队列不可用,那么阻塞生产者主要通过LockSupport.park(this)来实现。

        public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            // 此时当前节点已经不在同步队列中,while死循环
            while (!isOnSyncQueue(node)) {
                // 阻塞当前线程
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

继续进入源码,发现调用setBlocker先保存一下将要阻塞的线程,然后调用unsafe.park阻塞当前线程。

    public static void park(Object blocker) {
        Thread t = Thread.currentThread();
        setBlocker(t, blocker);
        UNSAFE.park(false, 0L);
        setBlocker(t, null);
    }

unsafe.park是个native方法:

public native void park(boolean isAbsolute, long time);

park这个方法会阻塞当前线程,只有以下4种情况中的一种发生时,该方法才会返回。

  1. 与park对应的unpark执行或已经执行时。“已经执行”是指unpark先执行,然后再执行park的情况。
  2. 线程被中断时。
  3. 等待完time参数指定的毫秒数时。
  4. 异常现象发生时,这个异常现象没有任何原因。

JVM是如何实现park方法:park在不同的操作系统中使用不同的方式实现,在Linux下使用的是系统方法pthread_cond_wait实现。实现代码在JVM源码路径src/os/linux/vm/os_linux.cpp里的os::PlatformEvent::park方法:

void os::PlatformEvent::park() {
    int v ;
            for (;;) {
                v = _Event ;
            if (Atomic::cmpxchg (v-1, &_Event, v) == v) break ;
            }
            guarantee (v >= 0, "invariant") ;
            if (v == 0) {
            // Do this the hard way by blocking ...
            int status = pthread_mutex_lock(_mutex);
            assert_status(status == 0, status, "mutex_lock");
            guarantee (_nParked == 0, "invariant") ;
            ++ _nParked ;
            while (_Event < 0) {
            // !!!    
            status = pthread_cond_wait(_cond, _mutex);
            // for some reason, under 2.7 lwp_cond_wait() may return ETIME ...
            // Treat this the same as if the wait was interrupted
            if (status == ETIME) { status = EINTR; }
            assert_status(status == 0 || status == EINTR, status, "cond_wait");
            }
            -- _nParked ;
            // In theory we could move the ST of 0 into _Event past the unlock(),
            // but then we'd need a MEMBAR after the ST.
            _Event = 0 ;
            status = pthread_mutex_unlock(_mutex);
            assert_status(status == 0, status, "mutex_unlock");
            }
            guarantee (_Event >= 0, "invariant") ;
            }
    }

pthread_cond_wait是一个多线程的条件变量函数,cond是condition的缩写,字面意思可以理解为线程在等待一个条件发生,这个条件是一个全局变量。方法接收两个参数:一个共享变量 _cond,一个互斥量 _mutex。而unpark方法在Linux下是使用pthread_cond_signal实现的。park方法在Windows下则是使用WaitForSingleObject实现的。

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

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

相关文章

高并发编程-2. 并发级别

此文章为笔记&#xff0c;为阅读其他文章的感受、补充、记录、练习、汇总&#xff0c;非原创&#xff0c;感谢每个知识分享者。 原文 文章目录 阻塞无饥饿(Starvation-Free)无障碍(Obstruction-Free)无锁(Lock-Free)无等待 由于临界区的存在&#xff0c;多线程之间的并发必须受…

PCIE上位机用什么工具?

可以了解一下神电测控出器的My FPGA开发套件&#xff0c;它可以用来开发FPGA板卡与上位机之间PCIE通信&#xff0c;而且就是用LabVIEW FPGA开发。它使用的是Xillybus PCIe IP核&#xff0c;神电将其封装成了在 LabVIEW FPGA 下的 PCIe CLIP 组件&#xff0c;可以方便的使用。而…

2023年最新前端学习路线【超详细版】

第一个月、HTMLCSS 第1周安排&#xff1a; 是了解前端入门的第一步&#xff0c;要学会IDE插件安装和使用&#xff0c;掌握H5语法和基础标签的应用&#xff0c;学完本周可以进行简单的页面搭建。 1、了解前端行业发展2、了解HTML5对于网页的意义3、学习常用标签/属性&#xf…

流控平台Sentinel搭建和接入教程

流量控制和限流是大型系统必不可少的组成部分&#xff0c;Sentinel是Alibaba提供的一款特别好用的专业工具&#xff0c;属于那种看起来很牛&#xff0c;用起来也很牛的工具&#xff0c;下面记录一下接入的过程。 一&#xff0c;搭建平台 1&#xff0c;下载jar包 地址&#x…

组件(lvs,keeplive,orm,mysql,分布式事务)

lvs LVS 已经集成到Linux内核系统中&#xff0c;ipvsadm 是 LVS 的命令行管理工具。 目前有三种 IP 负载均衡技术&#xff08; VS/NAT 网络地址转换 、VS/TUN IP 隧道技术实现虚拟服务器 和 VS/DR 直接路由&#xff09;&#xff1b; 八种调度算法&#xff1a;轮询 …

解决ComposerStaticInit类重复加载问题

Tp5中composer安装phpoffice后报错 Cannot declare class Composer\Autoload\ComposerStaticInit6c9266e89aa536667bf36f59ad8e6e6b, because the name is already in use 错误原因: 使用ThinkPHP5框架的项目&#xff0c;在引用某些包的时候&#xff0c;可能会通过其它包&…

java判断字符串是否和空字符串(““)相等、是否和空引用(null)相等,比较顺序不同导致出现死代码(Dead code)

我在用Java实现需求的时候&#xff0c;用到了字符串跟空字符串&#xff08;“”&#xff09;比较&#xff0c;跟空引用null比较&#xff0c;两个比较语句的顺序不同&#xff0c;一个顺序出现了死代码&#xff08;Dead code&#xff09;。 下面这个代码片段&#xff0c;字符串li…

【Rust 日报】2023-07-30 Lemmy v0.18.3发布

Lemmy v0.18.3发布 Lemmy 与 Reddit、Lobste.rs 或 Hacker News 等网站类似&#xff1a;你可以订阅你感兴趣的论坛&#xff0c;发布链接和讨论&#xff0c;然后进行投票或评论。但在幕后&#xff0c;Lemmy 和他们不同 —— 任何人都可以很容易地运行一个服务器&#xff0c;所有…

数字化转型才是王道!零售行业的关键五大指标都在这里了,快收藏

新零售是指结合线上和线下的销售模式&#xff0c;利用数字化技术和数据分析手段&#xff0c;通过线上平台吸引消费者&#xff0c;在线下店铺提供更加个性化和便捷的购物体验。 新零售不再将线上和线下视为两个独立的销售渠道&#xff0c;而是将其整合为一个完整的销售生态系统…

RabbitMQ 教程 | 第4章 RabbitMQ 进阶

&#x1f468;&#x1f3fb;‍&#x1f4bb; 热爱摄影的程序员 &#x1f468;&#x1f3fb;‍&#x1f3a8; 喜欢编码的设计师 &#x1f9d5;&#x1f3fb; 擅长设计的剪辑师 &#x1f9d1;&#x1f3fb;‍&#x1f3eb; 一位高冷无情的编码爱好者 大家好&#xff0c;我是 DevO…

Navicat for Mysql8.xx 字段注释中文乱码解决方

问题摘要&#xff1a; Navicat 中表描述中文均正常&#xff0c;只有字段注释中文乱码&#xff0c;原因是系统本身编码&#xff08;命令行chcp&#xff09;中为936&#xff08;GBK&#xff09;而不是65001&#xff08;UTF8)造成的 使用场景&#xff1a; Navicat for MySQL&am…

10分钟掌握seaborn绘制多子图

公众号&#xff1a;尤而小屋作者&#xff1a;Peter编辑&#xff1a;Peter 大家好&#xff0c;我是Peter~ 本文介绍如何使用seaborn绘制多子图 In [1]: import matplotlib.pyplot as plt %matplotlib inlineimport seaborn as sns import plotly_express as pximport warnin…

ELK高级搜索(一)

文章目录 ELK搜索1&#xff0e;简介1.1 内容1.2 面向 2&#xff0e;Elastic Stack2.1 简介2.2 特色2.3 组件介绍 3&#xff0e;Elasticsearch3.1 搜索是什么3.2 数据库搜索3.3 全文检索3.4 倒排索引3.5 Lucene3.6 Elasticsearch3.6.1 Elasticsearch的功能3.6.2 Elasticsearch使…

详解AMQP协议以及JAVA体系中的AMQP

目录 1.概述 1.1.简介 1.2.抽象模型 2.spring中的amqp 2.1.spring amqp 2.2.spring boot amqp 1.概述 1.1.简介 AMQP&#xff0c;Advanced Message Queuing Protocol&#xff0c;高级消息队列协议。 百度百科上的介绍&#xff1a; 一个提供统一消息服务的应用层标准高…

Vector - CAPL - 诊断模块函数(设置和获取)

目录 CanTpGetRxIdentifier CanTpGetTxIdentifier CanTpSetRxIdentifier CanTpSetTxIdentifier 代码示例 CanTpGetPadding & CanTpSetPadding 代码示例 CanTpGetAcceptOtherMode & CanTpSetAcceptOtherMode 代码示例 对于使用OSEK.dll文件调用发送诊断数据和接…

电子技术的发展

本资料仅用于学习和讨论如有侵权请反馈 1.1 第一台只能存储640M 什么是模拟信号: 2.1 把声波转化为电信号 2.2 微音器输出的某一段信号的波形 3、最简单有个蜂鸣器,再好一点有个喇叭 4、人说话的声音是个很复杂的声音 5、嵌入式实时操作系统: 6、结构到函数 6.1 学习环境…

【Java】快速入门JVM

文章目录 1. JVM简介2. 类加载简介3. 类加载的过程4. 双亲委派5. GC垃圾回收6. JVM的回收方式7. 分代回收 1. JVM简介 JVM&#xff08;Java虚拟机&#xff09;是一个名字为Java的进程,是用于执行Java程序的虚拟机。 JVM会从操作系统中申请一大块内存空间,又把这个内存空间划分…

我的会议(我的审批,会议签字附源码)

目录 前言&#xff1a; 3.我的审批&#xff1a; 3.1实现的特色功能&#xff1a; 3.2显示的效果 3.3思路&#xff1a; 3.4寻找相关的案例或者自己使用JavaScript去写一个类似的功能 3.5具体的步骤&#xff1a; 3.5.1添加静态的jsp代码&#xff08;我的审批数据的显示&…

投稿注意!APA格式超全示例详解,原本28天能录用,可能要拖延2个月

为什么同一本期刊有论文28天录用&#xff0c;有论文10个月才录用&#xff1f;结合近期征稿的这本经管类SSCI期刊&#xff0c;小编&#xff08;Unionpub学术&#xff09;整理了部分影响录用的几个因素&#xff0c;准备提交此期刊的作者可自查参考下&#xff1a; &#xff08;参…

【100天精通python】Day20:文件及目录操作_os模块和os.psth模块, 文件路径拼接,目录操作

目录 专栏导读 1 文件的目录操作 os模块的一些操作目录函数​编辑 os.path 模块的操作目录函数 2 相对路径和绝对路径 3 路径拼接 4 判断目录是否存在 5 创建目录、删除目录、遍历目录 专栏导读 专栏订阅地址&#xff1a;https://blog.csdn.net/qq_35831906/category_12…