百度360必应搜狗淘宝本站头条
当前位置:网站首页 > IT技术 > 正文

阻塞队列ArrayBlockingQueue的实现原理浅析

wptr33 2025-09-19 03:54 1 浏览

阻塞队列介绍

阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。这两个附加的操作是:在队列为空时,获取元素的线程会等待队列变为非空。当队列满时,存储元素的线程会等待队列可用。阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程。阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素。

阻塞队列例子

    public static void main(String[] args) {
        BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    int i = 0;
                    while(true) {
                        i++;
                        System.out.println("放入:"+i);
                        bq.put(i+""+new Date());
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }).start();
        new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    while(true) {
                        String i =bq.take();
                        System.out.println("取出:"+i);
                        Thread.sleep(2000);
                    }
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
            }
        }).start();
    }
}

阻塞队列原理

我们这里简要分析一下上面的程序是如何实现一个线程(生产者)放内容,等队列满了就暂停等待,直到队列有空位就继续放,另一个线程(消费者)取内容,队列为空就暂停等待,知道到队列有值就继续取。

1、初始化阻塞队列

BlockingQueue<String> bq = new ArrayBlockingQueue<String>(10);

我们这里是初始化长度为10的阻塞队列,点进去看看具体做了啥

public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
       throw new IllegalArgumentException();
    //1、初始化一个长度为capacity的数组,这里传的是10
    this.items = new Object[capacity];
    //2、初始化一个可重入锁
    lock = new ReentrantLock(fair);
    //3、初始化一个非空Condition
    notEmpty = lock.newCondition();
    //4、初始化一个非满Condition
    notFull =  lock.newCondition();
}

上面代码简单清晰易懂,第1步初始化一个长度为capacity的数组是指定这个阻塞队列的长度就是10,第2步是初始化一个可重入锁,这个是用于在队列中放数据的时候和取数据的时候做同步的,应为不可能说生产者和消费者同时放数据和取数据,会有并发问题,所以这里需要一个锁来同步,同步生产者和消费者在放数据和取数据的交互过程,生产者和消费者在跟队列放或者取完数据后就可以把锁释放去执行自己的业务逻辑了。

第3步和第4步是生成了两个Condition,那Condition是干啥的呢,这里简单说下,在使用Lock之前,我们都使用Object 的wait和notify实现同步的。如下代码

synchronize(obj){ 
    obj.wait();//等待
}
synchronize(obj){ 
    obj.notify();//唤醒 
}

有了lock后,现在是:

lock.lock(); 
condition.await(); //等待
lock.unlock();
lock.lock(); 
condition.signal(); //唤醒
lock.unlock();

相比Object更加的灵活。所以可以推测第3步和第4步就是用来在队列空或者满的时候进行阻塞和唤醒的,具体怎么用,我们得继续向下看源码。

2、生产者放数据

bq.put(i+""+new Date());

点进去查看源码

public void put(E e) throws InterruptedException {
        //1、检查元素是否是空的
        checkNotNull(e);
        //2、获取重入锁
        final ReentrantLock lock = this.lock;
        //3、锁住
        lock.lockInterruptibly();
        try {
            //4、如果队列满了
            while (count == items.length)
                //await阻塞当前线程,释放锁
                notFull.await();
            //5、如果没有满,则加入到队列中
            enqueue(e);
        } finally {
            //不管如何,最后都需要进行锁的释放
            lock.unlock();
        }
    }

逻辑还是特别清晰的,关键点就在于如果队列满了,就直接阻塞线程,也就是阻塞了生产者。

 notFull.await();

所以初始化的时候实例化的notFull的作用就是给生产者用的,可以推测,在消费者取走数据后,肯定会调用唤醒的方法。

notFull.signal();

也可以推测出,如果这里不满,那么执行完 enqueue(e)加入数据到队列中后,肯定会调用消费者线程的唤醒方法。

notEmpty.signal();

我们进入enqueue(e)方法的源码看看是否如推测一样。

private void enqueue(E x) {
       // assert lock.getHoldCount() == 1;
       // assert items[putIndex] == null;
       //获取队列
       final Object[] items = this.items;
       //新加入一个值
       items[putIndex] = x;
       //如果已经加入到最后了,那么下一次从第一位加入,否则在下一个位置加入
       if (++putIndex == items.length)
           putIndex = 0;
       //队列中的数目自增
       count++;
       //唤醒消费者
       notEmpty.signal();
    }

逻辑也很简单,不出所料,最后因为在队列中加入了数据,调用了消费在的Condition进行唤醒。

notEmpty.signal();

接下来我们再看看消费者怎么消费数据的,如果队列空了会怎么阻塞,其实都可以猜猜到,肯定是先获取锁,然后判断队列是否为空,若为空则阻塞,不为空则从队列中取树,后唤醒生产者,如果生产者本来就没有被阻塞也就不用唤醒。

3、消费者消费数据

String i =bq.take();

我们点进源码去看看

 public E take() throws InterruptedException {
        //1、获取锁
        final ReentrantLock lock = this.lock;
        //2、锁住
        lock.lockInterruptibly();
        try {
            //3、如果队列为空,则阻塞消费者
            while (count == 0)
                //4、阻塞,这里会释放锁
                notEmpty.await();
            //5、如果不为空,这里会取数据
            return dequeue();
        } finally {
            //7、释放锁
            lock.unlock();
        }
    }

