likes
comments
collection
share

项管让我用 Redis 实现延时队列，Excuse me？？

作者站长头像
站长
· 阅读数 54

🏆本文收录于《Spring Boot从入门到精通》，专门攻坚指数提升。

本专栏致力打造最硬核 Spring Boot 从零基础到进阶系列学习内容，🚀均为全网独家首发，打造精品专栏，专栏持续更新中…欢迎大家订阅持续学习。

环境说明：Windows10 + Idea2021.3.2 + Jdk1.8 + SpringBoot 2.3.1.RELEASE

 1. 前言🔥

        提到延时队列，相信各位同学并不会陌生，JDK原生提供了延时队列的使用，当然我们这里介绍的不是这种；在实际的项目中，如果我们有延时队列的场景，可以怎样去实现呢？举一个常见的例子，比如淘宝下单30分钟内，若没有支付，则自动取消订单，这该如何实现？

        bug菌同学扬言：“这个还不简单！，给我10秒钟时间”，上来就是一顿神级操作。

new Thread(() -> {
  // 休眠30分钟
  Thread.sleep(30 * 60 * 1000);
  //执行取消订单
  cancelOrder(); 
}).start();

        “斗宗强者，恐怖如斯！” ，哈哈，让同学们见笑啦！开胃小菜，开胃小菜...

        接下来请认真想一想，如果让我们来实现一个延时队列，可以怎么整？具体该如何实现呢？这将又会是干货满满的一期，全程无尿点不废话只抓重点教，具有非常好的学习效果，拿好小板凳准备就坐！希望学习的过程中大家认真听好好学，学习的途中有任何不清楚或疑问的地方皆可评论区留言或私信，bug菌将第一时间给予解惑，那么废话不多说，直接开整！Fighting！！ 

项管让我用 Redis 实现延时队列，Excuse me？？

2. 环境说明🔥

本地的开发环境：

  • 开发工具：IDEA 2021.3
  • JDK版本： JDK 1.8
  • Spring Boot版本：2.3.1 RELEASE
  • Maven版本：3.8.2

3. 实现思路🔥 

        我们都知道，Redis作为一款高性能的NoSQL数据库，具备快速读写、高并发、数据持久化等特点，非常适合用于实现延迟队列。Redis提供了丰富的数据结构，其中利用Redis的ZSET（有序集合）数据结构就可以实现一个简单的延迟队列。

        Redis的zset数据结构中的每个元素都有一个分数score和一个值value，我们可以将任务的执行时间戳作为score，将任务数据作为value，将任务插入到zset中，每个任务有一个唯一的id(比如订单id)，以及任务执行时间(比如30min)，任务内容(比如订单超时支付系统自动取消)等信息体。然后另起一个线程，该线程会周期性地从zset中取出score最小（即最早要执行的）的任务，如果该任务的score小于当前时间戳，则执行任务，否则等待一段时间再次检查，直到任务可以执行，执行任务后，通过Redis的remove命令删除已经成功执行的任务即可。

4. 详细步骤🔥 

        本文将介绍如何使用Redis的Sorted Set数据结构来实现延迟队列，并提供一个完整的示例代码。同时，我们还将会给出对应的测试用例和测试结果。 

        如下我先给同学们概括下，针对Spring Boot项目，如何利用Redis实现延迟队列的一些实现步骤？同学们请看：

  1. 引入spring-boot-starter-data-redis 和spring-boot-starter-test依赖。
  2. 配置redis。
  3. 创建消息类 DelayMessage
  4. 编写延时队列类 DelayQueue，提供添加消息、删除消息和获取消息方法。
  5. 创建消息处理器 DelayMessageHandler，处理已到期的消息。
  6. 编写测试类 DelayQueueTest，测试延时队列的添加、删除和处理消息功能。
  7. 在Spring Boot启动类中启动延时队列。
  8. 测试用例方法启动测试。

4.1 搭建Spring Boot应用

4.2 引入相关依赖

        引入相关依赖，使用Maven引入Redis 和 Test 的相关依赖，例如：

        <!--test测试类-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <!--集成redis-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

4.3 配置Redis

       配置Redis连接等相关信息，这里同学们可以闭眼cv，示例代码如下：

#redis配置
Spring:
  redis:
    database: 0    #Redis数据库索引（默认为0）
    host: 127.0.0.1  #redis服务器ip，由于我是搭建在本地，固指向本地ip
    port: 6379  #redis服务器连接端口
    password:    #redis服务器连接密码（默认为空）
    # 连接池配置
    jedis.pool:
      max-active: 20      #连接池最大连接数（使用负值表示没有限制）
      max-wait: -1     #连接池最大阻塞等待时间（使用负值表示没有限制）
      max-idle: 10        #连接池中的最大空闲连接
      min-idle: 0         #连接池中的最小空闲连接
      timeout: 1000      #连接超时时间（毫秒）。我设置的是1秒

4.4 创建消息类 DelayMessage

        这里我们定义一个消息类 DelayMessage，包含消息id、消息内容和到期时间(开始执行的时间)等，示例代码如下，仅供参考：

@Data
@AllArgsConstructor
@NoArgsConstructor
public class DelayMessage implements Serializable {
    //切记实例化
    private static final long serialVersionUID = -7671756385477179547L;

    private String id;       //消息id
    private String content;  //消息内容
    private long expireTime; //到期时间
}

4.5 创建延时队列类 DelayQueue

        创建一个延时队列类 DelayQueue，提供添加消息、删除消息和获取消息方法，具体代码实现如下，仅供参考：

@Component
public class DelayQueue {

    private static final String KEY = "delay_queue";

    @Autowired
    private RedisTemplate redisTemplate;

