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

2024-01-08 02:58
文章标签 java 实现 源码 学习 系列 队列 解析 阻塞 发包 delayqueue

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

文章目录

    • DelayQueue概述
    • 类图及重要字段
    • Delayed接口
    • Delayed元素案例
    • 构造器
    • void put(E e)
    • E take()
      • first = null 有什么用
    • 总结
    • 参考阅读

系列传送门：

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

DelayQueue概述

DelayQueue是一个支持延时获取元素的无界阻塞队列，使用PriorityQueue来存储元素。

队中的元素必须实现Delayed接口【Delay接口又继承了Comparable，需要实现compareTo方法】，每个元素都需要指明过期时间，通过getDelay(unit)获取元素剩余时间【剩余时间 = 到期时间 - 当前时间】，每次向优先队列中添加元素时根据compareTo方法作为排序规则。

当从队列获取元素时，只有过期的元素才会出队列。

使用场景: 缓存系统设计、定时任务调度等。

类图及重要字段

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>implements BlockingQueue<E> {// 独占锁实现同步private final transient ReentrantLock lock = new ReentrantLock();// 优先队列存放数据private final PriorityQueue<E> q = new PriorityQueue<E>();/*** 基于Leader-Follower模式的变体,用于尽量减少不必要的线程等待*/private Thread leader = null;/*** 与lock对应的条件变量*/private final Condition available = lock.newCondition();    
}
  1. 使用ReentrantLock独占锁实现线程同步，使用Condition实现等待通知机制。
  2. 基于Leader-Follower模式的变体，减少不必要的线程等待。
  3. 内部使用PriorityQueue优先级队列存储元素，且队列中元素必须实现Delayed接口。

Delayed接口

队中的元素必须实现Delayed接口【Delay接口又继承了Comparable，需要实现compareTo方法】，每个元素都需要指明过期时间，通过getDelay(unit)获取元素剩余时间【剩余时间 = 到期时间 - 当前时间】。

每次向优先队列中添加元素时根据compareTo方法作为排序规则，当然我们约定一下，默认q.peek()出来的就是最先过期的元素。

public interface Delayed extends Comparable<Delayed> {// 返回剩余时间long getDelay(TimeUnit unit);
}public interface Comparable<T> {// 定义比较方法public int compareTo(T o);
}

Delayed元素案例

