Java并发包源码学习系列:阻塞队列实现之ArrayBlockingQueue源码解析

本文主要是介绍Java并发包源码学习系列:阻塞队列实现之ArrayBlockingQueue源码解析,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

    • ArrayBlockingQueue概述
    • 类图结构及重要字段
    • 构造器
    • 出队和入队操作
      • 入队enqueue
      • 出队dequeue
    • 阻塞式操作
      • E take() 阻塞式获取
      • void put(E e) 阻塞式插入
      • E poll(timeout, unit) 阻塞式超时获取
      • boolean offer(e, timeout, unit) 阻塞式超时插入
    • 其他常规操作
      • boolean offer(E e)
      • E poll()
      • Boolean remove(Object o)
    • 总结
    • 参考阅读

系列传送门:

  • Java并发包源码学习系列:AbstractQueuedSynchronizer
  • Java并发包源码学习系列:CLH同步队列及同步资源获取与释放
  • Java并发包源码学习系列:AQS共享式与独占式获取与释放资源的区别
  • Java并发包源码学习系列:ReentrantLock可重入独占锁详解
  • Java并发包源码学习系列:ReentrantReadWriteLock读写锁解析
  • Java并发包源码学习系列:详解Condition条件队列、signal和await
  • Java并发包源码学习系列:挂起与唤醒线程LockSupport工具类
  • Java并发包源码学习系列:JDK1.8的ConcurrentHashMap源码解析
  • Java并发包源码学习系列:阻塞队列BlockingQueue及实现原理分析

ArrayBlockingQueue概述

ArrayBlockingQueue是由数组构成的有界阻塞队列,支持FIFO的次序对元素进行排序。

这是一个典型的有界缓冲结构,可指定大小存储元素,供生产线程插入,供消费线程获取,但注意,容量一旦指定,便不可修改。

队列空时尝试take操作和队列满时尝试put操作都会阻塞执行操作的线程。

该类还支持可供选择的公平性策略ReentrantLock可重入锁实现,默认采用非公平策略,当队列可用时,阻塞的线程都可以争夺访问队列的资格。

阻塞队列通过ReentrantLock + Condition实现并发环境下的等待通知机制:读操作和写操作都需要获取到AQS独占锁才能进行操作,如果队列为空,则读操作线程将会被包装为条件节点扔到读线程等待条件队列中,等待写线程写入新的元素,同时读线程将会被唤醒,反之亦然。

类图结构及重要字段

public class ArrayBlockingQueue<E> extends AbstractQueue<E>implements BlockingQueue<E>, java.io.Serializable {// 序列号, 用于序列化private static final long serialVersionUID = -817911632652898426L;// 底层存储数据的定长数组final Object[] items;// 移除操作的index,可以理解为队头位置int takeIndex;// 添加操作的index,可以理解为队尾位置int putIndex;// 元素个数int count;// 独占重入锁final ReentrantLock lock;// 等待takes的条件对象private final Condition notEmpty;// 等待puts的条件对象private final Condition notFull;// Itrs表示队列和迭代器之间的共享数据,其实用来存储多个迭代器实例的transient Itrs itrs = null;
}

构造器

使用ArrayBlockingQueue的时候,必须指定一个capacity阻塞队列的容量。可以传入可选的fair值,以采取不同公平性策略,默认使用非公平的策略。另外,可以传入集合对象,直接构造阻塞队列。

    // 必须指定容量, 默认采用非公平策略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); // 初始化locknotEmpty = lock.newCondition(); // 初始化条件变量notEmptynotFull =  lock.newCondition(); // 初始化条件变量notFull}// 另外,可指定传入集合直接构造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 exclusiontry {int i = 0;try {for (E e : c) { // 遍历赋值checkNotNull(e);items[i++] = e;}} catch (ArrayIndexOutOfBoundsException ex) {throw new IllegalArgumentException();}count = i;putIndex = (i == capacity) ? 0 : i;} finally {lock.unlock();}}

出队和入队操作

队列的操作最核心的部分莫过于入队和出队了,后面分析的方法基本上都基于这两个工具方法。

