ArrayBlockingQueue源码分析总结

2024-06-02 17:48

本文主要是介绍ArrayBlockingQueue源码分析总结,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

 

1,在说ArrayBlockingQueue之前,我们先说一下ReentrantLock:

   ReentrantLock和Synchronized的作用差不多, java在编写多线程程序时,都是为了保证线程安全,需要对数据同步
       1)相似点:
            这两种同步方式有很多相似之处,它们都是加锁方式同步,而且都是阻塞式的同步,也就是说当如果一个线程获得了对象锁,进入了同步块,其他访问该同步块的线程都必须阻塞在同步块外面等待,而进行线程阻塞和唤醒的代价是比较高的(操作系统需要在用户态与内核态之间来回切换,代价很高,不过可以通过对锁优化进行改善)。
       2)区别:
            这两种方式最大区别就是对于Synchronized来说,它是java语言的关键字,是原生语法层面的互斥,需要jvm实现。而ReentrantLock它是JDK 1.5之后提供的API层面的互斥锁,需要lock()和unlock()方法配合try/finally语句块来完成。
       3)至于Condition这个对象,当我们阅读到源码的时候,有的人可能很陌生,其实不要太对它纠结,它其实是封装了和锁相关的方法(wait(),notify(),notifyAll()),方便我们对某线程状态进行变更。

2,什么是阻塞队列

   答:阻塞队列其实就是生产者-消费者模型中的容器。当生产者往队列中添加元素时,如果队列已经满了,生产者所在的线程就会阻塞,直到消费者取元素时 notify 它;消费者去队列中取元素时,如果队列中是空的,消费者所在的线程就会阻塞,直到生产者放入元素 notify 它。具体到 Java 中,使用 BlockingQueue 接口表示阻塞队列:

3,实现原理

      ArrayBlockingQueue内部维护看一个数组,其内的方法别加了ReentrantLock锁来保证线程安全。

ArrayBlockingQueue是一个阻塞式的队列,继承自AbstractBlockingQueue,间接的实现了Queue接口和Collection接口。底层以数组的形式保存数据(实际上可看作一个循环数组)。常用的操作包括 add ,offer,put,remove,poll,take,peek。

public interface BlockingQueue<E> extends Queue<E> {
    //添加失败时会抛出异常,比较粗暴
    boolean add(E e);
    //添加失败时会返回 false,比较温婉,比 add 强
    boolean offer(E e);
    //添加元素时,如果没有空间,会阻塞等待;可以响应中断
    void put(E e) throws InterruptedException;
    //添加元素到队列中,如果没有空间会等待参数中的时间,超时返回,会响应中断
    boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException;
    //获取并移除队首元素,如果没有元素就会阻塞等待
    E take() throws InterruptedException;
    //获取并移除队首元素,如果没有就会阻塞等待参数的时间,超时返回
    E poll(long timeout, TimeUnit unit) throws InterruptedException;
    //返回队列中剩余的空间
    int remainingCapacity();
    //移除队列中某个元素,如果存在的话返回 true,否则返回 false
    boolean remove(Object o);
    //检查队列中是否包含某个元素,至少包含一个就返回 true
    public boolean contains(Object o);
    //将当前队列所有元素移动到给定的集合中,这个方法比反复地获取元素更高效
    //返回移动的元素个数
    int drainTo(Collection<? super E> c);
    //移动队列中至多 maxElements 个元素到指定的集合中
    int drainTo(Collection<? super E> c, int maxElements);
}
可以看到,在队列操作(添加/获取)当前不可用时,BlockingQueue 的方法有四种处理方式:
抛出异常  
对应的是 add(), remove(), element()
返回某个值(null 或者 false)  
offer(), poll(), peek()
阻塞当前线程,直到操作可以进行 
put(), take()
阻塞一段时间,超时后退出  
offer, poll()
BlockingQueue 中不允许有 null 元素,因此在 add(), offer(), put() 时如果参数是 null,会抛出空指针。null 是用来有异常情况时做返回值的。
七种阻塞队列的前三种
Java 中提供了 7 种 BlockingQueue 的实现,在看线程池之前我根本搞不清楚究竟选择哪个,直到完整地对比总结以后,发现其实也没什么复杂。
现在我们一起来看一下这 7 种实现。
1.ArrayBlockingQueue
ArrayBlockingQueue 是一个使用数组实现的、有界的队列,一旦创建后,容量不可变。队列中的元素按 FIFO 的顺序,每次取元素从头部取,加元素加到尾部。
默认情况下 ArrayBlockingQueue 不保证线程公平的访问队列,即在队列可用时,阻塞的线程都可以争夺访问队列的资格。
 不保证公平性有助于提高吞吐量。 
