【微服务】SpringBoot 整合Redis实现延时使命处置惩罚使用详解 ...

目录

一、前言
二、延迟使命的高频使用场景
三、延迟使命常用办理方案
3.1 Quartz
3.2 DelayQueue
3.2.1 Timer + TimerTask
3.2.2 ScheduledExecutorService
3.3 Redis sorted set
3.4 RabbitMQ
四、Redis实现延时队列使用实战
4.1 Redis Sorted Set 概述
4.1.1 Redis Sorted Set 先容
4.1.2 Redis Sorted Set 主要特点
4.1.3 Redis Sorted Set 应用场景
4.1.4 核心实现思绪
4.2 Jedis 实现延迟队列
4.2.1 前置准备
4.2.2 代码使用演示
4.3 SpringBoot 整合Redis实现延迟队列
4.3.1 添加依靠
4.3.2 添加配置文件
4.3.3 自定义redis序列化类
4.3.4 定义抽象使命
4.3.5 定义使命通知类
4.3.6 定义使命管理器
4.3.7 添加测试接口
4.3.8 效果测试
4.4 使用Redisson 实现延时使命队列
4.4.1 导入依靠
4.4.2 添加配置类
4.4.3 添加自定义消息监听类
4.4.4 自定义使命壅闭队列
4.4.5 添加测试接口
五、写在文末

---

一、前言

在分布式微服务开发中，延迟（延时）使命的场景或需求可以说许多，比如大家熟悉的下单之后答应延后15分钟完成支付，或者某些因为业务的原因需要延迟执行的场景等，同时延迟使命的存在，也让体系或业务有了更多的可扩展空间，本文将通过案例使用演示下如何基于Redis实现延时使命的办理方案。

二、延迟使命的高频使用场景

下面是几种在日常开发中高频接触到的延迟使命场景：

• 订单超时自动处置惩罚
  
  
  
  • 在电商范畴，延迟队列对于处置惩罚订单超时问题至关紧张。一旦用户下单，订单信息便进入延迟队列，并预设超时时长。若用户在此时间内未完成支付，订单信息将由消费者从队列中提取，并执行如取消订单、库存释放等后续使用，高效且自动化。
    
  
  
• 消息超时重发
  
  
  
  • 当消息发送失败，为了保障消息尽可能送到达用户，可以使用延迟重试，即在第一次发送失败之后，让这个动作延迟一定时间之后再次发送。
    
  
  
• 优惠券到期温馨提醒
  
  
  
  • 借助延迟队列，可实现优惠券到期前温馨提醒。将邻近过期的优惠券信息入队，并设定精确延迟时间。时间一到，体系自动提醒用户优惠券的到期日，引导他们实时享用优惠，提升用户体验。
    
  
  
• 异步通知与定时提醒
  
  
  
  • 延迟队列还能用于实现异步通知和定时提醒功能。用户完成使用后，体系将相关通知信息参加队列，并设定发送延时，确保在最佳时机向用户推送通知，既不打扰用户，又能保持信息的时效性。
    
  
  
• 定时提醒
  
  
  
  • 定时（将来的某个时间）发送邮件提醒或其他消息通知等。
    
  
  

   这些场景都需要使命在一段时间后自动执行，因此我们需要一种灵活、高效的办理方案来处置惩罚这类限时使命。
  
三、延迟使命常用办理方案

3.1 Quartz

Quartz 是一个定时使命框架，可通过其提供的API灵活实现定时使命的增编削查等功能，其功能强大之处正是在于用户可根据自身的需求对要执行的各种使命参数举行定制，因此也可以方便的实现延迟使命的配置和使用。

3.2 DelayQueue

JDK中自带了一些技能组件可以用于处置惩罚简朴的使命延时场景。下面简朴先容几种。

3.2.1 Timer + TimerTask

JDK中提供了一组实现延迟队列的API，比如：

• Timer + TimerTask
  
  
  
  • java.util包下，Timer类可以用来安排一个使命在指定的时间执行；
    
  • TimerTask是表示要执行的使命
    
  
  