入队enqueue

    private void enqueue(E x) {// assert lock.getHoldCount() == 1;// assert items[putIndex] == null;final Object[] items = this.items;// 把元素x放入数组items[putIndex] = x;// 下一个元素应该存放的下标位置if (++putIndex == items.length)putIndex = 0;count++;// 激活notEmpty的条件队列因调用take操作而被阻塞的一个线程notEmpty.signal();}
  1. 将元素x置入数组中。
  2. 计算下一个元素应该存放的下标位置。
  3. 元素个数器递增,这里count前加了锁,值都是从主内存中获取,不会存在内存不可见问题,并且更新也会直接刷新回主内存中。
  4. 最后激活notEmpty的条件队列因调用take操作而被阻塞的一个线程。

出队dequeue

    private E dequeue() {// assert lock.getHoldCount() == 1;// assert items[takeIndex] != null;final Object[] items = this.items;@SuppressWarnings("unchecked")// 获取元素E x = (E) items[takeIndex];// 置nullitems[takeIndex] = null;// 重新设置对头下标if (++takeIndex == items.length)takeIndex = 0;// 更新元素计数器count--;// 更新迭代器中的元素数据,itrs只用在使用迭代器的时候才实例化哦if (itrs != null)itrs.elementDequeued();// 激活notFull的条件队列因调用put操作而被阻塞的一个线程notFull.signal();return x;}
  1. 获取元素,并将当前位置置null。
  2. 重新设置队头下标。
  3. 元素计数器递减。
  4. 更新迭代器中的元素数据,itrs默认情况下都是为null的,只有使用迭代器的时候才会实例化Itrs。
  5. 激活notFull的条件队列因调用put操作而被阻塞的一个线程。

阻塞式操作

E take() 阻塞式获取

take操作将会获取当前队列头部元素并移除,如果队列为空则阻塞当前线程直到队列不为空,退出阻塞时返回获取的元素。

那,线程阻塞至何时如何知道呢,其实当前线程将会因notEmpty.await()被包装成等待节点置入notEmpty的条件队列中,一旦enqueue操作成功触发,也就是入队成功,将会执行notEmpty.signal()唤醒条件队列中等待的线程,被转移到AQS队列中参与锁的争夺。

如果线程在阻塞时被其他线程设置了中断标志,则抛出InterruptedException异常并返回。

    public E take() throws InterruptedException {final ReentrantLock lock = this.lock;// 可响应中断式地获取锁lock.lockInterruptibly();try {// 如果队列为空,则将当前线程包装为等待节点置入notEmpty的条件队列中while (count == 0)notEmpty.await();// 非空,则执行入队操作,入队时唤醒notFull的条件队列中的第一个线程return dequeue();} finally {lock.unlock();}}

void put(E e) 阻塞式插入

put操作将向队尾插入元素,如果队列未满则插入,如果队列已满,则阻塞当前线程直到队列不满。

如果线程在阻塞时被其他线程设置了中断标志,则抛出InterruptedException异常并返回。

    public void put(E e) throws InterruptedException {checkNotNull(e);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {// 如果队列满,则将当前线程包装为等待节点置入notFull的条件队列中while (count == items.length)notFull.await();// 非满,则执行入队操作,入队时唤醒notEmpty的条件队列中的第一个线程enqueue(e);} finally {lock.unlock();}}

E poll(timeout, unit) 阻塞式超时获取

在take阻塞式获取方法的基础上额外增加超时功能,传入一个timeout,获取不到而阻塞的时候,如果时间到了,即使还获取不到,也只能立即返回null。

    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();}}

boolean offer(e, timeout, unit) 阻塞式超时插入

在put阻塞式插入方法的基础上额外增加超时功能,传入一个timeout,获取不到而阻塞的时候,如果时间到了,即使还获取不到,也只能立即返回null。

    public boolean offer(E e, long timeout, TimeUnit unit)throws InterruptedException {checkNotNull(e);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();}}

其他常规操作

boolean offer(E e)

offer(E e)是非阻塞的方法,向队尾插入一个元素,如果队列未满,则插入成功并返回true;如果队列已满则丢弃当前元素,并返回false。

    public boolean offer(E e) {checkNotNull(e); // 如果插入元素为null,则抛出NullPointerException异常// 获取独占锁final ReentrantLock lock = this.lock; lock.lock();try {// 如果队列满, 则返回falseif (count == items.length)return false;else {// 否则则入队enqueue(e);return true;}} finally {lock.unlock();}}

E poll()

从队列头部获取并移除第一个元素,如果队列为空则返回null。

    public E poll() {final ReentrantLock lock = this.lock;lock.lock();try {// 如果为空,返回null, 否则执行出队操作return (count == 0) ? null : dequeue();} finally {lock.unlock();}}

Boolean remove(Object o)

移除队列中与元素o相等【指的是equals方法判定相同】的元素,移除成功返回true,如果队列为空或没有匹配元素,则返回false。

    public boolean remove(Object o) {if (o == null) return false;final Object[] items = this.items;final ReentrantLock lock = this.lock;lock.lock();try {if (count > 0) {// 获取当前队尾位置final int putIndex = this.putIndex;// 从队头开始遍历int i = takeIndex;do {// 找到了对应的元素的位置,removeAt删除该位置的元素if (o.equals(items[i])) {removeAt(i);return true;}if (++i == items.length)i = 0;} while (i != putIndex);}return false;} finally {lock.unlock();}}// 移除removeIndex位置的元素void removeAt(final int removeIndex) {// assert lock.getHoldCount() == 1;// assert items[removeIndex] != null;// assert removeIndex >= 0 && removeIndex < items.length;final Object[] items = this.items;// 如果要移除元素的位置正好就是 队头位置,和之前出队操作一样if (removeIndex == takeIndex) {// removing front item; just advanceitems[takeIndex] = null;if (++takeIndex == items.length)takeIndex = 0;count--;if (itrs != null)itrs.elementDequeued();} else {// an "interior" remove// slide over all others up through putIndex.final int putIndex = this.putIndex;// 移除的不是队头,那就要对应将后面的元素补充上来,并更新putIndex的位置for (int i = removeIndex;;) {int next = i + 1;if (next == items.length)next = 0;// 移除的不是队尾,后面的元素补充上来if (next != putIndex) {items[i] = items[next];i = next;} else {// 移除的是队尾元素items[i] = null;this.putIndex = i;break;}}count--;if (itrs != null)itrs.removedAt(removeIndex);}notFull.signal();}

总结

  • ArrayBlockingQueue基于数组的有界阻塞队列,必须指定容量大小,及队列中最多允许的元素个数。

  • 提供了take和put两个阻塞式的操作,还提供了阻塞式+超时机制的操作。

  • 阻塞队列通过ReentrantLock + Condition实现并发环境下的等待通知机制:读操作和写操作都需要获取到AQS独占锁才能进行操作,如果队列为空,则读操作线程将会被包装为条件节点扔到读线程等待条件队列中阻塞,等待写线程写入新的元素,并唤醒等待中的读线程,反之亦然。

本篇重点看了出队入队相关方法,其余部分如迭代器相关不是本文重点,如果想了解学习可以参看:ArrayBlockingQueue 迭代器

参考阅读

  • 《Java并发编程之美》
  • 《Java并发编程的艺术》

这篇关于Java并发包源码学习系列:阻塞队列实现之ArrayBlockingQueue源码解析的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

网页解析 lxml 库--实战

lxml库使用流程 lxml 是 Python 的第三方解析库,完全使用 Python 语言编写,它对 XPath表达式提供了良好的支 持,因此能够了高效地解析 HTML/XML 文档。本节讲解如何通过 lxml 库解析 HTML 文档。 pip install lxml lxm| 库提供了一个 etree 模块,该模块专门用来解析 HTML/XML 文档,下面来介绍一下 lxml 库

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

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

Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

Ilya(不是本人,claude AI)在社交媒体上分享了他在OpenAI学习到的15个Prompt撰写技巧。 以下是详细的内容: 提示精确化:在编写提示时,力求表达清晰准确。清楚地阐述任务需求和概念定义至关重要。例:不用"分析文本",而用"判断这段话的情感倾向:积极、消极还是中性"。 快速迭代:善于快速连续调整提示。熟练的提示工程师能够灵活地进行多轮优化。例:从"总结文章"到"用

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时,有没有考虑过 JVM 是如何把静态的字节码(byte code)转化为运行时对象的呢,这个问题看似简单,但清楚的同学相信也不会太多,这篇文章首先介绍 JVM 类初始化的机制,然后给出几个易出错的实例来分析,帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段,分别是:loading 、Linking、initialization

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权,允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true (Spring security默认会带有ROLE_前缀),去

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程,特意画了如下的类图,包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口,接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类,它们之间的依赖,构建起了整个框架。想要理解整个架构,最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文(security context)的信息。当前操作的用户是谁,该用户是否已经被认证,他拥有哪些角色权限…这些都被保

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明(认真看哦) AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证,则会调用 #attemptAuthentica

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门(一):登录与退出》 《Spring-Security-入门(二):基于数据库验证》 《Spring-Security-入门(三):密码加密》 《Spring-Security-入门(四):自定义-Filter》 《Spring-Security-入门(五):在 Sprin

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物