BlockingQueue(阻塞队列)详解

news2025/1/11 21:58:07

一. 前言

  在新增的Concurrent包中,BlockingQueue很好的解决了多线程中,如何高效安全“传输”数据的问题。通过这些高效并且线程安全的队列类,为我们快速搭建高质量的多线程程序带来极大的便利。本文详细介绍了BlockingQueue家庭中的所有成员,包括他们各自的功能以及常见使用场景。

二. 认识BlockingQueue

  阻塞队列,顾名思义,首先它是一个队列,而一个队列在数据结构中所起的作用大致如下图所示:


  从上图我们可以很清楚看到,通过一个共享的队列,可以使得数据由队列的一端输入,从另外一端输出;

  常用的队列主要有以下两种:(当然通过不同的实现方式,还可以延伸出很多不同类型的队列,DelayQueue就是其中的一种)

    先进先出(FIFO):先插入的队列的元素也最先出队列,类似于排队的功能。从某种程度上来说这种队列也体现了一种公平性。

    后进先出(LIFO):后插入队列的元素最先出队列,这种队列优先处理最近发生的事件。  


      多线程环境中,通过队列可以很容易实现数据共享,比如经典的“生产者”和“消费者”模型中,通过队列可以很便利地实现两者之间的数据共享。假设我们有若干生产者线程,另外又有若干个消费者线程。如果生产者线程需要把准备好的数据共享给消费者线程,利用队列的方式来传递数据,就可以很方便地解决他们之间的数据共享问题。但如果生产者和消费者在某个时间段内,万一发生数据处理速度不匹配的情况呢?理想情况下,如果生产者产出数据的速度大于消费者消费的速度,并且当生产出来的数据累积到一定程度的时候,那么生产者必须暂停等待一下(阻塞生产者线程),以便等待消费者线程把累积的数据处理完毕,反之亦然。然而,在concurrent包发布以前,在多线程环境下,我们每个程序员都必须去自己控制这些细节,尤其还要兼顾效率和线程安全,而这会给我们的程序带来不小的复杂度。好在此时,强大的concurrent包横空出世了,而他也给我们带来了强大的BlockingQueue。(在多线程领域:所谓阻塞,在某些情况下会挂起线程(即阻塞),一旦条件满足,被挂起的线程又会自动被唤醒),下面两幅图演示了BlockingQueue的两个常见阻塞场景:

       如上图所示:当队列中没有数据的情况下,消费者端的所有线程都会被自动阻塞(挂起),直到有数据放入队列。


   如上图所示:当队列中填满数据的情况下,生产者端的所有线程都会被自动阻塞(挂起),直到队列中有空的位置,线程被自动唤醒。

  这也是我们在多线程环境下,为什么需要BlockingQueue的原因。作为BlockingQueue的使用者,我们再也不需要关心什么时候需要阻塞线程,什么时候需要唤醒线程,因为这一切BlockingQueue都给你一手包办了。既然BlockingQueue如此神通广大,让我们一起来见识下它的常用方法:

三. BlockingQueue的核心方法

  1.放入数据

    (1)offer(anObject):表示如果可能的话,将anObject加到BlockingQueue里,即如果BlockingQueue可以容纳,则返回true,否则返回false.(本方法不阻塞当前执行方法

 的线程);       
       (2)offer(E o, long timeout, TimeUnit unit):可以设定等待的时间,如果在指定的时间内,还不能往队列中加入BlockingQueue,则返回失败。

    (3)put(anObject):把anObject加到BlockingQueue里,如果BlockQueue没有空间,则调用此方法的线程被阻断直到BlockingQueue里面有空间再继续.

  2. 获取数据

    (1)poll(time):取走BlockingQueue里排在首位的对象,若不能立即取出,则可以等time参数规定的时间,取不到时返回null;

    (2)poll(long timeout, TimeUnit unit):从BlockingQueue取出一个队首的对象,如果在指定时间内,队列一旦有数据可取,则立即返回队列中的数据。否则知道时间

超时还没有数据可取,返回失败。

    (3)take():取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入; 

    (4)drainTo():一次性从BlockingQueue获取所有可用的数据对象(还可以指定获取数据的个数),通过该方法,可以提升获取数据效率;不需要多次分批加锁或释放锁。