看它的主要属性:
public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    //使用数组保存的元素
    final Object[] items;
    //下一次取元素的索引
    int takeIndex;
    //下一次添加元素的索引
    int putIndex;
    //当前队列中元素的个数
    int count;
    /*
     * Concurrency control uses the classic two-condition algorithm
     * found in any textbook.
     */
    //全部操作的锁
    final ReentrantLock lock;
    //等待获取元素的锁
    private final Condition notEmpty;
    //等待添加元素的锁
    private final Condition notFull;
    //...
}
可以看到,ArrayBlockingQueue 使用可重入锁 ReentrantLock 实现的访问公平性,两个 Condition 保证了添加和获取元素的并发控制。
构造函数:
//指定队列的容量,使用非公平锁
public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
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();
}
//允许使用一个 Collection 来作为队列的默认元素
public ArrayBlockingQueue(int capacity, boolean fair,
                          Collection<? extends E> c) {
    this(capacity, fair);
    final ReentrantLock lock = this.lock;
    lock.lock(); // Lock only for visibility, not mutual exclusion
    try {
        int i = 0;
        try {
            for (E e : c) {    //遍历添加指定集合的元素
                if (e == null) throw new NullPointerException();
                items[i++] = e;
            }
        } catch (ArrayIndexOutOfBoundsException ex) {
            throw new IllegalArgumentException();
        }
        count = i;
        putIndex = (i == capacity) ? 0 : i;    //修改 putIndex 为 c 的容量 +1
    } finally {
        lock.unlock();
    }
}
可以看到,有三种构造函数:
默认的构造函数只指定了队列的容量,设置为非公平的线程访问策略
第二种构造函数中,使用 ReentrantLock 创建了 2 个 Condition 锁
第三种构造函数可以在创建队列时,将指定的元素添加到队列中
四种添加元素方法的实现
第一种 add():
public boolean add(E e) {
    return super.add(e);
}
//super.add() 的实现
public boolean add(E e) {
    if (offer(e))
        return true;
    else
        throw new IllegalStateException("Queue full");
}
add(E) 调用了父类的方法,而父类里调用的是 offer(E),如果返回 false 就泡出异常。
第二种 offer():
public boolean offer(E e) {
    if (e == null) throw new NullPointerException();
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        if (count == items.length)
            return false;
        else {
            enqueue(e);
            return true;
        }
    } finally {
        lock.unlock();
    }
}
可以看到 offer(E) 方法先拿到锁,如果当前队列中元素已满,就立即返回 false,这点比 add() 友好一些; 
 如果没满就调用 enqueue(E) 入队:
private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length) putIndex = 0;
    count++;
    notEmpty.signal();
}
可以看到,enqueue(E) 方法会将元素添加到数组队列尾部。 
 如果添加元素后队列满了,就修改 putIndex 为 0 ,0.0 为啥这样,先留着回头看。 
 添加后调用 notEmpty.signal() 通知唤醒阻塞在获取元素的线程。
第三种 put():
public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}
可以看到,put() 方法可以响应中断,当队列满了,就调用 notFull.await() 阻塞等待,等有消费者获取元素后继续执行; 
 可以添加时还是调用 enqueue(E)。
第四种 offer(E,long,TimeUnit):
public boolean offer(E e, long timeout, TimeUnit unit)
    throws InterruptedException {
    if (e == null) throw new NullPointerException();
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length) {
            if (nanos <= 0)
                return false;
            nanos = notFull.awaitNanos(nanos);
        }
        enqueue(e);
        return true;
    } finally {
        lock.unlock();
    }
}
可以看到 offer() 和 put() 方法很相似,不同之处在于允许设置等待超时时间,超过这么久如果还不能有位置,就返回 false;否则调用 enqueue(E),然后返回 true。
总体来看添加元素很简单嘛 
四种获取元素的实现:
第一种 poll():
public E poll() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return (count == 0) ? null : dequeue();
    } finally {
        lock.unlock();
    }
}
poll() 如果在队列中没有元素时会立即返回 null;如果有元素调用 dequeue():
private E dequeue() {
    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;
}
默认情况下 dequeue() 方法会从队首移除元素(即 takeIndex 位置)。 

 移除后会向后移动 takeIndex,如果已经到队尾,就归零。结合前面添加元素时的归零,可以看到,其实 ArrayBlockingQueue 是个环形数组。

然后调用 itrs. elementDequeued(),这个 itrs 是 ArrayBlockingQueue 的内部类 Itrs 的对象,看起来像是个迭代器,实际上它的作用是保证循环数组迭代时的正确性,具体实现比较复杂,这里暂不介绍。
第二种 take():
public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}
take() 方法可以响应中断,与 poll() 不同的是,如果队列中没有数据会一直阻塞等待,直到中断或者有元素,有元素时还是调用 dequeue() 方法。
第三种 带参数的 poll():
public E poll(long timeout, TimeUnit unit) throws InterruptedException {
    long nanos = unit.toNanos(timeout);
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0) {
            if (nanos <= 0)
                return null;
            nanos = notEmpty.awaitNanos(nanos);
        }
        return dequeue();
    } finally {
        lock.unlock();
    }
}
带参数的 poll() 方法相当于无参 poll() 和 take() 的中和版,允许阻塞一段时间,如果在阻塞一段时间还没有元素进来,就返回 null。
第四种 peek():
public E peek() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return itemAt(takeIndex); // null when queue is empty
    } finally {
        lock.unlock();
    }
}
final E itemAt(int i) {
    return (E) items[i];
}
peel() 方法很简单,直接返回数组中队尾的元素,并不会删除元素。如果队列中没有元素返回的是 null。

