blockingQueue = redissonClient.getBlockingQueue(name);
delayQueue = redissonClient.getDelayedQueue(blockingQueue);

protected RedissonDelayedQueue(QueueTransferService queueTransferService, Codec codec, final CommandAsyncExecutor commandExecutor, String name) {
    super(codec, commandExecutor, name);
    // ... 初始化相關(guān)隊(duì)列名稱（channelName, queueName, timeoutSetName）...
    // 1. 創(chuàng)建QueueTransferTask子類實(shí)例（這里是匿名實(shí)現(xiàn)類,QueueTransferTask是抽象類）
    QueueTransferTask task = new (commandExecutor.getConnectionManager()) {
        @Override
        protected RFuture<Long> pushTaskAsync() {
            // 核心Lua腳本，負(fù)責(zé)轉(zhuǎn)移到期任務(wù)
            return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
                    "local expiredValues = redis.call('zrangebyscore', KEYS[2], 0, ARGV[1], 'limit', 0, ARGV[2]); " +
                    "if #expiredValues > 0 then " +
                      "for i, v in ipairs(expiredValues) do " +
                         "local randomId, value = struct.unpack('dLc0', v);" +
                         "redis.call('rpush', KEYS[1], value);" +  // 轉(zhuǎn)移到目標(biāo)隊(duì)列
                         "redis.call('lrem', KEYS[3], 1, v);" +   // 從備份隊(duì)列移除
                      "end; " +
                      "redis.call('zrem', KEYS[2], unpack(expiredValues));" + // 從延遲隊(duì)列移除
                    "end; " +
                    // 返回下一個(gè)任務(wù)的到期時(shí)間
                    "local v = redis.call('zrange', KEYS[2], 0, 0, 'WITHSCORES'); " +
                    "if v[1] ~= nil then " +
                       "return v[2]; " +
                    "end " +
                    "return nil;",
                    Arrays.asList(getRawName(), timeoutSetName, queueName),
                    System.currentTimeMillis(), 100);
        }
        
        @Override
        protected RTopic getTopic() {
            return RedissonTopic.createRaw(LongCodec.INSTANCE, commandExecutor, channelName);
        }
    };
    // 2. 啟動(dòng)任務(wù)
    queueTransferService.schedule(queueName, task);
}

public class QueueTransferService {
    private final Map<String, QueueTransferTask> tasks = new ConcurrentHashMap();
  	//...
    public void schedule(String name, QueueTransferTask task) {
        this.tasks.compute(name, (k, t) -> {
            if (t == null) {
                task.start();
                return task;
            } else {
                t.incUsage();
                return t;
            }
        });
    }
    //....
}

public abstract class QueueTransferTask {
    //.....
    //開啟這個(gè)延遲任務(wù)，初始化2個(gè)監(jiān)聽器
    public void start() {
        RTopic schedulerTopic = this.getTopic();
        this.statusListenerId = schedulerTopic.addListener(new BaseStatusListener() {
            //當(dāng)連接時(shí)候調(diào)用一次pushTask，后文會(huì)提及作用
            public void onSubscribe(String channel) {
                QueueTransferTask.this.pushTask();
            }
        });
        this.messageListenerId = schedulerTopic.addListener(Long.class, new MessageListener<Long>() {
            //當(dāng)從監(jiān)聽的channel中,監(jiān)聽到消息時(shí)候會(huì)調(diào)用此方法，這里后文會(huì)提及作用。
            public void onMessage(CharSequence channel, Long startTime) {
                QueueTransferTask.this.scheduleTask(startTime);
            }
        });
    }
    private void pushTask() {
        if (this.usage != 0) {
            RFuture<Long> startTimeFuture = this.pushTaskAsync();
            //這里res就是剛才返回的最早過期時(shí)間戳
            startTimeFuture.whenComplete((res, e) -> {
                //有異常就重試
                if (e != null) {
                    if (!this.serviceManager.isShuttingDown(e)) {
                        log.error(e.getMessage(), e);
                        this.scheduleTask(System.currentTimeMillis() + 5000L);
                    }
                } else {
                    if (res != null) {
                        //將時(shí)間戳傳遞給scheduleTask進(jìn)行調(diào)度任務(wù)
                        this.scheduleTask(res);
                    }
                }
            });
        }
    }

