Redisson延迟队列实现原理与生产实践：从定时扫表到精准任务调度

做延迟任务这个需求我踩过不少坑最开始和大多数人一样第一个想到的就是定时任务扫表后来在订单超时关闭、自动确认收货这类场景里发现暴力扫表带来的延迟和数据库压力实在不好看才开始认真调研延迟队列方案。选型比完一圈后我在实际项目里用了 Redisson 的 RDelayedQueue整体接入成本低稳定性也够。1. 为什么最后选了Redisson而不是自己写定时器1.1 扫表方案看起来简单但越往后越难受先说说大家最常用的定时任务扫表方案。表里加一个状态字段和到期时间然后写个 xxl-job 之类的定时任务每隔一分钟扫一次把到期的数据捞出来处理。这个方案在数据量小、时效要求不高的场景下确实没问题实现成本几乎为零。但一旦业务量上来问题就跟着来了扫表时间间隔不好定。间隔太短数据库压力大高峰期一次扫出几千条过期数据处理不过来间隔太长业务延迟变大用户明明已经付款了 15 分钟系统还显示待支付。在电商场景里超时未支付订单的关单延迟直接就是用户体验问题。数据库会越来越慢。随着订单量增长表中积累的历史数据越来越多如果不做归档每一次扫描都相当于一次全表范围查询索引再优化也架不住高频扫。不够灵活。有些延迟任务的触发时间不是固定的比如用户在下单后 10 到 30 分钟任意时间点取消扫表方案很难按每个任务的独立时间精准触发。所以扫表不是不能做而是它更适合那种不需要精确到秒、数据量可控的后台任务比如定期给老用户发优惠券的提醒、清理过期日志。1.2 对比一轮后RDelayedQueue 综合成本最低当时团队内部其实已经有 RabbitMQ但它自带的延迟消息插件在部分运维环境里没启用而且为了一个关单需求去推动中间件升级跨部门沟通成本太高。消息队列的延迟队列功能业界已经做得很成熟但很多团队不一定有独立 MQ 集群单独为了延迟任务去引入一套重型中间件有点小题大做。延迟队列常用的实现思路大致有三种方案实现思路优点缺点数据库轮询定时任务扫描到期记录简单好理解延迟不可控、DB压力大消息队列延迟插件RabbitMQ延迟插件、RocketMQ定时消息可靠、吞吐高、生态好依赖中间件版本和运维限制多Redis 有序集合利用 ZSet 按时间排序客户端轮询到期任务轻量、精确到秒、无额外依赖可靠性依赖 Redis 持久化最后一种方案我们当然可以自己写无非就是往 ZSet 里塞任务score 设为到期时间再起一个守护线程循环取数据。但自己写的轮子要处理的问题很多客户端崩溃后任务还在不在多个消费者实例同时取任务会不会重复Redis 里存的对象怎么序列化这些坑逐一踩下来调试成本并不低。Redisson 的 RDelayedQueue 其实做的就是这件事它把基于 ZSet 的延迟队列封装成了开箱即用的Java API底层还处理了分布式锁、阻塞队列、序列化这些细节所以我最后选了它。这里也说清楚它不是万能的后面有一节专门讲边界。1.3 它到底适合解决什么问题用一句话概况Redis 里先存一个延迟队列到期后自动往另一个就绪队列丢消费者阻塞从就绪队列里拿任务处理。适合用它的场景有这么几个特征延迟时间可以很长比如 30 分钟、24 小时甚至几天对时间敏感度是秒级到分钟级不需要毫秒级任务量不是每秒几万条那种超高吞吐团队里已经用了 Redis不想再引入额外中间件任务丢失需要可接受或能有补偿兜底比如关单场景有定时任务兜底扫描。如果你的延迟时间很短毫秒级且吞吐极高Redisson 延迟队列不是最优解内存型的时间轮或者直接塞消息中间件会更合适。2. RDelayedQueue底层源码拆解它到底是怎么做到延迟的用起来简单但理解它底层是怎么工作的对排查问题和做调优非常重要。我当初就是没看原理直接上结果遇到一个诡异问题排查了很久后面会讲。2.1 核心结构ZSet List Hash 的组合玩法Redisson 的 RDelayedQueue 在 Redis 里依赖三种数据结构配合ZSet用于存放延迟任务的元数据member 是一个随机生成的UUIDscore 是任务的到期时间戳Hash存放任务的实际数据内容key 就是 ZSet 里那个 UUIDList真正给消费者阻塞获取的队列延迟任务到期后会被转移到这里。这三者的协作流程我用一个订单超时关闭的例子走一遍假设用户下单时间是T订单超时时间为 30 分钟。调delayedQueue.offer(order, 30, TimeUnit.MINUTES)之后Redisson 会往 Hash 里存一份订单数据同时往 ZSet 加一条记录score 是T 1800秒。此时任务还在延迟队列里躺着消费者看不到。单独在消费者侧调研的时候如果只调take()从队列里取你会发现它拿不到任务因为延迟任务还没转移过去。2.2 到期转移每个客户端都有个定时器在搬砖Redisson 不是靠 Redis 的过期键机制来实现延迟的因为 Redis Key 过期是基于惰性删除加定时删除不可靠而且任务删除时做不到把值安全地推送到另一个队列。真正的实现是客户端侧轮询转移。具体来说Redisson 在创建延迟队列的时候会维护一个基于 Netty 的定时任务调度器TimeoutScheduler默认每隔一小段时间旧版本是 300ms个别版本策略有调整就去 ZSet 里看有没有 score 小于当前时间戳的成员。如果有就执行这么几个动作从 ZSet 里按 score 范围取出已到期的成员用成员对应的 UUID 去 Hash 里取出真实的任务数据把取出的数据 push 到 List 队列里从 ZSet 和 Hash 里删除这条记录任务正式从延迟队列转移到就绪队列。这整个过程在 Redisson 里有DelayedQueue内部类专门负责核心方法是scheduleTask和processExpired这类逻辑。2.3 消费路径阻塞队列 监听器两种方式消费者拿任务是从 List 队列里取Redisson 给 RDelayedQueue 配套返回了一个RBlockingQueue这个队列支持阻塞获取也就是take()也有带超时的poll(long timeout, TimeUnit unit)。常见的有两种写法第一种是直接在主线程里 while 循环阻塞取RBlockingQueueOrderMessage blockingQueue redissonClient.getBlockingQueue(ORDER_DELAY_QUEUE); while (true) { OrderMessage message blockingQueue.take(); // 执行业务 handleOrderClose(message); }第二种是注册监听器内部其实还是走的阻塞取只是把逻辑封装成了事件回调RBlockingQueueOrderMessage blockingQueue redissonClient.getBlockingQueue(ORDER_DELAY_QUEUE); RDelayedQueueOrderMessage delayedQueue redissonClient.getDelayedQueue(blockingQueue); delayedQueue.addListener((MessageListenerOrderMessage) (messageId, message) - { handleOrderClose(message); });这两种方式内部本质一致都是用阻塞队列从 Redis List 里拉取。推荐哪种看团队习惯我倾向于 while take 的写法因为后面要加大分布式锁、幂等处理都比较自然。2.4 可靠性边界延迟队列不是消息队列RDelayedQueue 在 Redis 里面是持久化的只要 Redis 本身没丢数据任务放进队列后不会因为客户端重启而消失。到期转移是靠客户端触发的所以还有个细节如果你把生产者和消费者全部停掉在这期间到期的任务会堆积在 ZSet 里等客户端恢复后定时器会一次性把所有到期任务转移过去不会丢但会出现集中爆发处理的情况。另一个边界是Redisson 的延迟队列不提供消息确认机制。消费者take()出来之后任务就已经从 List 队列移除了如果此时消费者进程崩溃任务就真的丢了。对订单关单这种允许定时任务兜底的场景问题不大但如果你拿它来做支付回调通知这种不能丢的消息最好在外面包一层业务表记录状态。3. 完整接入实录从依赖到 Consumer 全流程代码部分我把当时项目的实现简化后放出来基本可以直接抄几个容易出错的地方我会专门标注。3.1 依赖引入和客户端初始化项目用的是 Spring Boot 2.7Redis 用的 Redisson 3.17 左右。dependency groupIdorg.redisson/groupId artifactIdredisson-spring-boot-starter/artifactId version3.17.7/version /dependencyRedisson 官方提供了 spring boot starter它会自动装配 RedissonClient但是默认配置读取的是 application 配置里的spring.redis或spring.data.redis。推荐显式用 config 构建客户端这样配置项更直观Configuration public class RedissonConfig { Bean(destroyMethod shutdown) public RedissonClient redissonClient() { Config config new Config(); config.useSingleServer() .setAddress(redis://127.0.0.1:6379) .setPassword(123456) .setConnectionPoolSize(8) .setConnectionMinimumIdleSize(4) .setIdleConnectionTimeout(10000) .setConnectTimeout(5000) .setTimeout(3000) .setRetryAttempts(3) .setRetryInterval(1000); return Redisson.create(config); } }如果用集群模式把useSingleServer换成useClusterServers即可。这里有几个参数后来在生产上验证过比较重要timeout和retryAttempts决定了 Redis 抖动时任务的稳定性建议 timeout 不要低于 3000ms不然大对象序列化容易超时connectionMinimumIdleSize保持有 4 个空闲连接能避免突发流量时新建连接的开销。3.2 生产者把订单消息送进延迟队列先定义消息体这里有个大坑就是序列化问题后面踩坑章节细说。消息体必须实现Serializable接口并且字段要尽量简单避免内部类导致反序列化失败public class OrderMessage implements Serializable { private static final long serialVersionUID 1L; private Long orderId; private Long userId; private Integer orderStatus; // 省略 getter/setter }生产者的逻辑很简单注入 RedissonClient然后组装队列Component public class OrderDelayProducer { private final RedissonClient redissonClient; public OrderDelayProducer(RedissonClient redissonClient) { this.redissonClient redissonClient; } public void sendOrderCloseTask(OrderMessage message, long delay, TimeUnit unit) { RBlockingQueueOrderMessage blockingQueue redissonClient.getBlockingQueue(ORDER_CLOSE_DELAY_QUEUE); RDelayedQueueOrderMessage delayedQueue redissonClient.getDelayedQueue(blockingQueue); delayedQueue.offer(message, delay, unit); } }这里有个非常重要的细节每次调用时我用redissonClient.getBlockingQueue重新获取了一次队列对象。Redisson 的 API 设计里这个方法每次返回的是同一个分布式队列的客户端视图底层是同一个 Redis key调用多次不会有副作用。但有一个和版本相关的坑如果你在应用启动时调了一次getBlockingQueue然后在另一个地方调了一次同名getDelayedQueue中间隔了很久有的版本客户端内部基于事件订阅去感知队列状态可能因为队列重复注册导致监听器无法生效。为了避免这些诡异问题我这里直接把延迟队列对象设计成单例用Component持有。另一个版本相关的规范是一个 RDelayedQueue 实例只能绑定一个 RBlockingQueue不能拿同一个延迟队列同时塞给两个不同的阻塞队列否则后注册的会覆盖前一个的到期转移目标。这个问题在低版本里遇到过升级到 3.16 后官方加了校验逻辑会更明显。3.3 消费者两种姿势取任务推荐独立线程跑消费者和业务线程池分开。我当时是单独抽了一个 DelayQueueConsumerSpring 容器启动后在PostConstruct里启动它Component public class OrderDelayConsumer { private static final Logger log LoggerFactory.getLogger(OrderDelayConsumer.class); private final RedissonClient redissonClient; private final OrderCloseHandler orderCloseHandler; public OrderDelayConsumer(RedissonClient redissonClient, OrderCloseHandler orderCloseHandler) { this.redissonClient redissonClient; this.orderCloseHandler orderCloseHandler; } PostConstruct public void start() { Thread consumerThread new Thread(this::consume, order-delay-consumer); consumerThread.setDaemon(true); consumerThread.start(); } private void consume() { RBlockingQueueOrderMessage blockingQueue redissonClient.getBlockingQueue(ORDER_CLOSE_DELAY_QUEUE); while (!Thread.currentThread().isInterrupted()) { try { OrderMessage message blockingQueue.take(); handle(message); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; } catch (Exception e) { log.error(处理延迟任务失败, e); } } } private void handle(OrderMessage message) { // 这里做幂等校验查数据库看订单是否已经是终态 // 是的话直接跳过不是的话执行关单逻辑 orderCloseHandler.closeOrder(message.getOrderId()); } }因为 RBlockingQueue 的take()是阻塞操作Redis 连接线程会阻塞住所以必须用独立线程跑不然占着 Servlet 线程太浪费。加了Thread.currentThread().isInterrupted()是为了应用优雅停机时能退出循环。这里有同学可能会问为什么不直接用addListener我对比下来监听器方式在一些异常场景下会静默停止消费比如 Redis 连接断掉重建后监听器的 EventListener 内部线程可能已经挂了应用却没有重启。而 while take 的写法遇到连接断开时take()会直接抛异常虽然这也不是我们希望看到的但至少日志里能看到方便告警。3.4 一次完整的测试验证我在本地起了一个 Redis写了个简单测试模拟下单 5 秒后自动关单Test public void testDelay() throws InterruptedException { OrderDelayProducer producer new OrderDelayProducer(redissonClient); OrderMessage message new OrderMessage(); message.setOrderId(1001L); message.setUserId(2001L); message.setOrderStatus(0); long start System.currentTimeMillis(); producer.sendOrderCloseTask(message, 5, TimeUnit.SECONDS); // 消费者拿到的时刻 OrderMessage received blockingQueue.take(); long cost System.currentTimeMillis() - start; System.out.println(收到消息距离发送时间: cost ms); Assert.isTrue(cost 5000, 延迟时间不足5秒); }正常运行后日志里会看到收到消息距离发送时间: 5123ms说明队列生效。注意这里会有 100ms 左右的误差这是 Redisson 轮询转移机制决定的后面讲为什么。4. 生产环境部署需要注意的几个关键点4.1 默认轮询间隔决定延迟精度Redisson 底层不是像 RabbitMQ 那样完全事件驱动而是每个节点启动一个定时任务每隔一定时间扫一次 ZSet。这个间隔在不同版本中不太一样有的版本是 300ms有的版本是按秒级运行。如果你拿到的延迟任务对精度要求是准点触发比如 10 秒后必须刚好是 10 秒那你需要知道这个误差的绝对值是轮询间隔 网络传输时间。官方在设计时并没有承诺毫秒级精度如果你需要秒级精度只能靠两条路升级到较新的 Redisson 版本它的内部调度更轻量自己封装一层把延迟时间按到期时间精确到下个 100ms 的槽位这样虽然也是轮询但粒度更细。我之前有一个需求是活动结束后立即发放奖励如果用延迟队列默认配置实际发放时间可能会晚 200ms 到 500ms这个误差用户完全无感。但是如果你做的是库存自动释放误差几百毫秒其实也没影响因为释放的瞬间恰恰是并发高峰稍微晚一点点还能起到削峰的作用。4.2 序列化配置最容易踩的雷区Redisson 默认的序列化方式是 FST这是个高性能的 Java 序列化库。但 FST 有个问题如果你改过消息类的字段名或者包名老任务反序列化直接报错因为序列化数据里的类描述信息变了。更稳妥的做法是统一使用 JSON 序列化尤其是消息类结构可能会变动的时候Configuration public class RedissonConfig { Bean(destroyMethod shutdown) public RedissonClient redissonClient() { Config config new Config(); config.useSingleServer() .setAddress(redis://127.0.0.1:6379); // 使用 Jackson JSON 序列化 config.setCodec(new org.redisson.client.codec.StringCodec()); // 或者 config.setCodec(new JsonJacksonCodec()); return Redisson.create(config); } }JsonJacksonCodec底层用的是 Jackson消息体会被序列化成 JSON 字符串存在 Hash 里排查问题的时候直接去 Redis 里看值肉眼可读比 FST 的二进制好排查得多。如果消息类字段类型是 LocalDateTime 这类 JDK8 时间类型记得在 ObjectMapper 里注册 JavaTimeModule否则序列化会抛异常。还有一点消息类里不要放接口类型字段。比如你定义了一个Object payloadJSON 反序列化的时候会丢失真实类型取出来是个 LinkedHashMap强转会报错。确保字段类型都是具体类。4.3 多个消费者实例是并行消费还是重复消费Redisson 的阻塞队列是基于 Redis List 的多个消费者实例同时take()时Redis 的 BRPOP 机制会保证每条消息只被一个消费者取走不会重复给到两个实例。这一点比很多自己写的轮子可靠。但要注意的是 RDelayedQueue 的到期转移机制是每客户端各跑各的。如果你起了 3 个应用实例这 3 个实例各自都会去扫 ZSet那么问题来了同一个到期任务会不会被 3 个实例同时转移然后重复投递Redisson 在实现的时候为了避免这个问题在转移逻辑里加了分布式锁基于 Redisson 的 RLock锁的 key 是延迟队列名。同一时刻只有一个客户端能拿到锁去执行转移所以同一任务不会被重复投递。这个机制在大部分场景下是可靠的。不过分布式锁本身也有主从切换时的极端不一致风险RedLock 也解决不了因此对于绝不能重复的任务消费端还是要做幂等。最简单可靠的幂等就是在业务表里用唯一索引约束或者先查再改时加事务锁。4.4 监控别漏掉 ZSet 堆积量用 Redisson 延迟队列最直观的监控指标就是延迟队列对应 ZSet 的长度。如果某个时刻看到 ZSet 成员数量飙升说明有很多任务在排队还没到期这是正常现象但如果过了应该到期的时间ZSet 数量还是不减就要警惕了。这种情况一般是消费者挂了或者 Redisson 的到期转移定时器出了问题。到时候排查思路是看 ZSet 里 score 最小的记录如果 score 已经是过去的时间说明有到期未转移任务看消费者进程日志确认 take 循环是否还在运行看 Redis 的连接状态如果是从库或不可写转移逻辑会一直失败。我当时的监控做法是用 XXL-JOB 每隔 5 分钟跑一个任务统计一下所有延迟队列 ZSet 的过期任务数和总任务数超过阈值直接企业微信告警。这样即使消费者静默挂掉也能在业务受损前发现。5. 踩坑实录我遇到过的真实问题篇幅原因我只挑有代表性的几个每一个都是我实际在线上踩过的。5.1 “任务消失之谜”其实就是消费线程挂了有一次线上反馈订单超时后没有自动关单排查了半天 Redis 里的数据ZSet 里的任务正常、List 里的任务也被取走了就是没有执行关单。最后发现是消费线程因为一次反序列化异常被业务代码抛出后没有 catch 住线程直接终止了。这是我前面强烈建议用 while take 加 catch Exception 的原因。业务代码里任何不可预料的异常都不能让它出 while 循环否则这个消费者进程表面上活着实际已经不消费了。加一行日志配上告警能第一时间发现问题。5.2 延迟时间不准实际比预期晚了 40 秒升级 Redisson 版本后有一天测试反馈说延迟 30 秒的任务实际 70 秒才触发。查了很久发现是低版本的getDelayedQueue在调用时如果队列不存在内部的调度任务初始化时机比较晚跨越了类加载或者 Spring 容器初始化的某个阶段导致第一次扫描延迟时间特别长。另外还有一个容易忽略的点如果你在 Spring 容器还没完全启动完成时就去delayedQueue.offer()而消费者线程又在容器启动后才开始跑那么整个期间 Redis 里没有任何客户端在执行定时转移任务任务只能等消费者跑起来才被扫描。也就是说应用启动期间不会发生任何到期转移。解决办法很简单确保消费者线程在有任务生产者之前已经初始化好。如果拿 Spring 的ApplicationReadyEvent来启动消费者线程比PostConstruct更稳妥因为前者确保所有 Bean 都准备好了。5.3 序列化报错改了字段名老任务全部取不出来有一次我们把订单消息类的 orderId 字段改成了 bizOrderId发版后线上直接报反序列化异常一堆任务卡死在队列里。因为旧数据是用 FST 序列化的字段名变更后 FST 读不出来。这个问题的教训是两方面的消息类的字段一旦上线尽量不要改名字要新增字段就用新的类版本号做好兼容最好从一开始就用 JSON 序列化至少报错的时候能看清楚 Redis 里的原始数据长什么样。我现在的做法是所有延迟任务的消息类统一用 JSON 序列化并且在类上标注JsonTypeInfo或者定义messageType字段这样以后升级消息结构的时候还能兼容老数据。5.4 消息重复消费消费端动作用了分布式锁但锁失效有个同事写的支付结果延迟通知消费端加了 Redisson 的分布式锁做防重但用的是锁的默认过期时间 30 秒。结果有一次 Redis 主从切换锁的续期线程还没来得及把锁续上主节点上的锁丢了两个消费者同时执行业务导致用户收到了两条重复通知。这个问题的本质是分布式锁本身在 Redis 主从切换时就不是绝对安全的。延迟队列场景下想让消息不重复最可靠的办法是消息幂等而不是锁幂等。也就是说处理任务时先查一下业务的终态如果已经处理过就直接丢弃。延迟队列的任务量一般不会特别大多一次查询对数据库完全没压力但换来的可靠性提升很明显。5.5 高峰期任务量暴增转移逻辑出现瓶颈有一年大促秒杀活动结束后会一次性产生一百多万条延迟任务发奖、短信通知、过期清理全塞进同一个延迟队列。结果 Redis 单实例 CPU 飙到 90%因为 ZSet 和 Hash 是同一个 key所有压力都在一个分片上。这种情况不要指望通过调优参数解决合理的做法是给延迟队列做分片。比如按订单 ID 对 10 取模建 10 个队列生产者和消费者按同样的取模规则路由把压力分摊到不同的 Redis slot。String queueName ORDER_CLOSE_DELAY_QUEUE_ (orderId % 10);这样虽然代码里多了一点路由逻辑但单队列的压力立刻降下来了排查问题的时候也能根据订单号快速定位它在哪个队列。6. 延迟任务的取消与更新很多人忽略的需求延迟任务发出去后实际业务里常常有取消需求。比如用户下单后 30 分钟未支付自动关单但用户在第 10 分钟主动取消了订单这时候消息还在延迟队列里到点还是会触发关单逻辑。Redisson 的 RDelayedQueue 不像 RabbitMQ 那样按消息 ID 删除那么方便。它本身没有提供按业务 ID 删除任务的方法但也不是完全没办法。我的做法通常有两种第一种是消费时二次校验。到了触发时间后消费者先去查业务表状态如果订单已经是取消状态直接跳过。这是最保险的兜底无论任务怎么堆积都能兜住。第二种是借助队列名设计。如果同一业务类型的任务可以按用户或订单分队列取消时直接把整个队列删掉redissonClient.getDelayedQueue(blockingQueue).destroy()不过这个操作比较暴力适合任务粒度比较大、一个用户同一时间只有一个任务在飞的场景。最推荐的做法还是第一种因为延迟任务的消费本质上是异步的任何前置条件的判断都不可避免存在时间差消费时查一次库做终态判断代价最小、效果最可靠。事实上这个思路适用于所有延迟任务场景不管是用消息队列还是 Redis 实现消费时幂等校验都是必做的。写在最后用 Redisson 实现延迟队列本质上就是用 Redis 的 ZSet 加 List 模拟了一套任务调度器利用 Redis 本身的持久化和分布式能力换来了一个无额外中间件依赖的轻量延迟队列。选型时要清楚它的边界可靠性不是百分百、精度是秒级、任务量有上限但只要场景匹配它给你的开发效率提升是实打实的。我实际用下来的体会是这类基于 Redis 的延迟队列并不适合承载核心资金链路里不能丢的消息它更适合做业务里的定时触发、状态流转、超时处理这类容忍小概率丢失但需要有兜底的场景。如果你正在选型不用一上来就上 RabbitMQ 或者自研时间轮先把 Redisson 方案跑起来等业务量确实大到它扛不住了再迁移到消息中间件也来得及。这个演进路径是最自然的也是我在多个项目里验证过的。