    /**
     * 添加消息到延时队列中
     */
    public void put(DelayMessage message) {
        redisTemplate.opsForZSet().add(KEY, message, message.getExpireTime());
    }

    /**
     * 从延时队列中删除消息
     */
    public void remove(DelayMessage message) {
        redisTemplate.opsForZSet().remove(KEY, message);
    }

    /**
     * 获取延时队列中已到期的消息
     */
    public List<DelayMessage> getExpiredMessages() {
        long minScore = 0;
        long maxScore = System.currentTimeMillis();
        Set<Object> messages = redisTemplate.opsForZSet().rangeByScore(KEY, minScore, maxScore);
        if (messages == null || messages.isEmpty()) {
            return Collections.emptyList();
        }
        List<DelayMessage> result = new ArrayList<>();
        for (Object message : messages) {
            result.add((DelayMessage) message);
        }
        return result;
    }

}

4.6 创建DelayMessageHandler(消息处理器)

        创建一个消息处理器 DelayMessageHandler，添加一个处理已到期的消息，写个定时任务，间隔1s轮询延时队列中是否有已到期的任务，如果获取不为空则进行消息处理逻辑，反之继续轮询，具体代码实现如下，仅供参考：

@Component
public class DelayMessageHandler {

    @Autowired
    private DelayQueue delayQueue;

    /**
     * 处理已到期的消息(轮询)
     */
    @Scheduled(fixedDelay = 1000)
    public void handleExpiredMessages() {
        //获取当前系统时间
        String dateTime = DateFormatUtils.format(new Date(), "hh:MM:ss");

        //扫描任务，并将需要执行的任务加入到任务队列中
        List<DelayMessage> messages = delayQueue.getExpiredMessages();
        System.out.println(dateTime + " 待处理消息：" + messages);
        //处理消息
        if (!messages.isEmpty()) {
            for (DelayMessage message : messages) {
                System.out.println(dateTime + " 处理消息：" + message.getContent());
                //成功处理消息后，便将消息进行移除。
                delayQueue.remove(message);
            }
        }
    }

}

4.7 编写测试类 DelayQueueTest

        编写测试类 DelayQueueTest，测试延时队列的添加、删除和处理消息功能。仅供参考：

@RunWith(SpringRunner.class)
@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
public class DelayQueueTest {
    @Autowired
    private DelayQueue delayQueue;

    @Test
    public void testDelayQueue() {
        //获取当前系统时间
        long newDate = System.currentTimeMillis();
        
        // 添加消息：1秒后到期
        DelayMessage message1 = new DelayMessage("1", "delay message 1", newDate + 1000);
        // 添加消息：3秒后到期
        DelayMessage message2 = new DelayMessage("2", "delay message 2", newDate + 3000);
        // 添加消息：6秒后到期
        DelayMessage message3 = new DelayMessage("3", "delay message 3", newDate + 6000);
        // 添加消息：10秒后到期
        DelayMessage message4 = new DelayMessage("4", "delay message 4", newDate + 10000);

        delayQueue.put(message1);//1s后执行
        delayQueue.put(message2);//3s后执行
        delayQueue.put(message3);//6s后执行
        delayQueue.put(message4);//10s后执行
    }
    
}

4.8 测试案例

        如上编写测试用例，我们先来预估测试用例的执行结果，如下：

  1. 添加消息：1秒后到期，  处理消息：delay message 1
  2. 添加消息：3秒后到期，  处理消息：delay message 1
  3. 添加消息：6秒后到期，  处理消息：delay message 3
  4. 添加消息：10秒后到期， 处理消息：delay message 4

        实际上，我们要进行一波验证，是否按添加的DelayMessage中的expireTime值自动执行？我们重启项目后，对测试用例手动执行，坐等观看控制台输出即可。

项管让我用 Redis 实现延时队列，Excuse me？？

        根据控制台的秒数来判断即可算出在第一次被执行的Message作为参照点，随后将时间延长加上指定执行秒数即可验证；很明显，跟预期结果完全一致。同学们，你们学废了吗？

4.9 小结

本次代码演示中主要使用了Redis的Sorted Set数据结构，包括以下操作：

  1. 添加元素：redisTemplate.opsForZSet().add(KEY, message, message.getExpireTime());
  2. 删除元素：redisTemplate.opsForZSet().remove(KEY, message);
  3. 获取指定score范围内的元素：redisTemplate.opsForZSet().rangeByScore(KEY, startScore, endScore);
  4. ...

        本文重点介绍了如何使用Spring Boot + Redis实现延迟队列，并提供了一个完整的示例代码。同时，我们还介绍了如何使用Sorted Set数据结构来存储延迟队列，并提供了相应的测试用例和测试结果。如果同学们对延迟队列有兴趣，可以在此基础上进一步学习和探究。

... ...

    ok，以上就是我这期的全部内容啦，如果还想学习更多，你可以看看如下的往期热文推荐哦，每天积累一个奇淫小知识，日积月累下去，你一定能成为令人敬仰的大佬。

「赠人玫瑰，手留余香」，咱们下期拜拜~~

5. 热文推荐💭

在入门及进阶之途，我必助你一臂之力，系统性学习，从入门到精通，带你不走弯路，直奔终点；投资自己，永远性价比最高，都这么说了，你还不赶紧来学？？

本文涉及所有源代码，均已上传至github开源，供同学们一对一参考 GitHub传送门，

同时，原创开源不易，欢迎给个star🌟，想体验下被🌟的感jio，非常感谢❗

6. 文末💭

项管让我用 Redis 实现延时队列，Excuse me？？

转载自:https://juejin.cn/post/7270464435298189324
评论
请登录