    //真正去規(guī)劃做任務(wù)的邏輯
    private void scheduleTask(Long startTime) {
        if (this.usage != 0) {
            if (startTime != null) {
                TimeoutTask oldTimeout = (TimeoutTask)this.lastTimeout.get();
                if (oldTimeout != null) {
                    oldTimeout.getTask().cancel();
                }

                long delay = startTime - System.currentTimeMillis();
                if (delay > 10L) {
                    //使用netty時(shí)間輪，安排任務(wù)
                    Timeout timeout = this.serviceManager.newTimeout(new TimerTask() {
                        public void run(Timeout timeout) throws Exception {
                            //執(zhí)行剛才匿名實(shí)現(xiàn)的task的pushTaskAsync方法
                            QueueTransferTask.this.pushTask();
                            TimeoutTask currentTimeout = (TimeoutTask)QueueTransferTask.this.lastTimeout.get();
                            if (currentTimeout != null && currentTimeout.getTask() == timeout) {
                                QueueTransferTask.this.lastTimeout.compareAndSet(currentTimeout, (Object)null);
                            }

                        }
                    }, delay, TimeUnit.MILLISECONDS);
                    this.lastTimeout.compareAndSet(oldTimeout, new TimeoutTask(startTime, timeout));
                } else {
                    this.pushTask();
                }

            }
        }
    }

}

@Override
public RFuture<Void> offerAsync(V e, long delay, TimeUnit timeUnit) {
  	if (delay < 0) {
   		throw new IllegalArgumentException("Delay can't be negative");
  	}
  	long delayInMs = timeUnit.toMillis(delay);
  	long timeout = System.currentTimeMillis() + delayInMs;
  	long randomId = ThreadLocalRandom.current().nextLong();
  	return commandExecutor.evalWriteNoRetryAsync(getRawName(), codec, RedisCommands.EVAL_VOID,
	//把 `timeout`（過期時(shí)間戳）、`randomId`（隨機(jī)ID）和 `encode(e)`（任務(wù)真實(shí)數(shù)據(jù)）打包成了一個(gè)二進(jìn)制的 `value`。
	//即使你添加了兩個(gè)一模一樣的任務(wù)（內(nèi)容一樣、時(shí)間一樣），因?yàn)橛?`randomId`，打包后的 `value` 也是不一樣的。這就允許隊(duì)列里存在重復(fù)的任務(wù)。
 	 "local value = struct.pack('dLc0', tonumber(ARGV[2]), string.len(ARGV[3]), ARGV[3]);" 
  	+ "redis.call('zadd', KEYS[2], ARGV[1], value);"
	//存入redisson_delay_queue:{name}，注意這里并不是存入結(jié)果List！！
	//用于保證 `RDelayedQueue` 這個(gè)對(duì)象本身有數(shù)據(jù)可查（比如你調(diào)用 `contains()` 或 `iterator()` 時(shí)，就是查這里）。
  	+ "redis.call('rpush', KEYS[3], value);"
	// 取出 ZSet 里排第一（最早要執(zhí)行）的任務(wù)
  	+ "local v = redis.call('zrange', KEYS[2], 0, 0); "
	//判斷：剛才新加的這個(gè)任務(wù)(value)，是不是就是那個(gè)排第一的(v[1])？
  	+ "if v[1] == value then "
	//如果是，說明新任務(wù)插隊(duì)成功，成為了最早的任務(wù)！發(fā)送 PUBLISH 通知客戶端："嘿，最早的時(shí)間變了，快調(diào)整鬧鐘！"
  	+ "redis.call('publish', KEYS[4], ARGV[1]); "
  	+ "end;",
  	Arrays.<Object>asList(getRawName(), timeoutSetName, queueName, channelName),
  	timeout, randomId, encode(e));
}