四. 常见BlockingQueue

  在了解了BlockingQueue的基本功能后,让我们来看看BlockingQueue家庭大致有哪些成员?

  1. ArrayBlockingQueue

  基于数组的阻塞队列实现,在ArrayBlockingQueue内部,维护了一个定长数组,以便缓存队列中的数据对象,这是一个常用的阻塞队列,除了一个定长数组外,ArrayBlockingQueue内部还保存着两个整形变量,分别标识着队列的头部和尾部在数组中的位置。

  ArrayBlockingQueue在生产者放入数据和消费者获取数据,都是共用同一个锁对象,由此也意味着两者无法真正并行运行,这点尤其不同于LinkedBlockingQueue;按照实现原理来分析,ArrayBlockingQueue完全可以采用分离锁,从而实现生产者和消费者操作的完全并行运行。Doug Lea之所以没这样去做,也许是因为ArrayBlockingQueue的数据写入和获取操作已经足够轻巧,以至于引入独立的锁机制,除了给代码带来额外的复杂性外,其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue间还有一个明显的不同之处在于,前者在插入或删除元素时不会产生或销毁任何额外的对象实例,而后者则会生成一个额外的Node对象。这在长时间内需要高效并发地处理大批量数据的系统中,其对于GC的影响还是存在一定的区别。而在创建ArrayBlockingQueue时,我们还可以控制对象的内部锁是否采用公平锁,默认采用非公平锁。

  2.LinkedBlockingQueue

  基于链表的阻塞队列,同ArrayListBlockingQueue类似,其内部也维持着一个数据缓冲队列(该队列由一个链表构成),当生产者往队列中放入一个数据时,队列会从生产者手中获取数据,并缓存在队列内部,而生产者立即返回;只有当队列缓冲区达到最大值缓存容量时(LinkedBlockingQueue可以通过构造函数指定该值),才会阻塞生产者队列,直到消费者从队列中消费掉一份数据,生产者线程会被唤醒,反之对于消费者这端的处理也基于同样的原理。而LinkedBlockingQueue之所以能够高效的处理并发数据,还因为其对于生产者端和消费者端分别采用了独立的锁来控制数据同步,这也意味着在高并发的情况下生产者和消费者可以并行地操作队列中的数据,以此来提高整个队列的并发性能。

  作为开发者,我们需要注意的是,如果构造一个LinkedBlockingQueue对象,而没有指定其容量大小,LinkedBlockingQueue会默认一个类似无限大小的容量(Integer.MAX_VALUE),这样的话,如果生产者的速度一旦大于消费者的速度,也许还没有等到队列满阻塞产生,系统内存就有可能已被消耗殆尽了。

  ArrayBlockingQueue和LinkedBlockingQueue是两个最普通也是最常用的阻塞队列,一般情况下,在处理多线程间的生产者消费者问题,使用这两个类足以。

  下面的代码演示了如何使用BlockingQueue:

  (1) 测试类

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

public class BlockingQueueTest {

    public static void main(String[] args) throws InterruptedException {
        // 声明一个容量为10的缓存队列
        BlockingQueue<String> queue = new LinkedBlockingQueue<String>(10);

        //new了三个生产者和一个消费者
        Producer producer1 = new Producer(queue);
        Producer producer2 = new Producer(queue);
        Producer producer3 = new Producer(queue);
        Consumer consumer = new Consumer(queue);

        // 借助Executors
        ExecutorService service = Executors.newCachedThreadPool();
        // 启动线程
        service.execute(producer1);
        service.execute(producer2);
        service.execute(producer3);
        service.execute(consumer);

        // 执行10s
        Thread.sleep(10 * 1000);
        producer1.stop();
        producer2.stop();
        producer3.stop();

        Thread.sleep(2000);
        // 退出Executor
        service.shutdown();
    }
}

     (2)生产者类

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者线程
 *
 * @author jackyuj
 */
public class Producer implements Runnable {

    private volatile boolean  isRunning = true;//是否在运行标志
    private BlockingQueue queue;//阻塞队列
    private static AtomicInteger count = new AtomicInteger();//自动更新的值
    private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;

    //构造函数
    public Producer(BlockingQueue queue) {
        this.queue = queue;
    }