4,总结

    一波源码看下来,ArrayBlockingQueue 使用可重入锁 ReentrantLock 控制队列的访问,两个 Condition 实现生产者-消费者模型,看起来很简单的样子,这背后要感谢 ReentrantLock 和 Condition 的功劳!

这篇关于ArrayBlockingQueue源码分析总结的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1024662

相关文章

HarmonyOS学习(七)——UI(五)常用布局总结

自适应布局 1.1、线性布局(LinearLayout) 通过线性容器Row和Column实现线性布局。Column容器内的子组件按照垂直方向排列,Row组件中的子组件按照水平方向排列。 属性说明space通过space参数设置主轴上子组件的间距,达到各子组件在排列上的等间距效果alignItems设置子组件在交叉轴上的对齐方式,且在各类尺寸屏幕上表现一致,其中交叉轴为垂直时,取值为Vert

学习hash总结

2014/1/29/   最近刚开始学hash,名字很陌生,但是hash的思想却很熟悉,以前早就做过此类的题,但是不知道这就是hash思想而已,说白了hash就是一个映射,往往灵活利用数组的下标来实现算法,hash的作用:1、判重;2、统计次数;

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题,本文将一起从代码层去分析为什么没有建立索引? 开源ERP项目地址:https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置

JAVA智听未来一站式有声阅读平台听书系统小程序源码

智听未来,一站式有声阅读平台听书系统 🌟&nbsp;开篇:遇见未来,从“智听”开始 在这个快节奏的时代,你是否渴望在忙碌的间隙,找到一片属于自己的宁静角落?是否梦想着能随时随地,沉浸在知识的海洋,或是故事的奇幻世界里?今天,就让我带你一起探索“智听未来”——这一站式有声阅读平台听书系统,它正悄悄改变着我们的阅读方式,让未来触手可及! 📚&nbsp;第一站:海量资源,应有尽有 走进“智听

git使用的说明总结

Git使用说明 下载安装(下载地址) macOS: Git - Downloading macOS Windows: Git - Downloading Windows Linux/Unix: Git (git-scm.com) 创建新仓库 本地创建新仓库:创建新文件夹,进入文件夹目录,执行指令 git init ,用以创建新的git 克隆仓库 执行指令用以创建一个本地仓库的

Java ArrayList扩容机制 (源码解读)

结论:初始长度为10,若所需长度小于1.5倍原长度,则按照1.5倍扩容。若不够用则按照所需长度扩容。 一. 明确类内部重要变量含义         1:数组默认长度         2:这是一个共享的空数组实例,用于明确创建长度为0时的ArrayList ,比如通过 new ArrayList<>(0),ArrayList 内部的数组 elementData 会指向这个 EMPTY_EL

如何在Visual Studio中调试.NET源码

今天偶然在看别人代码时,发现在他的代码里使用了Any判断List<T>是否为空。 我一般的做法是先判断是否为null,再判断Count。 看了一下Count的源码如下: 1 [__DynamicallyInvokable]2 public int Count3 {4 [__DynamicallyInvokable]5 get

SWAP作物生长模型安装教程、数据制备、敏感性分析、气候变化影响、R模型敏感性分析与贝叶斯优化、Fortran源代码分析、气候数据降尺度与变化影响分析

查看原文>>>全流程SWAP农业模型数据制备、敏感性分析及气候变化影响实践技术应用 SWAP模型是由荷兰瓦赫宁根大学开发的先进农作物模型,它综合考虑了土壤-水分-大气以及植被间的相互作用;是一种描述作物生长过程的一种机理性作物生长模型。它不但运用Richard方程,使其能够精确的模拟土壤中水分的运动,而且耦合了WOFOST作物模型使作物的生长描述更为科学。 本文让更多的科研人员和农业工作者

MOLE 2.5 分析分子通道和孔隙

软件介绍 生物大分子通道和孔隙在生物学中发挥着重要作用,例如在分子识别和酶底物特异性方面。 我们介绍了一种名为 MOLE 2.5 的高级软件工具,该工具旨在分析分子通道和孔隙。 与其他可用软件工具的基准测试表明,MOLE 2.5 相比更快、更强大、功能更丰富。作为一项新功能,MOLE 2.5 可以估算已识别通道的物理化学性质。 软件下载 https://pan.quark.cn/s/57

工厂ERP管理系统实现源码(JAVA)

工厂进销存管理系统是一个集采购管理、仓库管理、生产管理和销售管理于一体的综合解决方案。该系统旨在帮助企业优化流程、提高效率、降低成本,并实时掌握各环节的运营状况。 在采购管理方面,系统能够处理采购订单、供应商管理和采购入库等流程,确保采购过程的透明和高效。仓库管理方面,实现库存的精准管理,包括入库、出库、盘点等操作,确保库存数据的准确性和实时性。 生产管理模块则涵盖了生产计划制定、物料需求计划、