1. import java.util.Timer;
2. import java.util.TimerTask;
4. public class TimerExample {
5.     public static void main(String[] args) {
6.         Timer timer = new Timer();
7.         
8.         TimerTask task = new TimerTask() {
9.             @Override
10.             public void run() {
11.                 System.out.println("执行延时任务: " + System.currentTimeMillis());
12.             }
13.         };
15.         // 安排任务在延迟1000毫秒后执行
16.         timer.schedule(task, 1000);
17.         
18.         // 也可以周期性执行任务
19.         // timer.scheduleAtFixedRate(task, 1000, 2000); // 每2秒执行一次
20.     }
21. }

复制代码

工作原理：

• 创建一个Timer实例。
  
• 创建一个继续自TimerTask的匿名类，重写run()方法。
  
• 调用schedule()方法安排使命延迟执行。
  

3.2.2 ScheduledExecutorService

相较于Timer，ScheduledExecutorService提供了更强大的功能，可以更加灵活地管理线程和使命。它可以在一个线程池中执行多个定时使命。参考下面的示例：

1. import java.util.concurrent.Executors;
2. import java.util.concurrent.ScheduledExecutorService;
3. import java.util.concurrent.TimeUnit;
5. public class ScheduledExecutorServiceExample {
6.     public static void main(String[] args) {
7.         ScheduledExecutorService executorService = Executors.newScheduledThreadPool(1);
9.         // 安排任务在2秒后执行
10.         executorService.schedule(() -> {
11.             System.out.println("执行延时任务: " + System.currentTimeMillis());
12.         }, 2, TimeUnit.SECONDS);
14.         // 周期性执行任务
15.         // executorService.scheduleAtFixedRate(() -> {
16.         //     System.out.println("周期性任务: " + System.currentTimeMillis());
17.         // }, 1, 2, TimeUnit.SECONDS); // 每2秒执行一次
18.     }
19. }

复制代码

3.3 Redis sorted set

Redis Sorted Set（有序聚集）是一种数据结构，它将元素存储在一个有序的聚集中，每个元素都有一个唯一的分数（score）与之关联。
   具体做法是将使命的执行时间作为分数（score），使命的内容作为值（value），将使命按照执行时间排序存储在有序聚集中。然后周期性地查抄有序聚集中的使命，根据当前时间和使命的执行时间来决定是否执行使命。
  3.4 RabbitMQ

RabbitMQ 提供了死信队列可用于实现延迟使命的执行，具体来说，死信队列是那些主使命举行消息投递的时间投递失败了，但是这些消息不希望丢弃，还需要后续举行处置惩罚时，借助死信队列，就可以将此类消息投递到死信队列，让此类消息延时处置惩罚。

四、Redis实现延时队列使用实战

4.1 Redis Sorted Set 概述


4.1.1 Redis Sorted Set 先容

Redis 的 Sorted Set（有序聚集）是一种非常有用的数据结构，它不光存储了唯一的成员（member），还为每个成员关联了一个分数（score），从而使得成员能够根据分数举行排序。这使得 Sorted Set 在实现排行榜、优先级队列等应用场景中特殊有用。

4.1.2 Redis Sorted Set 主要特点

Redis Sorted Set 主要特点如下：

• 唯一性：Sorted Set 中的成员是唯一的，不能重复。
  
• 有序性：每个成员都有一个与之关联的分数（score），Redis 会根据这个分数对成员举行排序。分数可以雷同，当分数雷同时，成员按照字典序分列。
  
• 高效使用：支持高效的插入、删除和查找使用，时间复杂度为 O(log(N))，此中 N 是聚集中的元素数量。
  
• 范围查询：支持基于分数或排名的范围查询，如获取某个分数区间内的所有成员，或者获取指定排名范围内的成员。
  
• 元素和分数：每个元素都有一个唯一的分数与之关联，分数可以是整数或浮点数。
  
• 插入和删除：元素可以通过 ZADD 命令插入到聚集中，通过 ZREM 命令删除元素。
  
• 分数更新：可以通过 ZINCRBY 命令更新元素的分数。
  

   Sorted Set 的底层实现使用了跳跃表（Skip List）数据结构，跳跃表是一种高效的有序数据结构，它可以在 O(log n) 的时间复杂度内举行插入、删除和查找使用。
  