    public void run() {
        String data = null;
        Random r = new Random();

        System.out.println("启动生产者线程!");
        try {
            while (isRunning) {
                System.out.println("正在生产数据...");
                Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));//取0~DEFAULT_RANGE_FOR_SLEEP值的一个随机数

                data = "data:" + count.incrementAndGet();//以原子方式将count当前值加1
                System.out.println("将数据:" + data + "放入队列...");
                if (!queue.offer(data, 2, TimeUnit.SECONDS)) {//设定的等待时间为2s,如果超过2s还没加进去返回true
                    System.out.println("放入数据失败:" + data);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            System.out.println("退出生产者线程!");
        }
    }

    public void stop() {
        isRunning = false;
    }
}

        (3)消费者类

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * 消费者线程
 *
 * @author jackyuj
 */
public class Consumer implements Runnable {

    private BlockingQueue<String> queue;
    private static final int DEFAULT_RANGE_FOR_SLEEP = 1000;

    //构造函数
    public Consumer(BlockingQueue<String> queue) {
        this.queue = queue;
    }

    public void run() {
        System.out.println("启动消费者线程!");
        Random r = new Random();
        boolean isRunning = true;
        try {
            while (isRunning) {
                System.out.println("正从队列获取数据...");
                String data = queue.poll(2, TimeUnit.SECONDS);//有数据时直接从队列的队首取走,无数据时阻塞,在2s内有数据,取走,超过2s还没数据,返回失败
                if (null != data) {
                    System.out.println("拿到数据:" + data);
                    System.out.println("正在消费数据:" + data);
                    Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));
                } else {
                    // 超过2s还没数据,认为所有生产线程都已经退出,自动退出消费线程。
                    isRunning = false;
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
            Thread.currentThread().interrupt();
        } finally {
            System.out.println("退出消费者线程!");
        }
    }


}

3. DelayQueue

  DelayQueue中的元素只有当其指定的延迟时间到了,才能够从队列中获取到该元素。DelayQueue是一个没有大小限制的队列,因此往队列中插入数据的操作(生产者)永远不会被阻塞,而只有获取数据的操作(消费者)才会被阻塞。

  使用场景:

  DelayQueue使用场景较少,但都相当巧妙,常见的例子比如使用一个DelayQueue来管理一个超时未响应的连接队列。

  4. PriorityBlockingQueue

   基于优先级的阻塞队列(优先级的判断通过构造函数传入的Compator对象来决定),但需要注意的是PriorityBlockingQueue并不会阻塞数据生产者,而只会在没有可消费的数据时,阻塞数据的消费者。因此使用的时候要特别注意,生产者生产数据的速度绝对不能快于消费者消费数据的速度,否则时间一长,会最终耗尽所有的可用堆内存空间。在实现PriorityBlockingQueue时,内部控制线程同步的锁采用的是公平锁。

  5. SynchronousQueue

   一种无缓冲的等待队列,类似于无中介的直接交易,有点像原始社会中的生产者和消费者,生产者拿着产品去集市销售给产品的最终消费者,而消费者必须亲自去集市找到所要商品的直接生产者,如果一方没有找到合适的目标,那么对不起,大家都在集市等待。相对于有缓冲的BlockingQueue来说,少了一个中间经销商的环节(缓冲区),如果有经销商,生产者直接把产品批发给经销商,而无需在意经销商最终会将这些产品卖给那些消费者,由于经销商可以库存一部分商品,因此相对于直接交易模式,总体来说采用中间经销商的模式会吞吐量高一些(可以批量买卖);但另一方面,又因为经销商的引入,使得产品从生产者到消费者中间增加了额外的交易环节,单个产品的及时响应性能可能会降低。

  声明一个SynchronousQueue有两种不同的方式,它们之间有着不太一样的行为。公平模式和非公平模式的区别:

  如果采用公平模式:SynchronousQueue会采用公平锁,并配合一个FIFO队列来阻塞多余的生产者和消费者,从而体系整体的公平策略;

  但如果是非公平模式(SynchronousQueue默认):SynchronousQueue采用非公平锁,同时配合一个LIFO队列来管理多余的生产者和消费者,而后一种模式,如果生产者和消费者的处理速度有差距,则很容易出现饥渴的情况,即可能有某些生产者或者是消费者的数据永远都得不到处理。

