Redisson 源码分析及实际应用场景之实现延迟队列

//获取锁（unique_value可以是UUID等）SET resource_name unique_value NX PX  30000
//释放锁（lua脚本中，一定要比较value，防止误解锁）if redis.call("get",KEYS[1]) == ARGV[1] then    return redis.call("del",KEYS[1])else    return 0end

public class RedisTool {
    private static final String LOCK_SUCCESS = "OK";    private static final String SET_IF_NOT_EXIST = "NX";    private static final String SET_WITH_EXPIRE_TIME = "PX";    private static final Long RELEASE_SUCCESS = 1L;
    /**     * 获取分布式锁(加锁代码)     */    public static boolean getDistributedLock(Jedis jedis, String lockKey, String requestId, int expireTime) {
        String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
        if (LOCK_SUCCESS.equals(result)) {            return true;        }        return false;    }
    /**     * 释放分布式锁(解锁代码)     */    public static boolean releaseDistributedLock(Jedis jedis, String lockKey, String requestId) {
        String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else               return 0 end";        Object result = jedis.eval(script, Collections.singletonList(lockKey), C                                                   ollections.singletonList(requestId));
        if (RELEASE_SUCCESS.equals(result)) {            return true;        }        return false;
    }}

1、获取当前时间，以毫秒为单位。2、依次尝试从5个实例，使用相同的key和随机值（Redisson中给出的是UUID + ThreadId）获取锁。当向Redis请求获取锁时，客户端应该设置一个网络连接和响应超时时间，这个超时时间应该小于锁的失效时间。假如锁自动失效时间为10秒，则超时时间应该在5-50毫秒之间。这样可以避免服务器端Redis已经挂掉的情况下，客户端还在一直等待响应结果。如果服务器端没有在规定时间内响应，客户端应该尽快尝试去另外一个Redis实例请求获取锁。3、客户端使用当前时间减去开始获取锁时间（步骤1记录的时间）就得到获取锁使用的时间。当且仅当从大多数（N/2+1，这里是3个节点）的Redis节点都取到锁，并且使用的时间小于锁失效时间时，锁才算获取成功。4、如果取到了锁，key的真正有效时间等于有效时间减去获取锁所使用的时间（步骤3计算的结果）。5、如果因为某些原因，获取锁失败，客户端应该在所有的Redis实例上进行解锁即便某些Redis实例根本就没有加锁成功，防止某些节点获取到锁但是客户端没有得到响应而导致接下来的一段时间不能被重新获取锁。

// 1.构造redisson实现分布式锁必要的Config        Config config = new Config();        config.useSingleServer().setAddress("redis://127.0.0.1:5379").setPassword("123456").setDatabase(0);        // 2.构造RedissonClient        RedissonClient redissonClient = Redisson.create(config);        // 3.获取锁对象实例（无法保证是按线程的顺序获取到）        RLock rLock = redissonClient.getLock(lockKey);        try {            /**             * 4.尝试获取锁             * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败             * leaseTime   锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）             */            boolean res = rLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);            if (res) {                //成功获得锁，在这里处理业务            }        } catch (Exception e) {            throw new RuntimeException("aquire lock fail");        }finally{            //无论如何, 最后都要解锁            rLock.unlock();        }

@Override    public RLock getLock(String name) {        /**         *  构造并返回一个 RedissonLock 对象          * commandExecutor: 与 Redis 节点通信并发送指令的真正实现。需要说明一下，CommandExecutor 实现是通过 eval 命令来执行 Lua 脚本         * name: 锁的全局名称         * id: Redisson 客户端唯一标识，实际上就是一个 UUID.randomUUID()         */        return new RedissonLock(commandExecutor, name, id);    }

