简介

上次发文是在5个月前。讲了一篇 redisson 分布式锁的实现原理，这次讲讲延迟队列的实现原理。如果你在项目中想用一个可靠的分布式延迟队列，却发现配置繁琐、性能不稳定。那么这篇文章正好能帮你快速上手并解决常见痛点。

API 的使用

Redisson 的延迟队列是通过先获取一个阻塞队列，接下来包装成延迟队列实现的：

blockingQueue = redissonClient.getBlockingQueue;delayQueue = redissonClient.getDelayedQueue;

这一步已经把业务与底层细节隔离，你只需要关注任务的提交和消费即可。

主要实现

内部维护了四个数据结构，外界完全感知不到：

• redisson_delay_queue_timeout:{name}——按到期时间排序所有延迟任务；列表首位即最早要执行的任务。老实说，
• redisson_delay_queue:{name}——主要流程不直接使用。但用于查询或遍历，
• {name}——目标队列，已到期可被使用者取出的任务。
• redisson_delay_queue_channel:{name}——通知客户端开启/调整延迟任务。

初始化过程

调用 redissonClient.getDelayedQueue 时会创建一个 RedissonDelayedQueue。老实说，其构造函数主要做两件事：

1. 初始化相关命名空间。
2. 创建并启动一个匿名实现类的 QueueTransferTask该任务负责将到期任务从 ZSet 转移到目标队列，并返回下一个最早到期时间戳。

protected RedissonDelayedQueue(QueueTransferService queueTransferService。Codec codec,final CommandAsyncExecutor commandExecutor,String name) {
super;// ... 初始化名称
QueueTransferTask task = new QueueTransferTask) {
@Override
protected RFuture pushTaskAsync {
// Lua 脚本：转移已过期任务 & 返回下一个最早时间戳
return commandExecutor.evalWriteAsync。LongCodec.INSTANCE,RedisCommands.EVAL_LONG,"... Lua 脚本内容 ...",Arrays.asList,timeoutSetName,queueName),System.currentTimeMillis);}
@Override
protected RTopic getTopic {
return RedissonTopic.createRaw(LongCodec.INSTANCE。commandExecutor,channelName);}
}，queueTransferService.schedule;}

Pain Point: 很多人担心 Lua 脚本复杂导致维护成本高。但脚本只有一段主要原因，而且是原子执行，保证了数据一致性。若你想自定义行为，只需替换脚本即可，无需改动 Java 代码。

Schedule 与 Listener 的工作方式

A. start: 为主题注册两个监听器：

• Status Listener: 当订阅建立时立即执行一次 ,防止重启后遗留未转移的数据。
• : 接收 channel 中发布的新时间戳，触发重新调度。

public void start {
RTopic schedulerTopic = this.getTopic;不过，this.statusListenerId = schedulerTopic.addListener {
public void onSubscribe { QueueTransferTask.this.pushTask;}
})，this.messageListenerId = schedulerTopic.addListener(Long.class,-> QueueTransferTask.this.scheduleTask);}

B. pushTask: 调用 Lua 脚本完成一次转移，并得到下一个最早到期时间戳；若有异常则重试，C. scheduleTask: delay，接下来决定是否使用 Netty 的 HashedWheelTimer 定时器。若 delay> 10ms 则排入轮询；否则立即执行，这种设计避免了频繁定时器开启导致资源浪费的问题。 Pain Point: 在高并发环境中手动管理定时器比较容易出错。但这里利用 HashedWheelTimer 自动化处理，大幅降低开发者维护成本。

发送延迟任务

生产者提交任务时会调用以下方法：

@Override
public RFuture offerAsync {
if throw new IllegalArgumentException;long delayInMs = timeUnit.toMillis;long timeout = System.currentTimeMillis + delayInMs;long randomId = ThreadLocalRandom.current.nextLong;return commandExecutor.evalWriteNoRetryAsync(
getRawName,codec,RedisCommands.EVAL_VOID。// 构造二进制 value 并存入 ZSet 与 List
"local value = struct.pack,string.len,ARGV);"
+ "redis.call;"
+ "redis.call;"
// 检查是否成为新的最早任务
+ "local v = redis.call;"
+ "if v == value n"
+ " redis.call;"
+ "end，",Arrays.asList,timeoutSetName,queueName。channelName),timeout,randomId,encode);}

This script guarantees atomicity: task is added to both ZSet and List simultaneously;if it becomes earliest task it publishes a notification on channel.

• Pain Point: Merging Duplicate Tasks? Because each entry contains a unique randomId packed into binary value,identical payloads can coexist in queue without collision.

Channel 通知的关键性

If a client is currently blocked waiting for an existing longest-delay task。a newly submitted short-delay task arrives. The channel message alerts client that its current timer is no longer optimal;it cancels old timeout and schedules a new one aligned with earlier deadline.

• Pain Point: No “Clock Skew” Issue?其实， Because all timing decisions are based on server timestamps embedded in Lua scripts and propagated via channel messages。clock drift 娱乐ween clients is effectively neutralized.

为什么要这么复杂？话说回来，

• The four Redis structures decouple persistence from consumer logic—no external scheduler needed.
• A single Lua script atomically moves expired tasks and returns next deadline.
• The scheduled Netty timer ensures minimal CPU usage even under high throughput.
• Your code stays simple: just call offer/take;all heavy lifting happens inside Redisson.

If you’re still worried about “is this thread‑safe?” or “will my tasks ever miss ir deadline?”,rest assured—Redisson’s design guarantees atomicity at every step and continuous monitoring of deadlines via Netty timers.

再看接下来建议。如何监控和调优，

• *监控*的观点是。查看 redisson_delay_queue_timeout:{name} ZSet 的大小和平均分数差异，可快速定位“堆积”问题。
• 说到*调优*，根据业务特点设置合理的最低等待阈值。或在极端高并发场景改为更精细的轮询策略。
• *故障恢复*这方面，重新启动后自动触发一次 pushTask 确保无遗漏；若仍出现丢失，可检查网络连通性和持久化日志。