五. 小结

  BlockingQueue不光实现了一个完整队列所具有的基本功能,同时在多线程环境下,他还自动管理了多线间的自动等待于唤醒功能,从而使得程序员可以忽略这些细节,关注更高级的功能。

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

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

相关文章

上海亚商投顾:A股三大指数震荡涨跌各异 大消费全天活跃

上海亚商投顾前言&#xff1a;无惧大盘涨跌&#xff0c;解密龙虎榜资金&#xff0c;跟踪一线游资和机构资金动向&#xff0c;识别短期热点和强势个股。市场情绪沪指今日窄幅震荡&#xff0c;创业板指小幅冲高后回落&#xff0c;科创50指数盘中涨近1.5%&#xff0c;随后涨幅明显…

gdb与gdbserver的使用

GDB调试示例以调试可执行程序gdbDebug为例&#xff0c;gdbDebug.cpp内容如下&#xff1a;使用gdb 启动gdbDebug程序左侧为gdb调试&#xff0c;右侧为gdbDebug.cpp内容GDB与GDBServer调试示例以调试可执行程序gdbDebug为例&#xff0c;gdbDebug.cpp内容如下&#xff1a;使用gdbs…

StarRocks荣获2022年度最具潜力数据库奖

近日&#xff0c;墨天轮颁布了《2022年度数据库获奖名单》&#xff0c;通过墨天轮排行榜排名及年度得分、生态建设、市场活动、市场份额、专利数等38个综合指标进行遴选&#xff0c;评选出2022年的数据库重要奖项&#xff0c;期望能够通过多维度评选&#xff0c;呈现出数据库产…

客户在国内,挑选海外服务器供应商有什么技巧?

​  一直以来&#xff0c;基于互联网管理的严格要求&#xff0c;在使用中国大陆服务器放置网站时&#xff0c;是需要备案手续的&#xff0c;这个手续时长快则7天&#xff0c;慢则也有接近1个月的情况&#xff0c;复杂耗时&#xff0c;当然&#xff0c;这也是对建站成本的增加…

程序员想兼职赚钱?这几个渠道你一定要知道?

某一天当一个程序员&#xff0c;一拍脑门想要兼职&#xff0c;赚点小钱&#xff0c;于是他打开了知乎&#xff0c;打开了百度搜索兼职。结果弹出了一大部分有兼职要视频剪辑的&#xff0c;写文稿的等等等等。逛了一圈&#xff0c;发现根本没有自己合适的兼职。 我想说&#xff…

0201 设置/修改元素内容和属性

document.write()方法文本内容追加到</body>前面的位置文本中标签会被解析<script>document.write(hello world)document.write(<h3>你好世界</h3>)</script>innerText属性将文本内容添加更新到任意标签位置文本包含的标签不会被解析<style&g…

JavaScript 类与类型判断

JavaScript中的数据类型 JavaScript的数据类型大致分为两种&#xff1a;原始类型、对象类型。 原始类型&#xff1a;数字Number、字符串String、布尔值boolean、以及两个特殊值(null、undefined). 对象类型&#xff1a;数组Array、函数Function、日期Date、正则RegExp、错误Err…

Small RTOS51 学习笔记(10)Small RTOS51 的移植

个人笔记 文章目录准备一个 51 单片机工程将 Small RTOS51 相关文件添加到工程一个简单的程序运行效果遇到的问题准备一个 51 单片机工程 我打算拿一个现成的 51 单片机工程来移植 Small RTOS51&#xff0c; 当然&#xff0c;也可以重新创建一个新的工程。 将 Small RTOS51 相…

记录一次sql group by 优化记录

最近有个手动任务&#xff0c;需要计算每天的数据量&#xff0c;然后再进行处理。根据这种情况计算&#xff0c;sql是这样的SELECT FROM_UNIXTIME(publish_time / 1000, %Y-%m-%d) date,COUNT(*) as countFROMinfo_article_mainWHEREpublish_time BETWEEN ?AND ?GROUP BY dat…

Windows实时运动控制软核(六):LOCAL高速接口测试之Matlab

今天&#xff0c;正运动小助手给大家分享一下MotionRT7的安装和使用&#xff0c;以及使用Matlab对MotionRT7开发的前期准备。 01 MotionRT7简介 MotionRT7是深圳市正运动技术推出的跨平台运动控制实时内核&#xff0c;也是国内首家完全自主自研&#xff0c;自主可控的Windows…