@Override    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {        //取得最大等待时间        long time = unit.toMillis(waitTime);        //记录下当前时间        long current = System.currentTimeMillis();        //取得当前线程id（判断是否可重入锁的关键）        long threadId = Thread.currentThread().getId();        //1.尝试申请锁，返回还剩余的锁过期时间        Long ttl = tryAcquire(leaseTime, unit, threadId);        //2.如果为空，表示申请锁成功        if (ttl == null) {            return true;        }        //3.申请锁的耗时如果大于等于最大等待时间，则申请锁失败        time -= System.currentTimeMillis() - current;        if (time <= 0) {            /**             * 通过 promise.trySuccess 设置异步执行的结果为null             * Promise从Uncompleted-->Completed ,通知 Future 异步执行已完成             */            acquireFailed(threadId);            return false;        }        current = System.currentTimeMillis();
        /**         * 4.订阅锁释放事件，并通过await方法阻塞等待锁释放，有效的解决了无效的锁申请浪费资源的问题：         * 基于信息量，当锁被其它资源占用时，当前线程通过 Redis 的 channel 订阅锁的释放事件，一旦锁释放会发消息通知待等待的线程进行竞争         * 当 this.await返回false，说明等待时间已经超出获取锁最大等待时间，取消订阅并返回获取锁失败         * 当 this.await返回true，进入循环尝试获取锁         */        RFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);        //await 方法内部是用CountDownLatch来实现阻塞，获取subscribe异步执行的结果（应用了Netty 的 Future）        if (!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {            if (!subscribeFuture.cancel(false)) {                subscribeFuture.onComplete((res, e) -> {                    if (e == null) {                        unsubscribe(subscribeFuture, threadId);                    }                });            }            acquireFailed(threadId);            return false;        }
        try {            //计算获取锁的总耗时，如果大于等于最大等待时间，则获取锁失败            time -= System.currentTimeMillis() - current;            if (time <= 0) {                acquireFailed(threadId);                return false;            }
            /**             * 5.收到锁释放的信号后，在最大等待时间之内，循环一次接着一次的尝试获取锁             * 获取锁成功，则立马返回true，             * 若在最大等待时间之内还没获取到锁，则认为获取锁失败，返回false结束循环             */            while (true) {                long currentTime = System.currentTimeMillis();                // 再次尝试申请锁                ttl = tryAcquire(leaseTime, unit, threadId);                // 成功获取锁则直接返回true结束循环                if (ttl == null) {                    return true;                }
                //超过最大等待时间则返回false结束循环，获取锁失败                time -= System.currentTimeMillis() - currentTime;                if (time <= 0) {                    acquireFailed(threadId);                    return false;                }
                /**                 * 6.阻塞等待锁（通过信号量(共享锁)阻塞,等待解锁消息）：                 */                currentTime = System.currentTimeMillis();                if (ttl >= 0 && ttl < time) {                    //如果剩余时间(ttl)小于wait time ,就在 ttl 时间内，从Entry的信号量获取一个许可(除非被中断或者一直没有可用的许可)。                    getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);                } else {                    //则就在wait time 时间范围内等待可以通过信号量                    getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);                }
                //7.更新剩余的等待时间(最大等待时间-已经消耗的阻塞时间)                time -= System.currentTimeMillis() - currentTime;                if (time <= 0) {                    acquireFailed(threadId);                    return false;                }            }        } finally {            //7.无论是否获得锁,都要取消订阅解锁消息            unsubscribe(subscribeFuture, threadId);        }    }

<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {        internalLockLeaseTime = unit.toMillis(leaseTime);
        /**         * 通过 EVAL 命令执行 Lua 脚本获取锁，保证了原子性         */        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,                  // 1.如果缓存中的key不存在，则执行 hset 命令(hset key UUID+threadId 1),然后通过 pexpire 命令设置锁的过期时间(即锁的租约时间)                  // 返回空值 nil ，表示获取锁成功                  "if (redis.call('exists', KEYS[1]) == 0) then " +                      "redis.call('hset', KEYS[1], ARGV[2], 1); " +                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +                      "return nil; " +                  "end; " +                   // 如果key已经存在，并且value也匹配，表示是当前线程持有的锁，则执行 hincrby 命令，重入次数加1，并且设置失效时间                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +                      "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +                      "redis.call('pexpire', KEYS[1], ARGV[1]); " +                      "return nil; " +                  "end; " +                   //如果key已经存在，但是value不匹配，说明锁已经被其他线程持有，通过 pttl 命令获取锁的剩余存活时间并返回，至此获取锁失败                  "return redis.call('pttl', KEYS[1]);",                   //这三个参数分别对应KEYS[1]，ARGV[1]和ARGV[2]                   Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));    }

private <T> RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {    RFuture<Long> ttlRemainingFuture;    if (leaseTime != -1) {        ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);    } else {        ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime,                TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);    }
    CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {        // lock acquired        if (ttlRemaining == null) {            if (leaseTime != -1) {                internalLockLeaseTime = unit.toMillis(leaseTime);            } else {                // 锁过期时间续期                scheduleExpirationRenewal(threadId);            }        }        return ttlRemaining;    });    return new CompletableFutureWrapper<>(f);}

protected void scheduleExpirationRenewal(long threadId) {    ExpirationEntry entry = new ExpirationEntry();    ExpirationEntry oldEntry = EXPIRATION_RENEWAL_MAP.putIfAbsent(getEntryName(), entry);    if (oldEntry != null) {        oldEntry.addThreadId(threadId);    } else {        entry.addThreadId(threadId);        try {            renewExpiration();        } finally {            if (Thread.currentThread().isInterrupted()) {                cancelExpirationRenewal(threadId);            }        }    }}

private void renewExpiration() {    ExpirationEntry ee = EXPIRATION_RENEWAL_MAP.get(getEntryName());    if (ee == null) {        return;    }
    // 创建延迟任务    Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {        @Override        public void run(Timeout timeout) throws Exception {            ExpirationEntry ent = EXPIRATION_RENEWAL_MAP.get(getEntryName());            if (ent == null) {                return;            }            Long threadId = ent.getFirstThreadId();            if (threadId == null) {                return;            }
            // 真正的续期，调用 LUA 脚本续期            RFuture<Boolean> future = renewExpirationAsync(threadId);            future.whenComplete((res, e) -> {                if (e != null) {                    log.error("Can't update lock " + getRawName() + " expiration", e);                    EXPIRATION_RENEWAL_MAP.remove(getEntryName());                    return;                }
                // 如果续期成功                if (res) {                    // reschedule itself                    renewExpiration();                } else {                    cancelExpirationRenewal(null);                }            });        }    }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);
    ee.setTimeout(task);}

protected RFuture<Boolean> renewExpirationAsync(long threadId) {        return eval（getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +                        "return 1; " +                        "end; " +                        "return 0;",                Collections.singletonList(getRawName()),                internalLockLeaseTime, getLockName(threadId));    }

private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {    long threadId = Thread.currentThread().getId();    Long ttl = tryAcquire(-1, leaseTime, unit, threadId);    // lock acquired    if (ttl == null) {        return;    }
    CompletableFuture<RedissonLockEntry> future = subscribe(threadId);    RedissonLockEntry entry;    if (interruptibly) {        entry = commandExecutor.getInterrupted(future);    } else {        entry = commandExecutor.get(future);    }
    try {        while (true) {            ttl = tryAcquire(-1, leaseTime, unit, threadId);            // lock acquired            if (ttl == null) {                break;            }
            // waiting for message            if (ttl >= 0) {                try {                    // 阻塞锁的超时时间，等锁过期后再尝试加锁                    entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);                } catch (InterruptedException e) {                    if (interruptibly) {                        throw e;                    }                    entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);                }            } else {                if (interruptibly) {                    entry.getLatch().acquire();                } else {                    entry.getLatch().acquireUninterruptibly();                }            }        }    } finally {        unsubscribe(entry, threadId);    }//        get(lockAsync(leaseTime, unit));}

CompletableFuture future = subscribe(threadId);

@Override    public void unlock() {        try {            get(unlockAsync(Thread.currentThread().getId()));        } catch (RedisException e) {            if (e.getCause() instanceof IllegalMonitorStateException) {                throw (IllegalMonitorStateException) e.getCause();            } else {                throw e;            }        }    }

@Override    public <V> V get(RFuture<V> future) {        if (!future.isDone()) {   //任务还没完成            // 设置一个单线程的同步控制器            CountDownLatch l = new CountDownLatch(1);            future.onComplete((res, e) -> {                //操作完成时，唤醒在await()方法中等待的线程                l.countDown();            });
            boolean interrupted = false;            while (!future.isDone()) {                try {                    //阻塞等待                    l.await();                } catch (InterruptedException e) {                    interrupted = true;                    break;                }            }
            if (interrupted) {                Thread.currentThread().interrupt();            }        }        if (future.isSuccess()) {            return future.getNow();        }
        throw convertException(future);    }

protected RFuture<Boolean> unlockInnerAsync(long threadId) {        /**         * 通过 EVAL 命令执行 Lua 脚本获取锁，保证了原子性         */        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,                //如果分布式锁存在，但是value不匹配，表示锁已经被其他线程占用，无权释放锁，那么直接返回空值（解铃还须系铃人）                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +                    "return nil;" +                "end; " +                 //如果value匹配，则就是当前线程占有分布式锁，那么将重入次数减1                "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +                 //重入次数减1后的值如果大于0，表示分布式锁有重入过，那么只能更新失效时间，还不能删除                "if (counter > 0) then " +                    "redis.call('pexpire', KEYS[1], ARGV[2]); " +                    "return 0; " +                "else " +                 //重入次数减1后的值如果为0，这时就可以删除这个KEY，并发布解锁消息，返回1                    "redis.call('del', KEYS[1]); " +                    "redis.call('publish', KEYS[2], ARGV[1]); " +                    "return 1; "+                "end; " +                "return nil;",                //这5个参数分别对应KEYS[1]，KEYS[2]，ARGV[1]，ARGV[2]和ARGV[3]                Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));
    }

@Overridepublic RFuture<Void> unlockAsync(long threadId) {    // 解锁    RFuture<Boolean> future = unlockInnerAsync(threadId);
    // 撤销续期    CompletionStage<Void> f = future.handle((opStatus, e) -> {        cancelExpirationRenewal(threadId);
        if (e != null) {            throw new CompletionException(e);        }        if (opStatus == null) {            IllegalMonitorStateException cause = new IllegalMonitorStateException("attempt to unlock lock, not locked by current thread by node id: "                    + id + " thread-id: " + threadId);            throw new CompletionException(cause);        }
        return null;    });
    return new CompletableFutureWrapper<>(f);}

public class LockPubSub extends PublishSubscribe<RedissonLockEntry> {
    public static final Long UNLOCK_MESSAGE = 0L;    public static final Long READ_UNLOCK_MESSAGE = 1L;
    public LockPubSub(PublishSubscribeService service) {        super(service);    }    @Override    protected RedissonLockEntry createEntry(RPromise<RedissonLockEntry> newPromise) {        return new RedissonLockEntry(newPromise);    }
    @Override    protected void onMessage(RedissonLockEntry value, Long message) {
        /**         * 判断是否是解锁消息         */        if (message.equals(UNLOCK_MESSAGE)) {            Runnable runnableToExecute = value.getListeners().poll();            if (runnableToExecute != null) {                runnableToExecute.run();            }
            /**             * 释放一个信号量，唤醒等待的entry.getLatch().tryAcquire去再次尝试申请锁             */            value.getLatch().release();        } else if (message.equals(READ_UNLOCK_MESSAGE)) {            while (true) {                /**                 * 如果还有其他Listeners回调，则也唤醒执行                 */                Runnable runnableToExecute = value.getListeners().poll();                if (runnableToExecute == null) {                    break;                }                runnableToExecute.run();            }
            value.getLatch().release(value.getLatch().getQueueLength());        }    }
}

Config config1 = new Config();        config1.useSingleServer().setAddress("redis://172.0.0.1:5378").setPassword("a123456").setDatabase(0);        RedissonClient redissonClient1 = Redisson.create(config1);
        Config config2 = new Config();        config2.useSingleServer().setAddress("redis://172.0.0.1:5379").setPassword("a123456").setDatabase(0);        RedissonClient redissonClient2 = Redisson.create(config2);
        Config config3 = new Config();        config3.useSingleServer().setAddress("redis://172.0.0.1:5380").setPassword("a123456").setDatabase(0);        RedissonClient redissonClient3 = Redisson.create(config3);
        /**         * 获取多个 RLock 对象         */        RLock lock1 = redissonClient1.getLock(lockKey);        RLock lock2 = redissonClient2.getLock(lockKey);        RLock lock3 = redissonClient3.getLock(lockKey);
        /**         * 根据多个 RLock 对象构建 RedissonRedLock （最核心的差别就在这里）         */        RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);
        try {            /**             * 4.尝试获取锁             * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败             * leaseTime   锁的持有时间,超过这个时间锁会自动失效（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）             */            boolean res = redLock.tryLock((long)waitTimeout, (long)leaseTime, TimeUnit.SECONDS);            if (res) {                //成功获得锁，在这里处理业务            }        } catch (Exception e) {            throw new RuntimeException("aquire lock fail");        }finally{            //无论如何, 最后都要解锁            redLock.unlock();        }

public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {        long newLeaseTime = -1;        if (leaseTime != -1) {            newLeaseTime = unit.toMillis(waitTime)*2;        }        long time = System.currentTimeMillis();        long remainTime = -1;        if (waitTime != -1) {            remainTime = unit.toMillis(waitTime);        }        long lockWaitTime = calcLockWaitTime(remainTime);        /**         * 1. 允许加锁失败节点个数限制（N-(N/2+1)）         */        int failedLocksLimit = failedLocksLimit();        /**         * 2. 遍历所有节点通过EVAL命令执行lua加锁         */        List<RLock> acquiredLocks = new ArrayList<>(locks.size());        for (ListIterator<RLock> iterator = locks.listIterator(); iterator.hasNext();) {            RLock lock = iterator.next();            boolean lockAcquired;            /**             *  3.对节点尝试加锁             */            try {                if (waitTime == -1 && leaseTime == -1) {                    lockAcquired = lock.tryLock();                } else {                    long awaitTime = Math.min(lockWaitTime, remainTime);                    lockAcquired = lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);                }            } catch (RedisResponseTimeoutException e) {                // 如果抛出这类异常，为了防止加锁成功，但是响应失败，需要解锁所有节点                unlockInner(Arrays.asList(lock));                lockAcquired = false;            } catch (Exception e) {                // 抛出异常表示获取锁失败                lockAcquired = false;            }            if (lockAcquired) {                /**                 *4. 如果获取到锁则添加到已获取锁集合中                 */                acquiredLocks.add(lock);            } else {                /**                 * 5. 计算已经申请锁失败的节点是否已经到达 允许加锁失败节点个数限制 （N-(N/2+1)）                 * 如果已经到达， 就认定最终申请锁失败，则没有必要继续从后面的节点申请了                 * 因为 Redlock 算法要求至少N/2+1 个节点都加锁成功，才算最终的锁申请成功                 */                if (locks.size() - acquiredLocks.size() == failedLocksLimit()) {                    break;                }
                if (failedLocksLimit == 0) {                    unlockInner(acquiredLocks);                    if (waitTime == -1 && leaseTime == -1) {                        return false;                    }                    failedLocksLimit = failedLocksLimit();                    acquiredLocks.clear();                    // reset iterator                    while (iterator.hasPrevious()) {                        iterator.previous();                    }                } else {                    failedLocksLimit--;                }            }
            /**             * 6.计算 目前从各个节点获取锁已经消耗的总时间，如果已经等于最大等待时间，则认定最终申请锁失败，返回false             */            if (remainTime != -1) {                remainTime -= System.currentTimeMillis() - time;                time = System.currentTimeMillis();                if (remainTime <= 0) {                    unlockInner(acquiredLocks);                    return false;                }            }        }
        if (leaseTime != -1) {            List<RFuture<Boolean>> futures = new ArrayList<>(acquiredLocks.size());            for (RLock rLock : acquiredLocks) {                RFuture<Boolean> future = ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);                futures.add(future);            }            for (RFuture<Boolean> rFuture : futures) {                rFuture.syncUninterruptibly();            }        }
        /**         * 7.如果逻辑正常执行完则认为最终申请锁成功，返回true         */        return true;    }

@ToStringpublic class MyDelayed<T> implements Delayed {    /**     * 延迟时间     */    Long delayTime;
    /**     * 过期时间     */    Long expire;    /**     * 数据     */    T t;
    public MyDelayed(long delayTime, T t) {        this.delayTime = delayTime;        // 过期时间 = 当前时间 + 延迟时间        this.expire = System.currentTimeMillis() + delayTime;        this.t = t;    }
    /**     * 剩余时间 = 到期时间 - 当前时间     */    @Override    public long getDelay(TimeUnit unit) {        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);    }
    /**     * 优先级规则：两个任务比较，时间短的优先执行     */    @Override    public int compareTo(Delayed o) {        long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);        return (int) f;    }

@Datapublic class OrderInfo implements Serializable {    private static final long serialVersionUID = -2837036864073566484L;    /**     * 订单id     */    private Long id;    /**     * 订单金额     */    private Double salary;
    /**     * 订单创建时间   对于java8LocalDateTime 以下注解序列化反序列化用到     */    @JsonDeserialize(using = LocalDateTimeDeserializer.class)    @JsonSerialize(using = LocalDateTimeSerializer.class)    private LocalDateTime createTime;
}

public static void main(String[] args) throws InterruptedException {        OrderInfo orderInfo = new OrderInfo();        orderInfo.setCreateTime(LocalDateTimeUtil.parse("2022-07-01 15:00:00", "yyyy-MM-dd HH:mm:ss"));        MyDelayed<OrderInfo> myDelayed = new MyDelayed<>(10000L,orderInfo);        DelayQueue<MyDelayed<OrderInfo>> queue = new DelayQueue<>();        queue.add(myDelayed);        System.out.println(queue.take().getT().getCreateTime());        System.out.println("当前时间：" + LocalDateTime.now());    }

<dependency>      <groupId>org.redisson</groupId>      <artifactId>redisson-spring-boot-starter</artifactId>      <version>3.16.7</version> </dependency>

# 单节点配置singleServerConfig:  # 连接空闲超时，单位：毫秒  idleConnectionTimeout: 10000  # 连接超时，单位：毫秒  connectTimeout: 10000  # 命令等待超时，单位：毫秒  timeout: 3000  # 命令失败重试次数,如果尝试达到 retryAttempts（命令失败重试次数） 仍然不能将命令发送至某个指定的节点时，将抛出错误。  # 如果尝试在此限制之内发送成功，则开始启用 timeout（命令等待超时） 计时。  retryAttempts: 3  # 命令重试发送时间间隔，单位：毫秒  retryInterval: 1500  # 密码  password:  # 单个连接最大订阅数量  subscriptionsPerConnection: 5  # 客户端名称  clientName: null  # 节点地址  address: redis://127.0.0.1:6379  # 发布和订阅连接的最小空闲连接数  subscriptionConnectionMinimumIdleSize: 1  # 发布和订阅连接池大小  subscriptionConnectionPoolSize: 50  # 最小空闲连接数  connectionMinimumIdleSize: 32  # 连接池大小  connectionPoolSize: 64  # 数据库编号  database: 0  # DNS监测时间间隔，单位：毫秒  dnsMonitoringInterval: 5000# 线程池数量,默认值: 当前处理核数量 * 2#threads: 0# Netty线程池数量,默认值: 当前处理核数量 * 2#nettyThreads: 0# 编码codec: !<org.redisson.codec.JsonJacksonCodec> {}# 传输模式transportMode : "NIO"

@Configurationpublic class RedissonConfig {    @Bean    public RedissonClient redissonClient() throws IOException {        Config config = Config.fromYAML(RedissonConfig.class.getClassLoader().getResource("redisson.yml"));;        return Redisson.create(config);    }}

         // redisson  延迟队列        // Redisson的延时队列是对另一个队列的再包装，使用时要先将延时消息添加到延时队列中，        // 当延时队列中的消息达到设定的延时时间后，该延时消息才会进行进入到被包装队列中，因此，我们只需要对被包装队列进行监听即可。        RBlockingQueue<OrderInfo> blockingFairQueue = redissonClient.getBlockingQueue("my-test");
        RDelayedQueue<OrderInfo> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
        OrderInfo orderInfo = new OrderInfo();        // 订单生成时间        orderInfo.setCreateTime(LocalDateTime.now());        // 10秒钟以后将消息发送到指定队列        delayedQueue.offer(orderInfo, 10, TimeUnit.SECONDS);        RBlockingQueue<OrderInfo> outQueue = redissonClient.getBlockingQueue("my-test");
        OrderInfo orderInfo2 = outQueue.take();        System.out.println("订单生成时间" + orderInfo2.getCreateTime());        System.out.println("订单关闭时间" + LocalDateTime.now());
        // 在该对象不再需要的情况下，应该主动销毁。仅在相关的Redisson对象也需要关闭的时候可以不用主动销毁        delayedQueue.destroy();

// 这里订阅了一个固定的队列 redisson_delay_queue_channel:{my-test}，为了开启进程里面的延时任务"SUBSCRIBE" "redisson_delay_queue_channel:{my-test}"
// Redis Zrangebyscore 返回有序集合中指定分数区间的成员列表。有序集成员按分数值递增(从小到大)次序排列。// redisson_delay_queue_channel:{my-test} 是一个zset，当有延时数据存入Redisson队列时，就会在此队列中插入 数据，排序分数为延时的时间戳(毫秒 以下同理)。"zrangebyscore" "redisson_delay_queue_timeout:{my-test}" "0" "1656404479385" "limit" "0" "100"
// 取出第一个数，也就是判断上面执行的操作是否有下一页。（因为刚刚开始总是0的）除非是之前的操作（zrangebyscore）没有取完"zrange" "redisson_delay_queue_timeout:{my-test}" "0" "0" "WITHSCORES"
// 往zset里面设置 数据过期的时间戳（当前执行的时间戳+延时的时间毫秒值）内容就是订单数据"zadd" "redisson_delay_queue_timeout:{my-test}" "1656404489400" "b\x99M9\x9b\x0c\xd3\xc3\\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"
// 同步一份数据到list队列"rpush" "redisson_delay_queue:{my-test}" "b\x99M9\x9b\x0c\xd3\xc3\\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"
// 取出排序好的第一个数据，也就是最临近要触发的数据，然后发送通知"zrange" "redisson_delay_queue_timeout:{my-test}" "0" "0"
// 发送通知  之前第一步 SUBSCRIBE 订阅了 客户端收到通知后，就在自己进程里面开启延时任务（HashedWheelTimer），到时间后就可以从redis取数据发送"publish" "redisson_delay_queue_channel:{my-test}" "1656404489400"
// 这里就是取数据环节了"BLPOP" "my-test" "0"
// 在范围 0-过期时间  取出100条数据"zrangebyscore" "redisson_delay_queue_timeout:{my-test}" "0" "1656404489444" "limit" "0" "100"
// 将上面取到的数据push到阻塞队列 很显然能看到 com.example.mytest.domain.OrderInfo 是我们的订单数据"rpush" "my-test" "{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"
// 删除数据"lrem" "redisson_delay_queue:{my-test}" "1" "b\x99M9\x9b\x0c\xd3\xc3\\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}""zrem" "redisson_delay_queue_timeout:{my-test}" "b\x99M9\x9b\x0c\xd3\xc3\\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"
// 取zset第一个数据，有的话继续上面逻辑取数据"zrange" "redisson_delay_queue_timeout:{my-test}" "0" "0" "WITHSCORES"
// 退订"UNSUBSCRIBE" "redisson_delay_queue_channel:{my-test}"