4.1.3 Redis Sorted Set 应用场景

Sorted Set这种数据结构具备下面一些使用场景：

• 排行榜：使用 Sorted Set 可以轻松实现用户积分排名等功能，实时更新和查询用户的排名情况。
  
• 优先级队列：通太过数来表示使命的优先级，高分使命先被执行。
  
• 缓存体系：例如，可以使用 Sorted Set 来管理缓存项的有用期，方便地找到最早到期的缓存项举行清理。
  
• 交际网络：跟踪用户的关注列表、好友关系等，按照活跃度或其他指标排序展示。
  

4.1.4 核心实现思绪

使用Redis Sorted Set 实现延迟使命时，可以将使命的执行时间作为 Sorted Set 中的 score，然后按时间顺序处置惩罚使命，确保在指定时间执行。
   当需要添加新的延迟使命时，只需将使命的执行时间和内容添加到有序聚集中即可。当然，你可能需要一个后台进程或定时使命来不停地查抄有序聚集，以执行到期的使命。
  接下来使用Redis 的Sorted Set 完成几种方式来实现延时队列。

4.2 Jedis 实现延迟队列


4.2.1 前置准备

在maven的pom文件中引入如下依靠

1. <parent>
2.     <groupId>org.springframework.boot</groupId>
3.     <artifactId>spring-boot-starter-parent</artifactId>
4.     <version>3.2.4</version>
5.     <relativePath/>
6. </parent>
8. <properties>
9.     <docker.image.prefix>dcloud</docker.image.prefix>
10. </properties>
12. <dependencies>
14.     <dependency>
15.         <groupId>redis.clients</groupId>
16.         <artifactId>jedis</artifactId>
17.         <version>3.7.0</version>
18.     </dependency>
20.     <dependency>
21.         <groupId>org.springframework.boot</groupId>
22.         <artifactId>spring-boot-starter-web</artifactId>
23.     </dependency>
25.     <dependency>
26.         <groupId>org.projectlombok</groupId>
27.         <artifactId>lombok</artifactId>
28.         <scope>provided</scope>
29.     </dependency>
31. </dependencies>

复制代码

4.2.2 代码使用演示

在下面的代码中，我们通过在for循环为 delay_queue 这个Sorted Set类型的数据结构（队列）添加了几个元素作为生产使命的数据，中心会相隔一定的时间，只需要在步调运行之后，观察是否延时执行即可

1. import redis.clients.jedis.Jedis;
3. import java.util.Set;
5. public class JedisScoreTest {
7.     private static final String QUEUE_KEY = "delay_queue";
9.     public static void main(String[] args) throws Exception{
10.         Jedis jedis = new Jedis("localhost", 6379);
11.         // 添加元素到延时队列，其中timestamp是希望处理该元素的时间戳
12. //        long timestamp = System.currentTimeMillis() + 5000; // 10秒后处理
13. //        String item = "item1";
14. //        jedis.zadd(QUEUE_KEY, timestamp, item);
15.         for(int i=0;i<5;i++){
16.             String item = "item-"+i;
17.             long timestamp = System.currentTimeMillis() + (1+i) * 2000;
18.             jedis.zadd(QUEUE_KEY, timestamp, item);
19.             Thread.sleep(2000);
20.         }
21.         // 模拟处理延时队列中的任务
22.         pollItems(jedis);
23.         jedis.close();
24.     }
26.     private static void pollItems(Jedis jedis) {
27.         while (true) {
28.             long currentTimestamp = System.currentTimeMillis();
29.             // 获取当前时间之前的所有元素
30.             Set<String> items = jedis.zrangeByScore(QUEUE_KEY, 0, currentTimestamp, 0, 1);
31.             if (items.isEmpty()) {
32.                 // 没有可处理的元素，休眠一会儿继续轮询
33.                 try {
34.                     Thread.sleep(1000);
35.                 } catch (InterruptedException e) {
36.                     Thread.currentThread().interrupt();
37.                     break;
38.                 }
39.             } else {
40.                 // 处理找到的元素
41.                 for (String item : items) {
42.                     // 处理逻辑...
43.                     System.out.println("Processing item: " + item);
44.                     // 移除已处理的元素
45.                     jedis.zrem(QUEUE_KEY, item);
46.                 }
47.             }
48.         }
49.     }
51. }

