Spring Boot中使用Redis和Lua脚本实现延时队列

在现代微服务架构中，延时队列是一种非常实用的功能。它允许我们将某个操作推迟到未来的某个时间执行，比如发送定时消息、处理定时任务等。本文将介绍如何在Spring Boot中使用Redis和Lua脚本实现延时队列。

一、环境准备

首先，我们需要确保项目中已经引入了Spring Boot和Spring Data Redis的依赖，同时需要有一个Redis服务器的环境。

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

二、Redis延时队列的基本原理

我们可以利用Redis的Sorted Set（有序集合）来实现延时队列。每个任务在插入时，会被赋予一个时间戳，这样可以通过时间戳的顺序来管理任务的执行。Lua脚本的使用可以确保操作的原子性。

三、Lua脚本实现

我们需要编写一个Lua脚本，用于从Sorted Set中获取当前时间应该执行的任务。以下是示例的Lua脚本：

local current_time = ARGV[1]

-- 获取需要处理的任务
local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], 0, current_time)

-- 删除已经处理的任务
if #tasks > 0 then
    redis.call('ZREM', KEYS[1], unpack(tasks))
end

return tasks

在这个Lua脚本中，我们使用ZRANGEBYSCORE命令获取当前时间之前（包括当前时间）所有的任务，并使用ZREM命令将它们从有序集合中删除。

四、Spring Boot实现延时队列

接下来，我们在Spring Boot中实现这个延时队列。首先，需要配置Redis的相关信息：

spring:
  redis:
    host: localhost
    port: 6379

接下来，我们编写一个服务类来实现延时队列的功能。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import javax.annotation.PreDestroy;
import java.time.Instant;
import java.util.Set;

@Service
public class DelayQueueService {

    private final RedisTemplate<String, String> redisTemplate;
    private static final String QUEUE_NAME = "delayQueue";

    @Autowired
    public DelayQueueService(RedisTemplate<String, String> redisTemplate) {
        this.redisTemplate = redisTemplate;
        // 启动任务调度
        startTaskScheduler();
    }

    public void addTask(String task, long delay) {
        long score = Instant.now().toEpochMilli() + delay;
        redisTemplate.opsForZSet().add(QUEUE_NAME, task, score);
    }

    public void startTaskScheduler() {
        new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(1000);  // 每秒检查一次
                    long currentTime = Instant.now().toEpochMilli();
                    String script = "local current_time = ARGV[1] " +
                                    "local tasks = redis.call('ZRANGEBYSCORE', KEYS[1], 0, current_time) " +
                                    "if #tasks > 0 then " +
                                    "    redis.call('ZREM', KEYS[1], unpack(tasks)) " +
                                    "end " +
                                    "return tasks"; 

                    Set<String> tasks = redisTemplate.execute(redisTemplate.getConnectionFactory().getConnection().eval(script, 
                            ReturnType.MULTI, 1, QUEUE_NAME.getBytes(), String.valueOf(currentTime).getBytes()));

                    // 处理任务
                    if (tasks != null) {
                        tasks.forEach(task -> {
                            System.out.println("执行任务: " + task);
                            // 在此处添加任务处理逻辑
                        });
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }).start();
    }
}

五、使用示例

现在我们可以在任何地方通过DelayQueueService的addTask方法添加任务：

@Autowired
private DelayQueueService delayQueueService;

public void someMethod() {
    delayQueueService.addTask("task1", 5000); // 5秒后执行
}

六、总结

通过结合Redis的Sorted Set和Lua脚本，我们能够高效地实现一个延时队列。在实际应用中，可以根据具体需求扩展该功能，比如增加任务的优先级、提出任务处理失败的重试机制等。这种实现是高度可扩展且易于维护的，使用Spring Boot和Redis的结合更是让我们在构建微服务架构时得心应手。