学习了Delayed接口之后，我们看一个实际的案例，加深印象，源于：《Java并发编程之美》。

    static class DelayedElement implements Delayed {private final long delayTime; // 延迟时间private final long expire; // 到期时间private final String taskName; // 任务名称public DelayedElement (long delayTime, String taskName) {this.delayTime = delayTime;this.taskName = taskName;expire = now() + delayTime;}final long now () {return System.currentTimeMillis();}// 剩余时间 = 到期时间 - 当前时间@Overridepublic long getDelay (TimeUnit unit) {return unit.convert(expire - now(), TimeUnit.MILLISECONDS);}@Overridepublic int compareTo (Delayed o) {return (int) (getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));}@Overridepublic String toString () {final StringBuilder res = new StringBuilder("DelayedElement [ ");res.append("delay = ").append(delayTime);res.append(", expire = ").append(expire);res.append(", taskName = '").append(taskName).append('\'');res.append(" ] ");return res.toString();}}public static void main (String[] args) {// 创建delayQueue队列DelayQueue<DelayedElement> delayQueue = new DelayQueue<>();// 创建延迟任务Random random = new Random();for (int i = 0; i < 10; i++) {DelayedElement element = new DelayedElement(random.nextInt(500), "task: " + i);delayQueue.offer(element);}// 依次取出任务并打印DelayedElement ele = null;try {for (; ; ) {while ((ele = delayQueue.take()) != null) {System.out.println(ele);}}} catch (InterruptedException ex) {ex.printStackTrace();}}
// 打印结果
DelayedElement [ delay = 2, expire = 1611995426061, taskName = 'task: 4' ] 
DelayedElement [ delay = 52, expire = 1611995426111, taskName = 'task: 2' ] 
DelayedElement [ delay = 80, expire = 1611995426139, taskName = 'task: 5' ] 
DelayedElement [ delay = 132, expire = 1611995426191, taskName = 'task: 0' ] 
DelayedElement [ delay = 174, expire = 1611995426233, taskName = 'task: 9' ] 
DelayedElement [ delay = 175, expire = 1611995426234, taskName = 'task: 7' ] 
DelayedElement [ delay = 326, expire = 1611995426385, taskName = 'task: 3' ] 
DelayedElement [ delay = 447, expire = 1611995426506, taskName = 'task: 8' ] 
DelayedElement [ delay = 452, expire = 1611995426511, taskName = 'task: 1' ] 
DelayedElement [ delay = 486, expire = 1611995426545, taskName = 'task: 6' ]
  • 实现了compareTo方法，定义比较规则为越早过期的排在队头。
  • 实现了getDelay方法，计算公式为：剩余时间 = 到期时间 - 当前时间。

构造器

DelayQueue构造器相比于前几个，就显得非常easy了。

    public DelayQueue() {}public DelayQueue(Collection<? extends E> c) {this.addAll(c);}

void put(E e)

因为DelayQueue是无界队列，不会因为边界问题产生阻塞，因此put操作和offer操作是一样的。

    public void put(E e) {offer(e);}public boolean offer(E e) {// 获取独占锁final ReentrantLock lock = this.lock;lock.lock();try {// 加入优先队列里q.offer(e);// 判断堆顶元素是不是刚刚插入的元素// 如果判断为true，说明当前这个元素是将最先过期if (q.peek() == e) {// 重置leader线程为nullleader = null; // 激活available变量条件队列中的一个线程available.signal();}return true;} finally {lock.unlock();}}

E take()

take方法将会获取并移除队列里面延迟时间过期的元素 ，如果队列里面没有过期元素则陷入等待。

    public E take() throws InterruptedException {// 获取独占锁final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {for (;;) {// 瞅一瞅谁最快过期E first = q.peek();// 队列为空，则将当前线程置入available的条件队列中，直到里面有元素if (first == null)available.await();else {// 看下还有多久过期long delay = first.getDelay(NANOSECONDS);// 哇，已经过期了，就移除它并返回if (delay <= 0)return q.poll();first = null; // don't retain ref while waiting// leader不为null表示其他线程也在执行take// 则将当前线程置入available的条件队列中if (leader != null)available.await();else {// 如果leader为null，则选择当前线程作为leader线程Thread thisThread = Thread.currentThread();leader = thisThread;try {// 等待delay时间，时间到之后，会出条件队列，继续竞争锁available.awaitNanos(delay);} finally {if (leader == thisThread)leader = null;}}}}} finally {if (leader == null && q.peek() != null)available.signal();lock.unlock();}}

first = null 有什么用

如果不设置first = null，将会引起内存泄露。

  • 线程A到达，队首元素没有到期，设置leader = 线程A，并且执行available.awaitNanos(delay);等待元素过期。
  • 这时线程B来了，因为leader != null，则会available.await();阻塞，线程C、D、E同理。
  • 线程A阻塞完毕了，再次循环，获取列首元素成功，出列。

这个时候列首元素应该会被回收掉，但是问题是它还被线程B、线程C持有着，所以不会回收，如果线程增多，且队首元素无限期的不能回收，就会造成内存泄漏。

总结

DelayQueue是一个支持延时获取元素的无界阻塞队列，使用PriorityQueue来存储元素。

队中的元素必须实现Delayed接口【Delay接口又继承了Comparable，需要实现compareTo方法】，每个元素都需要指明过期时间，通过getDelay(unit)获取元素剩余时间【剩余时间 = 到期时间 - 当前时间】，每次向优先队列中添加元素时根据compareTo方法作为排序规则。

基于Leader-Follower模式使用leader变量，减少不必要的线程等待。

DelayQueue是无界队列，因此插入操作是非阻塞的。但是take操作从队列获取元素时，是阻塞的，阻塞规则为：

  • 当一个线程调用队列的take方法，如果队列为空，则将会调用available.await()陷入阻塞。
  • 如果队列不为空，则查看队列的队首元素是否过期，根据getDelay的返回值是否小于0判断，如果过期则返回该元素。
  • 如果队首元素未过期，则判断当前线程是否为leader线程，如果不是，表明有其他线程在执行take操作，就调用available.await()陷入阻塞。
  • 如果没有其他线程在执行take，就将当前线程设置为leader，并等待队首元素过期，available.awaitNanos(delay)。
  • leader线程退出take之后，将会调用available.signal()唤醒一个follower线程，接着回到开始那步。

参考阅读

  • 《Java并发编程的艺术》

  • 《Java并发编程之美》

  • 【死磕Java并发】—–J.U.C之阻塞队列：DelayQueue

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


http://www.chinasem.cn/article/582123。

相关文章

网页解析 lxml 库--实战

网页解析 lxml 库--实战

lxml库使用流程 lxml 是 Python 的第三方解析库，完全使用 Python 语言编写，它对 XPath表达式提供了良好的支 持，因此能够了高效地解析 HTML/XML 文档。本节讲解如何通过 lxml 库解析 HTML 文档。 pip install lxml lxm| 库提供了一个 etree 模块，该模块专门用来解析 HTML/XML 文档，下面来介绍一下 lxml 库
阅读更多...
HarmonyOS学习(七)——UI（五）常用布局总结

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

自适应布局 1.1、线性布局（LinearLayout） 通过线性容器Row和Column实现线性布局。Column容器内的子组件按照垂直方向排列，Row组件中的子组件按照水平方向排列。 属性说明space通过space参数设置主轴上子组件的间距，达到各子组件在排列上的等间距效果alignItems设置子组件在交叉轴上的对齐方式，且在各类尺寸屏幕上表现一致，其中交叉轴为垂直时，取值为Vert
阅读更多...
Ilya-AI分享的他在OpenAI学习到的15个提示工程技巧

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

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

JVM 的类初始化机制

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

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

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

浅析Spring Security认证过程

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

Spring Security--Architecture Overview

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

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

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
Spring Security 从入门到进阶系列教程

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

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
Java架构师知识体认识

Java架构师知识体认识

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

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2