复制代码

运行上面的代码，通过控制台效果可以看到模仿了延时使命的执行效果

4.3 SpringBoot 整合Redis实现延迟队列


4.3.1 添加依靠

首先，确保你的Spring Boot项目中已经配置好了Redis依靠。在pom.xml文件中添加如下依靠：

1. <dependency>
2.         <groupId>org.springframework.boot</groupId>
3.         <artifactId>spring-boot-starter-data-redis</artifactId>
4. </dependency>

复制代码

4.3.2 添加配置文件

在配置文件中添加下面的信息

1. server:
2.   port: 8081
4. spring:
5.   data:
6.     redis:
7.       host: localhost
8.       port: 6379

复制代码

4.3.3 自定义redis序列化类

为了制止后续使用时出现对象序列化问题，发起在这里添加一个序列化的类

1. package com.congge.web;
3. import org.springframework.boot.autoconfigure.AutoConfigureBefore;
4. import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
5. import org.springframework.cache.annotation.EnableCaching;
6. import org.springframework.context.annotation.Bean;
7. import org.springframework.context.annotation.Configuration;
8. import org.springframework.data.redis.connection.RedisConnectionFactory;
9. import org.springframework.data.redis.core.RedisTemplate;
10. import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
11. import org.springframework.data.redis.serializer.StringRedisSerializer;
13. /**
14. * redis配置
15. */
16. @Configuration
17. @EnableCaching
18. @AutoConfigureBefore(RedisAutoConfiguration.class)
19. public class RedisConfig{
21.     @Bean
22.     public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
23.         RedisTemplate<String, Object> template = new RedisTemplate<>();
24.         template.setConnectionFactory(connectionFactory);
25.         // 使用StringRedisSerializer来序列化和反序列化redis的key值
26.         StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
27.         template.setKeySerializer(stringRedisSerializer);
28.         template.setValueSerializer(stringRedisSerializer);
30.         GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
31.         // Hash的key也采用StringRedisSerializer的序列化方式
32.         template.setHashKeySerializer(stringRedisSerializer);
33.         // Hash的key也采用GenericJackson2JsonRedisSerializer的序列化方式
34.         template.setHashValueSerializer(genericJackson2JsonRedisSerializer);
35.         template.afterPropertiesSet();
36.         return template;
37.     }
38. }

复制代码

4.3.4 定义抽象使命

定义使用使命的常用接口方法，方便上传API调用，参考下面的代码：

1. package com.congge.demo;
2. public interface RedisDelayTask {
3.     /**
4.      *  任务ID
5.      * @return
6.      */
7.     String getId();
9.     /**
10.      *  队列中的值
11.      * @return
12.      */
13.     String getValue();
15.     /**
16.      *  延迟时间（单位：s）
17.      * @return
18.      */
19.     long getDelayTime();
21.     /**
22.      *  任务执行
23.      */
24.     void execute();
25. }

复制代码

1. package com.congge.demo;
4. public abstract class AbstractRedisDelayTask implements RedisDelayTask {
6.     protected String id;
7.     protected String value;
8.     private long delayTime;
10.     public AbstractRedisDelayTask(String id, String value, long delayTime) {
11.         this.id = id;
12.         this.value = value;
13.         this.delayTime = delayTime;
14.     }
16.     @Override
17.     public String getId() {
18.         return id;
19.     }
21.     public void setId(String id) {
22.         this.id = id;
23.     }
25.     @Override
26.     public String getValue() {
27.         return value;
28.     }
30.     public void setValue(String value) {
31.         this.value = value;
32.     }
34.     @Override
35.     public long getDelayTime() {
36.         return delayTime;
37.     }
39.     public void setDelayTime(long delayTime) {
40.         this.delayTime = delayTime;
41.     }
43. }

复制代码

4.3.5 定义使命通知类

参考下面的代码

