<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-redis</artifactId>
    <version>${version}</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>${version}</version>
</dependency>

    /**
     * 添加 ZSet 元素
     *
     * @param key
     * @param value
     * @param score
     */
    @Override
    public boolean add(String key, Object value, double score) {
        return redisTemplate.opsForZSet().add(key, value, score);
    }
    
    /**
     * 返回 分?jǐn)?shù)范圍內(nèi) 指定 count 數(shù)量的元素集合, 并且從 offset 下標(biāo)開始(從小到大,不帶分?jǐn)?shù)的集合)
     *
     * @param key
     * @param min
     * @param max
     * @param offset 從指定下標(biāo)開始
     * @param count  輸出指定元素?cái)?shù)量
     * @return
     */
    @Override
    public Set<Object> rangeByScore(String key, double min, double max, long offset, long count) {
        return redisTemplate.opsForZSet().rangeByScore(key, min, max, offset, count);
    }

    /**
     * Zset 刪除一個(gè)或多個(gè)元素
     *
     * @param key
     * @param values
     * @return
     */
    @Override
    public Long removeZset(String key, Object... values) {
        return redisTemplate.opsForZSet().remove(key, values);
    }

@Getter
@ToString
public class DelayMsg extends ApplicationEvent {
    private String msg;
    private String topic;

    public DelayMsg(Object source, String msg, String topic) {
        super(source);
        this.msg = msg;
        this.topic = topic;
    }
}

/**
 * 從zset中取出score小于當(dāng)前時(shí)間戳的數(shù)據(jù)
 *
 * @param key
 * @return
 */
public String getDelayOne(String key) {
    //先查后刪,一次拿3個(gè)做備選，這樣搶占到的概率就會(huì)高一些
    Set<Object> sets = redisService.rangeByScore(key, 0, System.currentTimeMillis(), 0, 3);
    if (CollectionUtils.isEmpty(sets)) {
        return null;
    }

    for (Object val : sets) {
        if (1L.equals(redisService.removeZset(key, val))) {
            // 刪除成功，表示搶占到
            return val.toString();
        }
    }
    return null;
}

/**
 * 每5s定時(shí)輪訓(xùn)消息
 */
@Scheduled(fixedRate = 5000)
public void schedule() {
   for (String specialTopic : topic) {
       String msg = redisDelayQueue.getDelayOne(specialTopic);
       logger.info("開始輪訓(xùn)獲取消息 {}", msg);
       if (StringUtil.isNotEmpty(msg)) {
       	   //使用Spring推送事件處理
           applicationContext.publishEvent(new DelayMsg(this, msg, specialTopic));
       }
   }
}

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@EventListener
public @interface Consumer {
    String topic();
}

@Aspect
@Component
public class ConsumerAspect {

    @Around("@annotation(consumer)")
    public Object around(ProceedingJoinPoint joinPoint, Consumer consumer) throws Throwable {
        Object[] args = joinPoint.getArgs();
        boolean check = false;
        for (Object obj : args) {
            if (obj instanceof DelayMsg) {
                check = consumer.topic().equals(((DelayMsg) obj).getTopic());
            }
        }
        if (!check) {
            // 不滿足條件，直接忽略
            return null;
        }
        // topic匹配成功，執(zhí)行
        return joinPoint.proceed();
    }
}

	//使用自定義的consumer注解監(jiān)聽topic延遲隊(duì)列
    @Consumer(topic = RedisKeyConstant.DELAY_QUEUE)
    public void consumer(DelayMsg delayMsg) {
        logger.info("預(yù)約單延時(shí)確認(rèn): " + delayMsg.getMsg() + " at:" + System.currentTimeMillis());
        //延遲業(yè)務(wù)具體實(shí)現(xiàn)
        //...
        //...
    }

@Component
public class DelayListWrapper {
    private Logger logger = LoggerFactory.getLogger(RedisDelayQueue.class);
    @Autowired
    RedisService redisService;

    private Set<String> topic = new CopyOnWriteArraySet<>();

    /**
     *
     * @param key 隊(duì)列名稱
     * @param val 消息內(nèi)容
     * @param delayTime 過期時(shí)間
     */
    public void publish(String key, Object val, long delayTime) {
        topic.add(key);
        String strVal = val instanceof String ? (String) val : JSONObject.toJSONString(val);
        redisService.add(key, strVal, System.currentTimeMillis() + delayTime);
        logger.info("key為：{},time:{}", key,System.currentTimeMillis() + delayTime);
    }
}

@Autowired
private DelayListWrapper delayListWrapper;
...
delayListWrapper.publish(RedisKeyConstant.DELAY_QUEUE, xxxId, xxx);