• 盘点JAVA中延时任务的几种实现方式

    场景描述

    ①需要实现一个定时发布系统通告的功能，如何实现？ ②支付超时，订单自动取消，如何实现？

    实现方式

    一、挂起线程

    推荐指数：★★☆ 优点： JDK原生(JUC包下)支持，无需引入新的依赖； 缺点： (1)基于内存，应用重启(或宕机)会导致任务丢失 (2)基于内存挂起线程实现延时，不支持集群 (3)代码耦合性大，不易维护 (4)一个任务就要新建一个线程绑定任务的执行，容易造成资源浪费

    ①配置延迟任务专用线程池

    1. /**
    2.  * 线程池配置
    3.  */
    4. @Configuration
    5. @EnableAsync
    6. @EnableConfigurationProperties(ThreadPoolProperties.class)
    7. public class ThreadPoolConfig {
    8.  
    9. 	//ThreadPoolProperties的配置依据需求和服务器配置自行配置
    10.     @Resource
    11.     private ThreadPoolProperties threadPoolProperties;
    12.     //延迟任务队列容量
    13.     private final static int DELAY_TASK_QUEUE_CAPACITY = 100;
    14.  
    15.     @Bean
    16.     public ThreadPoolTaskExecutor delayTaskExecutor() {
    17.         log.info("start delayTaskExecutor");
    18.         ThreadPoolTaskExecutor threadPool = new ThreadPoolTaskExecutor();
    19.         //配置核心线程数
    20.         threadPool.setCorePoolSize(threadPoolProperties.getCorePoolSize());
    21.         //配置最大线程数
    22.         threadPool.setMaxPoolSize(threadPoolProperties.getMaxPoolSize());
    23.         //配置队列大小
    24.         threadPool.setQueueCapacity(DELAY_TASK_QUEUE_CAPACITY);
    25.         //线程最大存活时间
    26.         threadPool.setKeepAliveSeconds (threadPoolProperties.getKeepAliveSeconds());
    27.         //配置线程池中的线程的名称前缀
    28.         threadPool.setThreadNamePrefix(threadPoolProperties.getThreadNamePrefix());
    29.  
    30.         // rejection-policy：当pool已经达到max size的时候执行的策略
    31.         threadPool.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
    32.         //执行初始化
    33.         threadPool.initialize();
    34.         return threadPool;
    35.     }
    36. }
    37. 复制代码

    ②创建延时任务

    在需要执行的代码块创建延时任务

    1. delayTaskExecutor.execute(() -> {
    2.     try {
    3.         //线程挂起指定时间
    4.         TimeUnit.MINUTES.sleep(time);
    5.         //执行业务逻辑
    6.         doSomething();
    7.     } catch (InterruptedException e) {
    8.         log.error("线程被打断，执行业务逻辑失败");
    9.     }
    10. });
    11. 复制代码

    二、ScheduledExecutorService 延迟任务线程池

    推荐指数：★★★ 优点： 代码简洁，JDK原生支持 缺点： (1)基于内存，应用重启(或宕机)会导致任务丢失 (2)基于内存存放任务，不支持集群 (3)一个任务就要新建一个线程绑定任务的执行，容易造成资源浪费

    1. class Task implements Runnable{
    2.  
    3.     @Override
    4.     public void run() {
    5.         System.out.println(Thread.currentThread().getId()+":"+Thread.currentThread().getName());
    6.         System.out.println("scheduledExecutorService====>>>延时器");
    7.     }
    8. }
    9. public class ScheduleServiceTest {
    10.     public static void main(String[] args) {
    11.         ScheduledExecutorService scheduledExecutorService=new ScheduledThreadPoolExecutor(10);
    12.         scheduledExecutorService.schedule(new Task(),1, TimeUnit.SECONDS);
    13.         scheduledExecutorService.schedule(new Task(),2, TimeUnit.SECONDS);
    14.         scheduledExecutorService.schedule(new Task(),1, TimeUnit.SECONDS);
    15.     }
    16. }
    17. 复制代码

    三、DelayQueue(延时队列)

    推荐指数：★★★☆ 优点： (1)JDK原生(JUC包下)支持，无需引入新的依赖； (2)可以用一个线程对整个延时队列按序执行； 缺点： (1)基于内存，应用重启(或宕机)会导致任务丢失 (2)基于内存存放队列，不支持集群 (3)依据compareTo方法排列队列，调用take阻塞式的取出第一个任务(不调用则不取出)，比较不灵活，会影响时间的准确性

    ①新建一个延时任务

    1. public class DelayTask implements Delayed {
    2.  
    3.     private Integer taskId;
    4.  
    5.     private long executeTime;
    6.  
    7.     DelayTask(Integer taskId, long executeTime) {
    8.         this.taskId = taskId;
    9.         this.executeTime = executeTime;
    10.     }
    11.  
    12.     /**
    13.      * 该任务的延时时长
    14.      * @param unit
    15.      * @return
    16.      */
    17.     @Override
    18.     public long getDelay(TimeUnit unit) {
    19.         return executeTime - System.currentTimeMillis();
    20.     }
    21.  
    22.     @Override
    23.     public int compareTo(Delayed o) {
    24.         DelayTask t = (DelayTask) o;
    25.         if (this.executeTime - t.executeTime <= 0) {
    26.             return -1;
    27.         } else {
    28.             return 1;
    29.         }
    30.     }
    31.  
    32.     @Override
    33.     public String toString() {
    34.         return "延时任务{" +
    35.                 "任务编号=" + taskId +
    36.                 ", 执行时间=" + new Date(executeTime) +
    37.                 '}';
    38.     }
    39.  
    40.     /**
    41.      * 执行具体业务代码
    42.      */
    43.     public void doTask(){
    44.         System.out.println(this+":");
    45.         System.out.println("线程ID-"+Thread.currentThread().getId()+":线程名称-"+Thread.currentThread().getName()+":do something!");
    46.     }
    47. }
    48. 复制代码

    ②执行延时任务

    1. public class TestDelay {
    2.     public static void main(String[] args) throws InterruptedException {
    3.         // 新建3个任务，并依次设置超时时间为 30s 10s 60s
    4.         DelayTask d1 = new DelayTask(1, System.currentTimeMillis() + 3000L);
    5.         DelayTask d2 = new DelayTask(2, System.currentTimeMillis() + 1000L);
    6.         DelayTask d3 = new DelayTask(3, System.currentTimeMillis() + 6000L);
    7.  
    8.         DelayQueue<DelayTask> queue = new DelayQueue<>();
    9.         queue.add(d1);
    10.         queue.add(d2);
    11.         queue.add(d3);
    12.  
    13.         System.out.println("开启延时队列时间：" + new Date()+"\n");
    14.  
    15.         // 从延时队列中获取元素
    16.         while (!queue.isEmpty()) {
    17.             queue.take().doTask();
    18.         }
    19.         System.out.println("\n任务结束");
    20.     }
    21. }
    22. 复制代码

    执行结果：

    四、Redis-为key指定超时时长，并监听失效key

    推荐指数：★★★☆ 优点： 对于有依赖redis的业务且有延时任务的需求，能够快速对接 缺点： (1)客户端断开后重连会导致所有事件丢失 (2)高并发场景下，存在大量的失效key场景会导出失效时间存在延迟 (3)若有多个监听器监听该key，是会重复消费这个过期事件的，需要特定逻辑判断

    ① 修改Redis配置文件并重启Redis

    1. notify-keyspace-events Ex
    2. 复制代码

    注意： redis配置文件不能有空格，否则会启动报错

    ②Java中关于Redis的配置类

    redisTemplate实例bean需要自定义生成； RedisMessageListenerContainer 是redis-key过期监听需要的监听器容器；

    1. @Configuration
    2. @Slf4j
    3. public class RedisConfiguration {
    4.     /**
    5.      * Redis配置
    6.      * @param factory
    7.      * @return
    8.      */
    9.     @Bean(name = "redisTemplate")
    10.     public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory factory) {
    11.         RedisTemplate<Object, Object> template = new RedisTemplate<>();
    12.         RedisSerializer<String> redisSerializer = new StringRedisSerializer();
    13.  
    14.         template.setConnectionFactory(factory);
    15.         //key序列化方式
    16.         template.setKeySerializer(redisSerializer);
    17.         //value序列化
    18.         template.setValueSerializer(redisSerializer);
    19.         //value hashmap序列化
    20.         template.setHashValueSerializer(redisSerializer);
    21.         //key hashmap序列化
    22.         template.setHashKeySerializer(redisSerializer);
    23.  
    24.         return template;
    25.     }
    26.  
    27.     /**
    28.      * 消息监听器容器bean
    29.      * @param connectionFactory
    30.      * @return
    31.      */
    32.     @Bean
    33.     public RedisMessageListenerContainer container(LettuceConnectionFactory connectionFactory) {
    34.  
    35.         RedisMessageListenerContainer container = new RedisMessageListenerContainer();
    36.         container.setConnectionFactory(connectionFactory);
    37.         return container;
    38.     }
    39. }
    40. 复制代码

    ③监听器代码

    1. @Slf4j
    2. @Component
    3. public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
    4.     private static final String TEST_REDIS_KEY = "testExpired";
    5.     public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer,
    6.                                       RedisTemplate redisTemplate) {
    7.         super(listenerContainer);
    8.         /**
    9.          * 设置一个Redis延迟过期key(key名：testExpired，过期时间：30秒)
    10.          */
    11.         redisTemplate.opsForValue().set(TEST_REDIS_KEY, "1", 20, TimeUnit.SECONDS);
    12.         log.info("设置redis-key");
    13.     }
    14.  
    15.     @Override
    16.     public void onMessage(Message message, byte[] pattern) {
    17.         try {
    18.             String expiredKey = message.toString();
    19.             if (TEST_REDIS_KEY.equals(expiredKey)) {
    20.                 //业务处理
    21.                 log.info(expiredKey + "过期，触发回调");
    22.             }
    23.         } catch (Exception e) {
    24.             log.error("key 过期通知处理异常，{}", e);
    25.         }
    26.  
    27.     }
    28. }
    29. 复制代码

    测试结果：

    五、时间轮

    推荐指数：★★★★ 优点： (1)对于大量定时任务，时间轮可以仅用一个工作线程对编排的任务进行顺序运行； (2)自动运行，可以自定义时间轮每轮的tick数，tick间隔，灵活且时间精度可控 缺点： (1)基于内存，应用重启(或宕机)会导致任务丢失 (2)基于内存存放任务，不支持集群

    1. public class WheelTimerTest {
    2.  
    3.     public static void main(String[] args) {
    4.  
    5.         //设置每个格子是 100ms, 总共 256 个格子
    6.         HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(100, TimeUnit.MILLISECONDS, 256);
    7.  
    8.         //加入三个任务，依次设置超时时间是 10s 5s 20s
    9.  
    10.         System.out.println("加入一个任务，ID = 1, time= " + LocalDateTime.now());
    11.         hashedWheelTimer.newTimeout(timeout -> {
    12.             System.out.println(Thread.currentThread().getName());
    13.             System.out.println("执行一个任务，ID = 1, time= " + LocalDateTime.now());
    14.         }, 10, TimeUnit.SECONDS);
    15.  
    16.         System.out.println("加入一个任务，ID = 2, time= " + LocalDateTime.now());
    17.         hashedWheelTimer.newTimeout(timeout -> {
    18.             System.out.println(Thread.currentThread().getName());
    19.             System.out.println("执行一个任务，ID = 2, time= " + LocalDateTime.now());
    20.         }, 5, TimeUnit.SECONDS);
    21.  
    22.         System.out.println("加入一个任务，ID = 3, time= " + LocalDateTime.now());
    23.         hashedWheelTimer.newTimeout(timeout -> {
    24.             System.out.println(Thread.currentThread().getName());
    25.             System.out.println("执行一个任务，ID = 3, time= " + LocalDateTime.now());
    26.         }, 20, TimeUnit.SECONDS);
    27.         System.out.println("加入一个任务，ID = 4, time= " + LocalDateTime.now());
    28.         hashedWheelTimer.newTimeout(timeout -> {
    29.             System.out.println(Thread.currentThread().getName());
    30.             System.out.println("执行一个任务，ID = 4, time= " + LocalDateTime.now());
    31.         }, 20, TimeUnit.SECONDS);
    32.  
    33.         System.out.println("等待任务执行===========");
    34.     }
    35. }
    36. 复制代码

    六、消息队列-延迟队列

    针对任务丢失的代价过大，高并发的场景 推荐指数：★★★★ 优点： 支持集群，分布式，高并发场景； 缺点： 引入额外的消息队列，增加项目的部署和维护的复杂度。

    场景：为一个委托指定期限，委托到期后，委托关系终止，相关业务权限移交回原拥有者 这里采用的是RabbitMq的死信队列加TTL消息转化为延迟队列的方式(RabbitMq没有延时队列)

    ①声明一个队列设定其的死信队列

    1. @Configuration
    2. public class MqConfig {
    3.     public static final String GLOBAL_RABBIT_TEMPLATE = "rabbitTemplateGlobal";
    4.  
    5.     public static final String DLX_EXCHANGE_NAME = "dlxExchange";
    6.     public static final String AUTH_EXCHANGE_NAME = "authExchange";
    7.  
    8.     public static final String DLX_QUEUE_NAME = "dlxQueue";
    9.     public static final String AUTH_QUEUE_NAME = "authQueue";
    10.     public static final String DLX_AUTH_QUEUE_NAME = "dlxAuthQueue";
    11.  
    12.     @Bean
    13.     @Qualifier(GLOBAL_RABBIT_TEMPLATE)
    14.     public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
    15.         RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
    16.         return rabbitTemplate;
    17.     }
    18.  
    19.     @Bean
    20.     @Qualifier(AUTH_EXCHANGE_NAME)
    21.     public Exchange authExchange() {
    22.         return ExchangeBuilder.directExchange (AUTH_EXCHANGE_NAME).durable (true).build ();
    23.     }
    24.  
    25.     /**
    26.      * 死信交换机
    27.      * @return
    28.      */
    29.     @Bean
    30.     @Qualifier(DLX_EXCHANGE_NAME)
    31.     public Exchange dlxExchange() {
    32.         return ExchangeBuilder.directExchange (DLX_EXCHANGE_NAME).durable (true).build ();
    33.     }
    34.  
    35.     /**
    36.      * 记录日志的死信队列
    37.      * @return
    38.      */
    39.     @Bean
    40.     @Qualifier(DLX_QUEUE_NAME)
    41.     public Queue dlxQueue() {
    42.         // Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    43.         return QueueBuilder.durable (DLX_QUEUE_NAME).build ();
    44.     }
    45.  
    46.     /**
    47.      * 委托授权专用队列
    48.      * @return
    49.      */
    50.     @Bean
    51.     @Qualifier(AUTH_QUEUE_NAME)
    52.     public Queue authQueue() {
    53.         return QueueBuilder
    54.                 .durable (AUTH_QUEUE_NAME)
    55.                 .withArgument("x-dead-letter-exchange", DLX_EXCHANGE_NAME)
    56.                 .withArgument("x-dead-letter-routing-key", "dlx_auth")
    57.                 .build ();
    58.     }
    59.  
    60.     /**
    61.      * 委托授权专用死信队列
    62.      * @return
    63.      */
    64.     @Bean
    65.     @Qualifier(DLX_AUTH_QUEUE_NAME)
    66.     public Queue dlxAuthQueue() {
    67.         // Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, Map arguments)
    68.         return QueueBuilder
    69.                 .durable (DLX_AUTH_QUEUE_NAME)
    70.                 .withArgument("x-dead-letter-exchange", DLX_EXCHANGE_NAME)
    71.                 .withArgument("x-dead-letter-routing-key", "dlx_key")
    72.                 .build ();
    73.     }
    74.  
    75.     @Bean
    76.     public Binding bindDlxQueueExchange(@Qualifier(DLX_QUEUE_NAME) Queue dlxQueue, @Qualifier(DLX_EXCHANGE_NAME) Exchange dlxExchange){
    77.         return BindingBuilder.bind (dlxQueue).to (dlxExchange).with ("dlx_key").noargs ();
    78.     }
    79.  
    80.     /**
    81.      * 委托授权专用死信队列绑定关系
    82.      * @param dlxAuthQueue
    83.      * @param dlxExchange
    84.      * @return
    85.      */
    86.     @Bean
    87.     public Binding bindDlxAuthQueueExchange(@Qualifier(DLX_AUTH_QUEUE_NAME) Queue dlxAuthQueue, @Qualifier(DLX_EXCHANGE_NAME) Exchange dlxExchange){
    88.         return BindingBuilder.bind (dlxAuthQueue).to (dlxExchange).with ("dlx_auth").noargs ();
    89.     }
    90.  
    91.     /**
    92.      * 委托授权专用队列绑定关系
    93.      * @param authQueue
    94.      * @param authExchange
    95.      * @return
    96.      */
    97.     @Bean
    98.     public Binding bindAuthQueueExchange(@Qualifier(AUTH_QUEUE_NAME) Queue authQueue, @Qualifier(AUTH_EXCHANGE_NAME) Exchange authExchange){
    99.         return BindingBuilder.bind (authQueue).to (authExchange).with ("auth").noargs ();
    100.     }
    101.  
    102. }
    103.  
    104. 复制代码

    ②发送含过期时间的消息

    向授权交换机，发送路由为"auth"的消息(指定了业务所需的超时时间) =》发向MqConfig.AUTH_QUEUE_NAME 队列

    1. rabbitTemplate.convertAndSend(MqConfig.AUTH_EXCHANGE_NAME, "auth", "类型：END，信息:{id:1,fromUserId:111,toUserId:222,beginData:20201204,endData:20211104}", message -> {
    2.             /**
    3.              * MessagePostProcessor:消息后置处理
    4.              * 为消息设置属性，然后返回消息，相当于包装消息的类
    5.              */
    6.  
    7.             //业务逻辑：过期时间=xxxx
    8.             String ttl = "5000";
    9.             //设置消息的过期时间
    10.             message.getMessageProperties ().setExpiration (ttl);
    11.             return message;
    12.         });
    13. 复制代码

    ③超时后队列MqConfig.AUTH_QUEUE_NAME会将消息转发至其配置的死信路由"dlx_auth",监听该死信队列即可消费定时的消息

    1.  	/**
    2.      * 授权定时处理
    3.      * @param channel
    4.      * @param message
    5.      */
    6.     @RabbitListener(queues = MqConfig.DLX_AUTH_QUEUE_NAME)
    7.     public void dlxAuthQ(Channel channel, Message message) throws IOException {
    8.         System.out.println ("\n死信原因：" + message.getMessageProperties ().getHeaders ().get ("x-first-death-reason"));
    9.         //1.判断消息类型：1.BEGIN 2.END
    10.         try {
    11.             //2.1 类型为授权到期(END)
    12.             //2.1.1 修改报件办理人
    13.             //2.1.2 修改授权状态为0(失效)
    14.  
    15.             //2.2 类型为授权开启(BEGIN)
    16.             //2.2.1 修改授权状态为1(开启)
    17.             System.out.println (new String(message.getBody (), Charset.forName ("utf8")));
    18.             channel.basicAck (message.getMessageProperties ().getDeliveryTag (),  false);
    19.             System.out.println ("已处理，授权相关信息修改成功");
    20.         } catch (Exception e) {
    21.             //拒签消息
    22.             channel.basicNack (message.getMessageProperties ().getDeliveryTag (), false, false);
    23.             System.out.println ("授权相关信息处理失败, 进入死信队列记录日志");
    24.         }
    25.     }
  • 相关阅读:
    阿里云布置net core 项目
    安卓期末大作业——日记APP
    React核心概念
    MyBatis基本用法 && 什么是自动化测试 && Spring事务和事务传播机制 && 性能测试概念和术语 && Loadrunner安装
    实例036：算素数
    GTK渲染摄像头图像数据
    30 | 工欲善其事必先利其器：后端性能测试工具原理与行业常用工具简介
    MaxKey单点登录认证系统v3.5.10GA发布
    【数据结构与算法】简单排序
    Vue3+移动端适配屏幕+默认横屏展示
  • 原文地址：https://blog.csdn.net/m0_71777195/article/details/128145764