重学SpringBoot3-集成Redis（八）之限时任务（延迟队列）

重学SpringBoot3-集成Redis（八）之限时任务（延迟队列）

CoderJia 41 2024-10-08

在分布式系统中，延迟任务（或限时任务）是一种常见的需求，通常用于实现延迟执行、定时处理或消息超时等场景。Redis 作为高性能的内存数据库，具备非常灵活的 Sorted Set（有序集合） 数据结构，可以很容易地实现延迟队列，满足限时任务的需求。

在本篇文章中，我们将介绍如何通过 Redis 和 Spring Boot 3 来实现 限时任务（也称为延迟任务或延迟队列），让你能够轻松管理任务的延时执行。

1. 延迟任务的场景

延迟任务的应用场景非常广泛，包括但不限于以下场景：

  1. 订单超时取消：用户下单后未支付，超过一定时间自动取消订单。
  2. 消息超时重发：当消息发送失败，可以延迟重试。
  3. 定时提醒：比如发送通知或定时邮件。

这些场景都需要任务在一段时间后自动执行，因此我们需要一种灵活、高效的解决方案来处理这类限时任务。

2. Redis Sorted Set基本原理

Redis Sorted Set（有序集合）是一种数据结构，它将元素存储在一个有序的集合中，每个元素都有一个唯一的分数（score）与之关联。Sorted Set 的基本原理如下：

  1. 元素和分数：每个元素都有一个唯一的分数与之关联，分数可以是整数或浮点数。
  2. 有序集合：元素按照分数的大小顺序存储在集合中，分数越小的元素越靠近集合的头部。
  3. 唯一性：集合中每个元素的分数必须是唯一的，如果两个元素的分数相同，则后一个元素会覆盖前一个元素。
  4. 插入和删除：元素可以通过 ZADD 命令插入到集合中，通过 ZREM 命令删除元素。
  5. 范围查询：可以通过 ZRANGE 命令查询集合中某个范围内的元素，范围可以是分数范围或索引范围，ZRANGEBYSCORE 可以根据 score 范围查找元素。
  6. 分数更新：可以通过 ZINCRBY 命令更新元素的分数。

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

3. 使用 Redis Sorted Set 实现延迟队列

在实现延迟任务时，我们可以将任务的执行时间作为 Sorted Set 中的 score，然后按时间顺序处理任务，确保在指定时间执行。

3.1. 引入依赖

首先，在 pom.xml 中引入 Redis 相关依赖，相关配置请参考重学SpringBoot3-集成Redis（一）基本使用：

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>

3.2. 配置 Redis

在 application.yml 中配置 Redis 的连接信息：

spring:
  data:
    redis:
      host: 1.94.26.81
      port: 6379            # Redis 端口
      password: redis123456 # 如果有密码可以在这里配置
      lettuce:
        pool:
          max-active: 100    # 最大并发连接数
          max-idle: 50       # 最大空闲连接数
          min-idle: 10       # 最小空闲连接数

3.3. 延迟队列的任务存储和处理

接下来，我们通过 Redis 的 Sorted Set 来存储任务，并定时检查任务是否到期。

3.3.1 任务存储

每当有新的任务需要延迟执行时，我们将其加入到 Redis 的 Sorted Set 中，score 为该任务的执行时间戳。

package com.coderjia.boot310redis.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * @author CoderJia
 * @create 2024/10/7 下午 05:10
 * @Description
 **/
@Service
public class DelayedTaskService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String DELAYED_QUEUE_KEY = "delayedQueue";

    // 添加任务到延迟队列
    public void addTaskToQueue(String taskId, long delayInSeconds) {
        long executeTime = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(delayInSeconds);
        redisTemplate.opsForZSet().add(DELAYED_QUEUE_KEY, taskId, executeTime);
        System.out.println("Added task " + taskId + " to the queue, will be executed in " + delayInSeconds + " seconds.");
    }
}

3.3.2 任务处理

为了定期检查是否有任务到期，我们使用 Spring 的 @Scheduled 注解创建一个定时任务，定时从 Redis 中获取即将到期的任务并执行。

package com.coderjia.boot310redis.config;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;

import java.util.Set;

/**
 * @author CoderJia
 * @create 2024/10/7 下午 05:10
 * @Description
 **/