1. package com.congge.demo;
3. import lombok.extern.slf4j.Slf4j;
5. @Slf4j
6. public class NoticeTask extends AbstractRedisDelayTask {
8.     public NoticeTask(String id, String value, long delayTime) {
9.         super(id, value, delayTime);
10.     }
12.     @Override
13.     public void execute() {
14.         log.info("task execute, {}", this);
15.     }
16. }

复制代码

4.3.6 定义使命管理器

具体处置惩罚使命相关的逻辑

1. package com.congge.demo;
3. import jakarta.annotation.Resource;
4. import org.springframework.beans.factory.InitializingBean;
5. import org.springframework.data.redis.core.RedisTemplate;
6. import org.springframework.data.redis.core.ZSetOperations;
7. import org.springframework.stereotype.Component;
8. import org.springframework.util.CollectionUtils;
10. import java.util.Map;
11. import java.util.Set;
12. import java.util.concurrent.ConcurrentHashMap;
13. import java.util.concurrent.TimeUnit;
15. @Component
16. public class RedisDelayQueueManager implements InitializingBean {
18.     @Resource
19.     private RedisTemplate redisTemplate;
21.     /**
22.      *  任务列表
23.      */
24.     private Map<String, RedisDelayTask> tasks = new ConcurrentHashMap<>();
26.     /**
27.      *  添加延迟任务到队列
28.      * @param task
29.      */
30.     public void addTask(RedisDelayTask task) {
31.         long delayedTime = System.currentTimeMillis() + TimeUnit.MILLISECONDS.convert(task.getDelayTime(), TimeUnit.SECONDS);
32.         boolean addSuccess = redisTemplate.opsForZSet().add(task.getId(), task.getValue(), delayedTime);
33.         if (addSuccess) {
34.             tasks.put(task.getId(), task);
35.         }
36.     }
38.     /**
39.      * 检查并执行任务
40.      */
41.     private void checkAndExecuteTask() {
42.         while (true) {
43.             Set<String> taskIds = tasks.keySet();
44.             for (String taskId : taskIds) {
45.                 // score就是任务要执行的时间点，如果<=当前时间，说明任务该执行了
46.                 Set<ZSetOperations.TypedTuple<String>> tuples = redisTemplate.opsForZSet().rangeByScoreWithScores(taskId, 0, System.currentTimeMillis());
47.                 if (!CollectionUtils.isEmpty(tuples)) {
48.                     for (ZSetOperations.TypedTuple<String> tuple : tuples) {
49.                         // 获取（移除）一个任务并执行
50.                         RedisDelayTask task = tasks.remove(taskId);
51.                         System.out.println( "taskId: " + task.getId());
52.                         System.out.println("taskValue: " + task.getValue());
53.                         if (task != null) {
54.                             task.execute();
55.                             // 任务执行之后从队列中删除
56.                             redisTemplate.opsForZSet().remove(taskId, tuple.getValue());
57.                         }
58.                     }
59.                 }
60.             }
61.         }
62.     }
64.     @Override
65.     public void afterPropertiesSet() throws Exception {
66.         // 另开一个线程执行任务
67.         new Thread(() -> {
68.             checkAndExecuteTask();
69.         }, "redis-delay-task").start();
70.     }
71. }

复制代码

4.3.7 添加测试接口

为了方便背面的测试在这里添加一个测试用的接口

1. package com.congge.web;
3. import com.congge.demo.NoticeTask;
4. import com.congge.demo.RedisDelayQueueManager;
5. import jakarta.annotation.Resource;
6. import org.springframework.web.bind.annotation.GetMapping;
7. import org.springframework.web.bind.annotation.RestController;
9. @RestController
10. public class DelayTaskController {
12.     @Resource
13.     private RedisDelayQueueManager redisDelayQueueManager;
15.     //localhost:8081/add/task
16.     @GetMapping("/add/task")
17.     public void addTask()  {
18.         NoticeTask task = new NoticeTask("task1", "value1", 3);
19.         redisDelayQueueManager.addTask(task);
21.         NoticeTask task2 = new NoticeTask("task2", "value2", 8);
22.         redisDelayQueueManager.addTask(task2);
23.     }
25. }

复制代码

4.3.8 效果测试