逻辑也很简单,反正一开始只要生成则没有在放数据的过程中,或者已经阻塞,则消费者在第2步骤都可以获得锁进行操作,我们直接看如果不为空dequeue()如何取数据,可以推测,里面取完数据后会进行唤醒生产者的操作,点进去看看。

 private E dequeue() {
        // assert lock.getHoldCount() == 1;
        // assert items[takeIndex] != null;
        //获取当前数组
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        //取的数据
        E x = (E) items[takeIndex];
        items[takeIndex] = null;
        //下一次就取下一个,但是如果已经到最后了,下一次就取第一个
        if (++takeIndex == items.length)
            takeIndex = 0;
        //队列数据减一
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        //唤醒生产者
        notFull.signal();
        return x;
    }

源码也很简单,逻辑清楚,最后也不出所料调用了唤醒生产者的方法。

notFull.signal();

总结

这大概是我见过JUC中最清晰易懂的源码了。

相关推荐

高性能并发队列Disruptor使用详解

基本概念Disruptor是一个高性能的异步处理框架,是一个轻量的Java消息服务JMS,能够在无锁的情况下实现队列的并发操作Disruptor使用环形数组实现了类似队列的功能,并且是一个有界队列....

Disruptor一个高性能队列_java高性能队列

Disruptor一个高性能队列前言说到队列比较熟悉的可能是ArrayBlockingQueue、LinkedBlockingQueue这两个有界队列,大多应用在线程池中使用能保证线程安全,但其安全性...

谈谈防御性编程_防御性策略

防御性编程对于程序员来说是一种良好的代码习惯,是为了保护自己的程序在不可未知的异常下,避免带来更大的破坏性崩溃,使得程序在错误发生时,依然能够云淡风轻的处理,但很多程序员入行很多年,写出的代码依然都是...

有人敲门,开水开了,电话响了,孩子哭了,你先顾谁?

前言哎呀,这种情况你肯定遇到过吧!正在家里忙活着,突然——咚咚咚有人敲门,咕噜咕噜开水开了,铃铃铃电话响了,哇哇哇孩子又哭了...我去,四件事一起来,人都懵了!你说先搞哪个?其实这跟我们写Java多线...

面试官:线程池如何按照core、max、queue的执行顺序去执行?

前言这是一个真实的面试题。前几天一个朋友在群里分享了他刚刚面试候选者时问的问题:"线程池如何按照core、max、queue的执行循序去执行?"。我们都知道线程池中代码执行顺序是:co...

深入剖析 Java 中线程池的多种实现方式

在当今高度并发的互联网软件开发领域,高效地管理和利用线程资源是提升程序性能的关键。Java作为一种广泛应用于后端开发的编程语言,为我们提供了丰富的线程池实现方式。今天,就让我们深入探讨Java中...

并发编程之《彻底搞懂Java线程》_java多线程并发解决方案详解

目录引言一、核心概念:线程是什么?...

Redis怎么实现延时消息_redis实现延时任务

一句话总结Redis可通过有序集合(ZSET)实现延时消息:将消息作为value,到期时间戳作为score存入ZSET。消费者轮询用ZRANGEBYSCORE获取到期消息,配合Lua脚本保证原子性获取...

CompletableFuture真的用对了吗?盘点它最容易被误用的5个场景

在Java并发编程中,CompletableFuture是处理异步任务的利器,但不少开发者在使用时踩过这些坑——线上服务突然雪崩、异常悄无声息消失、接口响应时间翻倍……本文结合真实案例,拆解5个最容易...

接口性能优化技巧,有点硬_接口性能瓶颈

背景我负责的系统到2021年初完成了功能上的建设,开始进入到推广阶段。随着推广的逐步深入,收到了很多好评的同时也收到了很多对性能的吐槽。刚刚收到吐槽的时候,我们的心情是这样的:...

禁止使用这5个Java类,每一个背后都有一段&quot;血泪史&quot;

某电商平台的支付系统突然报警:大量订单状态异常。排查日志发现,同一笔订单被重复支付了三次。事后复盘显示,罪魁祸首竟是一行看似无害的SimpleDateFormat代码。在Java开发中,这类因使用不安...

无锁队列Disruptor原理解析_无锁队列实现原理

队列比较队列...

Java并发队列与容器_java 并发队列

【前言:无论是大数据从业人员还是Java从业人员,掌握Java高并发和多线程是必备技能之一。本文主要阐述Java并发包下的阻塞队列和并发容器,其实研读过大数据相关技术如Spark、Storm等源码的,...

线程池工具及拒绝策略的使用_线程池处理策略

线程池的拒绝策略若线程池中的核心线程数被用完且阻塞队列已排满,则此时线程池的资源已耗尽,线程池将没有足够的线程资源执行新的任务。为了保证操作系统的安全,线程池将通过拒绝策略处理新添加的线程任务。...

【面试题精讲】ArrayBlockingQueue 和 LinkedBlockingQueue 区别?

有的时候博客内容会有变动,首发博客是最新的,其他博客地址可能会未同步,认准...