public class Order implements Delayed {
    /**
     * 延遲時(shí)間
     */
    @JsonFormat(locale = "zh", timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
    private long time;
    String name;
    public Order(String name, long time, TimeUnit unit) {
        this.name = name;
        this.time = System.currentTimeMillis() + (time > 0 ? unit.toMillis(time) : 0);
    }
    @Override
    public long getDelay(TimeUnit unit) {
        return time - System.currentTimeMillis();
    }
    @Override
    public int compareTo(Delayed o) {
        Order Order = (Order) o;
        long diff = this.time - Order.time;
        if (diff <= 0) {
            return -1;
        } else {
            return 1;
        }
    }
}

public class DelayQueueDemo {
    public static void main(String[] args) throws InterruptedException {
        Order Order1 = new Order("Order1", 5, TimeUnit.SECONDS);
        Order Order2 = new Order("Order2", 10, TimeUnit.SECONDS);
        Order Order3 = new Order("Order3", 15, TimeUnit.SECONDS);
        DelayQueue<Order> delayQueue = new DelayQueue<>();
        delayQueue.put(Order1);
        delayQueue.put(Order2);
        delayQueue.put(Order3);
        System.out.println("訂單延遲隊(duì)列開始時(shí)間:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        while (delayQueue.size() != 0) {
            /**
             * 取隊(duì)列頭部元素是否過期
             */
            Order task = delayQueue.poll();
            if (task != null) {
                System.out.format("訂單:{%s}被取消, 取消時(shí)間:{%s}\n", task.name, LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            }
            Thread.sleep(1000);
        }
    }
}

<dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

@EnableScheduling
@SpringBootApplication
public class DelayqueueApplication {
	public static void main(String[] args) {
		SpringApplication.run(DelayqueueApplication.class, args);
	}
}

@Component
public class QuartzDemo {
    //每隔五秒
    @Scheduled(cron = "0/5 * * * * ? ")
    public void process(){
        System.out.println("我是定時(shí)任務(wù)！");
    }
}

 zadd delayqueue 3 order3

    /**
     * 消費(fèi)消息
     */
    public void pollOrderQueue() {
        while (true) {
            Set<Tuple> set = jedis.zrangeWithScores(DELAY_QUEUE, 0, 0);
            String value = ((Tuple) set.toArray()[0]).getElement();
            int score = (int) ((Tuple) set.toArray()[0]).getScore();
            Calendar cal = Calendar.getInstance();
            int nowSecond = (int) (cal.getTimeInMillis() / 1000);
            if (nowSecond >= score) {
                jedis.zrem(DELAY_QUEUE, value);
                System.out.println(sdf.format(new Date()) + " removed key:" + value);
            }
            if (jedis.zcard(DELAY_QUEUE) <= 0) {
                System.out.println(sdf.format(new Date()) + " zset empty ");
                return;
            }
            Thread.sleep(1000);
        }
    }

