基于Redis Zset 实现延迟队列
延迟队列是一种消息队列的变体，允许消息在指定的时间后才会被消费者处理。基于 Redis 的 有序集合（Zset） 是实现延迟队列最简单、高效的方案之一。其核心原理是利用 Zset 的 score 字段存储消息的执行时间戳，消费者通过轮询获取 score <= 当前时间 的消息进行处理。
一、核心原理
1.1 数据结构定义
| 数据结构 | 作用 | 字段说明 |
|---|---|---|
| Zset（延迟队列） | 存储延迟消息 | member = 消息体（JSON），score = 执行时间戳 |
| Hash（消息详情） | 存储消息完整内容（可选） | 当消息体较大时，Zset 仅存储消息 ID |
1.2 工作流程
二、完整代码实现
2.1 消息实体定义
java
@Data @Builder @NoArgsConstructor @AllArgsConstructor public class DelayMessage<T> { /** * 消息唯一ID（用于幂等和重试） */ private String id; /** * 消息体（业务数据） */ private T data; /** * 延迟时间（毫秒） */ private Long delayMillis; /** * 执行时间戳（毫秒） */ private Long executeTime; /** * 创建时间 */ private Long createTime; /** * 重试次数 */ private Integer retryCount; /** * 最大重试次数 */ private Integer maxRetryCount; /** * 消息业务类型 */ private String topic; public String toJson() { return JSON.toJSONString(this); } public static <T> DelayMessage<T> fromJson(String json, Class<T> dataClass) { return JSON.parseObject(json, new TypeReference<DelayMessage<T>>() {}); } }2.2 延迟队列实现类
java
@Component @Slf4j public class RedisDelayQueue { @Autowired private StringRedisTemplate redisTemplate; private static final String QUEUE_KEY_PREFIX = "delay_queue:"; /** * 添加延迟消息 * @param queueName 队列名称（用于区分不同业务） * @param message 消息体 * @param delayMillis 延迟时间（毫秒） */ public <T> void addMessage(String queueName, T message, long delayMillis) { String queueKey = QUEUE_KEY_PREFIX + queueName; String messageId = UUID.randomUUID().toString(); long executeTime = System.currentTimeMillis() + delayMillis; DelayMessage<T> delayMsg = DelayMessage.<T>builder() .id(messageId) .data(message) .delayMillis(d