flutter 中stack 控件的 大小是如何确定的

stack 控件 stack 是我们在flutter中常用到的控件&#xff0c;然而stack的大小是如何确定的是一个值得探究的问题&#xff0c;自己在网上也进行了搜索&#xff0c;但是总是不能解释自己遇到的新情况&#xff0c;所以我这里就根据目前的经验对stack大小是如何确定的进行一下总结…

【Java基础】006 -- 运算符

目录 一、算数运算符 1、基本用法 2、高级用法 ①、数字相加 ②、字符串相加 ③、字符相加 二、自增自减运算符 1、基本用法 三、赋值运算符 四、关系运算符 五、逻辑运算符 1、四种逻辑运算符 2、短路逻辑运算符 六、三元运算符 1、什么是三元运算符 2、三元运算符格式 七、运…

规则引擎-drools-3.3-drl文件构成-rule部分-条件Condition

文章目录drl文件构成-rule部分条件部分 LHS模式&#xff08;Pattern&#xff09;、绑定变量属性约束DRL中支持的规则条件元素&#xff08;关键字&#xff09;运算符比较操作符条件元素继承条件元素do对应多then条件drl文件构成-rule部分 drl文件构成&#xff0c;位于官网的第5…

工程师是怎样对待开源

工程师如何对待开源 本文是笔者作为一个在知名科技企业内从事开源相关工作超过 20 年的工程师&#xff0c;亲身经历或者亲眼目睹很多工程师对待开源软件的优秀实践&#xff0c;也看到了很多 Bad Cases&#xff0c;所以想把自己的一些心得体会写在这里&#xff0c;供工程师进行…

递归、dfs、回溯、剪枝,一针见血的

一、框架&#xff1a;回溯搜索的遍历过程&#xff1a;回溯法⼀般是在集合中递归搜索&#xff0c;集合的⼤⼩构成了树的宽度&#xff0c;递归的深度构成的树的深度。for循环就是遍历集合区间&#xff0c;可以理解⼀个节点有多少个孩⼦&#xff0c;这个for循环就执⾏多少次。back…

那些提升工作效率的Windows常用快捷键

那些提升工作效率的Windows常用快捷键 前言 在我们日常工作中&#xff0c;掌握一些常用的电脑快捷键&#xff0c;可以让办公效率事半功倍&#xff0c;熟用快捷键可以极大增加我们的工作效率&#xff0c;更重要的是键盘操作看起来更让人赏心悦目&#xff01; 我们通常将快捷键…

【C++】作用域与函数重载

【C】作用域与函数重载 1、作用域 1.1 作用域的作用 作用域——scope 通常来说&#xff0c;一段程序代码中所用到的名字并不总是有效/可用的&#xff0c;而限定这个名字的可用性的代码范围就是这个名字的作用域。 简单来说&#xff0c;作用域的使用减少了代码中名字的重复冲…

13、稀疏矩阵

目录 一、稀疏矩阵的生成 1.利用sparse函数建立一般的稀疏矩阵 2.利用特定函数建立稀疏矩阵 二、稀疏矩阵的运算 一、稀疏矩阵的生成 1.利用sparse函数建立一般的稀疏矩阵 稀疏矩阵指令的调用格式&#xff1a; 示例1&#xff1a;输入一个稀疏矩阵 Asparse([1 2 3 4 5],[…

教你一招完美解决 pptx 库安装失败的问题

上一篇&#xff1a;Python的序列结构及常用操作方法&#xff0c;学完这一篇你就彻底懂了 文章目录前言一、pptx 库是什么&#xff1f;二、安装失败原因及解决方案总结前言 昨天有粉丝问我&#xff0c;为什么Python的 pptx 库老是安装失败&#xff1f;加上国内镜像源也不行&…

分布式微服务2

目录 Nacos注册中心 下载 启动 快速入门 1.在父工程中添加spring-cloud-alilbaba的管理依赖子模块添加nacos的客户端依赖 2.子模块添加nacos的客户端依赖 3.子模块配置文件 4.启动 Nacos服务分级存储模型 集群配置 nacos的负载均衡 Nacos环境隔离 新建命名空间 N…