    @Override
    public <T> RBlockingQueue<T> getBlockingQueue(String key) {
        return redissonClient.getBlockingQueue(key);
    }
    @Override
    public <T> RDelayedQueue<T> getDelayedQueue(RBlockingQueue<T> rBlockingQueue) {
        return redissonClient.getDelayedQueue(rBlockingQueue);
    }

// 將庫存入庫任務(wù)存放到延時(shí)隊(duì)列中
    @Override
    public void awardStockConsumeSendQueue(StrategyAwardStockKeyVO strategyAwardStockKeyVO) {
        String cacheKey = Constants.RedisKey.STRATEGY_AWARD_COUNT_QUERY_KEY;
        // 獲取對(duì)應(yīng)key的堵塞隊(duì)列
        RBlockingQueue<StrategyAwardStockKeyVO> blockingQueue = redisService.getBlockingQueue(cacheKey);
        // 獲取對(duì)應(yīng)堵塞隊(duì)列的延時(shí)隊(duì)列
        RDelayedQueue<StrategyAwardStockKeyVO> delayedQueue = redisService.getDelayedQueue(blockingQueue);
        // 將任務(wù)對(duì)象(你自己定義的對(duì)象)放入延時(shí)隊(duì)列中，三秒后會(huì)放入堵塞隊(duì)列
        delayedQueue.offer(strategyAwardStockKeyVO, 3, TimeUnit.SECONDS);
    }
    // 從堵塞隊(duì)列中獲取任務(wù)
    @Override
    public StrategyAwardStockKeyVO takeStockQueueValue() throws InterruptedException {
        String cacheKey = Constants.RedisKey.STRATEGY_AWARD_COUNT_QUERY_KEY;
        RBlockingQueue<StrategyAwardStockKeyVO> destinationQueue = redisService.getBlockingQueue(cacheKey);
        return destinationQueue.poll();
    }

// 庫存數(shù)據(jù)同步任務(wù)
    @Scheduled(cron = "0/5 * * * * ?")
    public void awardStockUpdate() {
        try {
            log.info("定時(shí)任務(wù)，更新獎(jiǎng)品消耗庫存【延遲隊(duì)列獲取】");
            StrategyAwardStockKeyVO strategyAwardStockKeyVO = raffleStock.takeStockQueueValue();
            if (null == strategyAwardStockKeyVO) return;
            log.info("定時(shí)任務(wù)，更新獎(jiǎng)品消耗庫存 strategyId:{} awardId:{}", strategyAwardStockKeyVO.getStrategyId(), strategyAwardStockKeyVO.getAwardId());
            raffleStock.updateStrategyAwardStock(strategyAwardStockKeyVO.getStrategyId(), strategyAwardStockKeyVO.getAwardId());
        } catch (Exception e) {
            log.error("定時(shí)任務(wù)，更新獎(jiǎng)品消耗庫存失敗", e);
        }
    }

notify-keyspace-events Ex

@Configuration
public class RedisListenerConfig {
    @Bean
    RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(connectionFactory);
        return container;
    }
}

@Component
public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
        super(listenerContainer);
    }
    @Override
    public void onMessage(Message message, byte[] pattern) {
        String expiredKey = message.toString();
        System.out.println("監(jiān)聽到key：" + expiredKey + "已過期");
    }
}

 set xiaofu 123 ex 3

監(jiān)聽到過期的key為：xiaofu

public void send(String delayTimes) {
        amqpTemplate.convertAndSend("order.pay.exchange", "order.pay.queue","大家好我是延遲數(shù)據(jù)", message -> {
            // 設(shè)置延遲毫秒值
            message.getMessageProperties().setExpiration(String.valueOf(delayTimes));
            return message;
        });
    }
}

/**
     * 延時(shí)隊(duì)列
     */
    @Bean(name = "order.delay.queue")
    public Queue getMessageQueue() {
        return QueueBuilder
                .durable(RabbitConstant.DEAD_LETTER_QUEUE)
                // 配置到期后轉(zhuǎn)發(fā)的交換
                .withArgument("x-dead-letter-exchange", "order.close.exchange")
                // 配置到期后轉(zhuǎn)發(fā)的路由鍵
                .withArgument("x-dead-letter-routing-key", "order.close.queue")
                .build();
    }

public HashedWheelTimer(ThreadFactory threadFactory, long tickDuration, TimeUnit unit, int ticksPerWheel) {
        this(threadFactory, tickDuration, unit, ticksPerWheel, true);
    }

public class NettyDelayQueue {
    public static void main(String[] args) {
        final Timer timer = new HashedWheelTimer(Executors.defaultThreadFactory(), 5, TimeUnit.SECONDS, 2);
        //定時(shí)任務(wù)
        TimerTask task1 = new TimerTask() {
            public void run(Timeout timeout) throws Exception {
                System.out.println("order1  5s 后執(zhí)行 ");
                timer.newTimeout(this, 5, TimeUnit.SECONDS);//結(jié)束時(shí)候再次注冊(cè)
            }
        };
        timer.newTimeout(task1, 5, TimeUnit.SECONDS);
        TimerTask task2 = new TimerTask() {
            public void run(Timeout timeout) throws Exception {
                System.out.println("order2  10s 后執(zhí)行");
                timer.newTimeout(this, 10, TimeUnit.SECONDS);//結(jié)束時(shí)候再注冊(cè)
            }
        };
        timer.newTimeout(task2, 10, TimeUnit.SECONDS);
        //延遲任務(wù)
        timer.newTimeout(new TimerTask() {
            public void run(Timeout timeout) throws Exception {
                System.out.println("order3  15s 后執(zhí)行一次");
            }
        }, 15, TimeUnit.SECONDS);
    }
}