运行上面的代码，请求一下接口，通过控制台输出效果可以看到，添加的使命已经延时执行了

4.4 使用Redisson 实现延时使命队列


4.4.1 导入依靠

导入Redisson 依靠包

1.         <dependency>
2.             <groupId>org.redisson</groupId>
3.             <artifactId>redisson</artifactId>
4.             <version>3.33.0</version>
5.         </dependency>

复制代码

4.4.2 添加配置类

自定义一个RedissonClient 的类

1. import org.redisson.Redisson;
2. import org.redisson.api.RedissonClient;
3. import org.redisson.config.Config;
4. import org.springframework.boot.context.properties.ConfigurationProperties;
5. import org.springframework.context.annotation.Bean;
6. import org.springframework.context.annotation.Configuration;
8. @Configuration
9. @ConfigurationProperties(prefix = "spring.data.redis")
10. public class RedissonConfig {
12.     private static final String REDIS_URL  = "redis://localhost:6379";
13.     private static final Integer TIMEOUT = 3000;
15.     @Bean
16.     RedissonClient redissonClient() {
17.         Config config = new Config();
18.         config.useSingleServer()
19.                 .setAddress(REDIS_URL)
20.                 .setTimeout(TIMEOUT);
21.         return Redisson.create(config);
22.     }
24. }

复制代码

4.4.3 添加自定义消息监听类

该类用于处置惩罚队列中的使命消息

1. package com.congge.config;
3. import jakarta.annotation.PostConstruct;
4. import jakarta.annotation.Resource;
5. import org.springframework.stereotype.Component;
7. @Component
8. public class MessageListener {
10.     @Resource
11.     private RedissonConfigQueue redissonConfigQueue;
13.     @PostConstruct
14.     public void init() {
15.         new Thread(() -> {
16.             while (true) {
17.                 try {
18.                     //阻塞等待队列内有可消费延时消息，避免无意义的循环占用CPU
19.                     String string = redissonConfigQueue.blockingQueue().take().toString();
20.                     System.out.println(string);
21.                 } catch (InterruptedException e) {
22.                     throw new RuntimeException(e);
23.                 }
24.             }
25.         }).start();
26.     }
28. }

复制代码

4.4.4 自定义使命壅闭队列

该类用于接收使命并参加到壅闭队列

1. package com.congge.config;
3. import lombok.RequiredArgsConstructor;
4. import org.redisson.api.RBlockingQueue;
5. import org.redisson.api.RDelayedQueue;
6. import org.redisson.api.RedissonClient;
7. import org.springframework.stereotype.Component;
9. @Component
10. @RequiredArgsConstructor
11. public class RedissonConfigQueue {
13.     private final RedissonClient redissonClient;
15.     private final static String queueName = "orderQueue";
17.     /**
18.      * 创建一个阻塞队列
19.      */
20.     public RBlockingQueue<Object> blockingQueue() {
21.         return redissonClient.getBlockingQueue(queueName);
22.     }
24.     /**
25.      * 延时队列
26.      * 阻塞队列放到延迟队列里面，消息放在延迟队列里面，时间到后 消息投递到阻塞队列，消费者监听阻塞队列
27.      */
28.     public RDelayedQueue<Object> getDelayedQueue(RBlockingQueue<Object> blockQueue) {
29.         return redissonClient.getDelayedQueue(blockQueue);
30.     }
32. }

复制代码

4.4.5 添加测试接口

添加一个测试接口方便看效果

1. @Resource
2. private RedissonConfigQueue redissonConfigQueue;
4. //localhost:8081/add/task/v2
5. @GetMapping("/add/task/v2")
6. public String addTaskV2()  {
7.     redissonConfigQueue.getDelayedQueue(redissonConfigQueue.blockingQueue())
8.             .offer("我是消息", 6, TimeUnit.SECONDS);//消息内容,延时时间,时间单位（秒）
9.     return UUID.randomUUID().toString();
10. }

复制代码

五、写在文末

本文通过现实使用案例演示了在微服务中如何实现使命的延时处置惩罚办理方案，可以为技能选型提供参考，希望对看到的同砚有用，本篇到此结束，感谢观看。


免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。