@Slf4j
@Service
public class DelayedTaskProcessor {

    @Autowired
    private StringRedisTemplate redisTemplate;

    private static final String DELAYED_QUEUE_KEY = "delayedQueue";

    // 定时任务，检查是否有任务到期
    @Scheduled(fixedRate = 5000) // 每隔5秒执行一次
    public void processDelayedTasks() {
        long currentTime = System.currentTimeMillis();
        Set<String> tasks = redisTemplate.opsForZSet().rangeByScore(DELAYED_QUEUE_KEY, 0, currentTime);

        if (tasks != null && !tasks.isEmpty()) {
            for (String taskId : tasks) {
                // 执行任务
                executeTask(taskId);
                // 从队列中移除已执行的任务
                redisTemplate.opsForZSet().remove(DELAYED_QUEUE_KEY, taskId);
                log.info("Task " + taskId + " is executed.");
            }
        }
    }

    // 模拟任务执行
    private void executeTask(String taskId) {
        // 执行任务
        log.info("Task " + taskId + " is executing...");
    }
}

在这个示例中：

  • addTaskToQueue(String taskId, long delayInSeconds) 方法将任务加入到 Redis 的 Sorted Set 中，延迟 delayInSeconds 秒执行。
  • @Scheduled(fixedRate = 5000) 每隔 1 秒扫描一次 Redis，查找是否有任务的执行时间已到期。如果有，则执行该任务，并从队列中移除。

3.4. 测试效果

在你的业务逻辑中调用上面创建的 addTaskToQueue 方法添加任务到延迟队列中去。

package com.coderjia.boot310redis.demos.web;

import com.coderjia.boot310redis.service.DelayedTaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author CoderJia
 * @create 2024/10/7 下午 05:18
 * @Description
 **/
@Slf4j
@RestController
public class DelayedTaskController {

    @Autowired
    private DelayedTaskService delayedTaskService;

    @GetMapping("/addDelayedTask")
    public String addDelayedTask(@RequestParam("taskId") String taskId, @RequestParam("delay") Long delay) {
        log.info("Adding task to the queue, taskId:{}, delay:{}", taskId, delay);
        delayedTaskService.addTaskToQueue(taskId, delay);
        return "Added task success";
    }
}

3.4.1 测试添加任务

执行以下请求添加 3 条任务到延迟队列中，

http://localhost:8080/addDelayedTask?taskId=3&delay=20

添加任务

任务执行

新增2条任务，先添加任务的后执行，测试效果。

先添加任务的后执行

3.4.2 存储格式

存储格式

4. 优化建议

4.1. 任务幂等性

在分布式环境中，任务可能会被多个节点同时执行。确保任务的幂等性非常重要，可以通过 Redisson 分布式锁来保证同一时刻只有一个节点在执行任务。

        // 获取分布式锁
        RLock lock = redissonClient.getLock("scheduledTaskLock");

        try {
            // 尝试获取锁，最多等待 1 秒，锁定时间 10 秒
            if (lock.tryLock(1, 10, TimeUnit.SECONDS)) {
                try {
                    // 执行任务
                    System.out.println("Executing distributed scheduled task...");
                } finally {
                    lock.unlock(); // 释放锁
                }
            } else {
                System.out.println("Another instance is executing the task...");
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

4.2. 持久化

延迟队列可以与持久化存储结合起来，确保任务在 Redis 失败或重启时不会丢失。可以使用 Redis 持久化功能或将任务信息存储在数据库中。

4.3. 高并发处理

对于大量延迟任务，可以通过增加 Redis 集群的规模或使用更高效的数据结构来提升处理性能。

5. 总结

通过 Redis Sorted Set 和 Spring Boot 3，我们可以轻松实现限时任务的调度。Redis 的高性能和有序集合特性为我们提供了实现延迟队列的基础，而 Spring Boot 的定时任务调度则帮助我们定期处理这些任务。

在实际场景中，限时任务的应用非常广泛，比如订单超时处理、消息重发等场景，借助 Redis 我们可以有效管理这些延迟任务并确保系统的高效运行。

希望这篇文章能够帮助你更好地理解如何使用 Spring Boot 3 与 Redis 实现延迟队列。如果你在项目中遇到了相关问题，欢迎在评论区分享你的问题与经验。

SpringBoot3