Java实现Redis延迟队列：从原理到高可用架构

在现代分布式系统中延迟队列是一种至关重要的组件。它允许我们将消息或任务放入队列直到指定的延迟时间到达后才被消费。这种机制广泛应用于订单超时自动取消、支付后定时发送通知、任务重试等场景。虽然RabbitMQ和RocketMQ等专业消息中间件都支持延迟消息但在很多轻量级场景下利用Redis来实现延迟队列是一个更简单、高效的选择。本教程将带你深入理解Redis延迟队列的原理并手把手教你用Java实现一个高可用的延迟队列。核心原理有序集合的巧妙应用实现Redis延迟队列的核心在于利用Redis的**有序集合Sorted Set简称ZSET**数据结构。ZSET中的每个成员Member都关联一个分数Score这个分数是一个浮点数Redis会根据分数对成员进行自动排序。利用这一特性我们可以将任务的执行时间戳作为Score将任务内容或ID作为Member。工作流程如下生产消息入队当需要添加一个延迟任务时计算出它的执行时间当前时间 延迟时间然后使用ZADD命令将其添加到ZSET中。消费消息出队消费者不断轮询ZSET使用ZRANGEBYSCORE命令获取所有分数执行时间小于等于当前时间的任务。这些就是已经“到期”的任务。处理与删除消费者获取到到期任务后进行处理处理完成后使用ZREM命令将其从ZSET中移除。这种实现方式利用了ZSET的自动排序特性使得获取最早到期的任务变得非常高效。基础实现基于Jedis的手动封装为了让你更清晰地理解底层原理我们首先使用Jedis客户端手动实现一个延迟队列。Maven依赖首先确保你的项目中包含了Jedis的依赖。dependency groupIdredis.clients/groupId artifactIdjedis/artifactId version4.3.1/version /dependency核心代码实现下面是一个完整的延迟队列工具类它封装了入队、出队和监听的核心逻辑。import redis.clients.jedis.Jedis; import redis.clients.jedis.JedisPool; import redis.clients.jedis.JedisPoolConfig; import redis.clients.jedis.Tuple; import java.util.Set; public class RedisDelayQueue { private final JedisPool jedisPool; private final String queueKey; public RedisDelayQueue(String host, int port, String queueKey) { this.jedisPool new JedisPool(new JedisPoolConfig(), host, port); this.queueKey queueKey; } /** * 添加延迟任务 * param task 任务内容 * param delayMillis 延迟时间毫秒 */ public void addTask(String task, long delayMillis) { try (Jedis jedis jedisPool.getResource()) { // Score为当前时间 延迟时间 long score System.currentTimeMillis() delayMillis; // Member为任务内容 jedis.zadd(queueKey, score, task); System.out.println(任务已添加: task , 将在 delayMillis ms 后执行); } } /** * 获取并移除一个到期的任务 * return 到期的任务内容如果没有到期任务则返回null */ public String getTask() { try (Jedis jedis jedisPool.getResource()) { long now System.currentTimeMillis(); // 获取所有分数小于等于当前时间的任务即已到期任务 // 我们只取第一个保证一次只处理一个 SetTuple tasks jedis.zrangeByScoreWithScores(queueKey, 0, now, 0, 1); if (tasks ! null !tasks.isEmpty()) { Tuple taskTuple tasks.iterator().next(); String task taskTuple.getElement(); // 获取后立即从队列中移除 jedis.zrem(queueKey, task); return task; } return null; } } /** * 启动消费者监听 */ public void startConsumer() { System.out.println(消费者已启动正在监听队列...); new Thread(() - { while (true) { String task getTask(); if (task ! null) { // 处理任务 handleTask(task); } else { // 没有任务短暂休眠以避免CPU空转 try { Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } } }).start(); } private void handleTask(String task) { System.out.println(正在处理任务: task); // 模拟业务处理耗时 try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println(任务处理完成: task); } public static void main(String[] args) { RedisDelayQueue queue new RedisDelayQueue(localhost, 6379, my_delay_queue); // 启动消费者 queue.startConsumer(); // 添加一些测试任务 queue.addTask(订单1超时取消, 3000); // 3秒后 queue.addTask(订单2超时取消, 5000); // 5秒后 queue.addTask(订单3超时取消, 3000); // 3秒后 } }进阶方案使用Redisson实现生产级队列虽然手动实现有助于理解原理但在生产环境中我们更推荐使用Redisson。Redisson是一个强大的Java驻内存数据网格客户端它为我们提供了开箱即用的RDelayedQueue解决了手动实现中的许多痛点。Redisson的优势简化开发无需手动编写轮询和删除逻辑。原子性保障内部使用Lua脚本保证获取和删除操作的原子性避免并发问题。高性能底层基于发布/订阅模式消费者可以阻塞等待新任务无需频繁轮询大大降低了CPU和网络开销。Maven依赖dependency groupIdorg.redisson/groupId artifactIdredisson/artifactId version3.21.3/version /dependency核心代码实现Redisson的实现方式非常优雅它将延迟队列RDelayedQueue和最终的执行队列RBlockingQueue分离。import org.redisson.Redisson; import org.redisson.api.RBlockingQueue; import org.redisson.api.RDelayedQueue; import org.redisson.api.RedissonClient; import org.redisson.config.Config; import java.util.concurrent.TimeUnit; public class RedissonDelayQueueDemo { public static void main(String[] args) throws InterruptedException { // 1. 创建Redisson客户端 Config config new Config(); config.useSingleServer().setAddress(redis://127.0.0.1:6379); RedissonClient redissonClient Redisson.create(config); // 2. 获取一个阻塞队列它将作为延迟任务的最终目的地 RBlockingQueueString blockingQueue redissonClient.getBlockingQueue(my_task_queue); // 3. 基于阻塞队列创建一个延迟队列 RDelayedQueueString delayedQueue redissonClient.getDelayedQueue(blockingQueue); // 4. 启动消费者线程从阻塞队列中获取任务 new Thread(() - { while (true) { try { // take()方法会阻塞直到有任务可用 String task blockingQueue.take(); System.out.println(消费者收到任务: task); // 处理任务... } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } } }).start(); // 5. 生产者添加延迟任务 System.out.println(添加延迟任务...); delayedQueue.offer(订单A-5秒后执行, 5, TimeUnit.SECONDS); delayedQueue.offer(订单B-2秒后执行, 2, TimeUnit.SECONDS); delayedQueue.offer(订单C-10秒后执行, 10, TimeUnit.SECONDS); // 保持主线程运行 Thread.sleep(20000); // 关闭客户端 redissonClient.shutdown(); } }生产环境的挑战与解决方案在实际应用中我们必须考虑一些边界情况以确保队列的可靠性。如何保证消息不丢失在手动实现中如果消费者获取到任务后在处理完成前进程崩溃那么这个任务就会永久丢失。一个常见的解决方案是引入二次确认机制。消费者获取到期任务后不立即删除而是将其移动到一个“处理中”的临时集合。处理完成后再从“处理中”集合删除。同时需要一个独立的“看门狗”线程定期检查“处理中”集合里的任务是否超时。如果超时则将其重新放回主队列实现自动重试。如何避免并发问题在多线程环境下多个消费者可能同时获取到同一个到期任务导致任务被重复执行。Redisson方案Redisson的RBlockingQueue.take()操作是原子的天然支持多消费者竞争不会有重复消费的问题。手动方案可以使用Redis的ZREMRANGEBYSCORE命令该命令会原子性地移除指定分数范围的成员并返回被移除的成员。这样获取和删除就合并成了一个原子操作。如何监控队列状态运维监控是生产环境不可或缺的一环。你可以轻松地通过Redis命令来监控队列。查看队列长度ZCARD my_delay_queue查看下一个即将执行的任务ZRANGE my_delay_queue 0 0 WITHSCORES查看积压的未到期任务ZCOUNT my_delay_queue (当前时间戳 inf