@Service
public class DelayQueueService {
    private static final String DELAY_QUEUE_KEY = "delay_queue";


    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    public void addToDelayQueue(String message,long delayTime){
        redisTemplate.opsForZSet().add(DELAY_QUEUE_KEY,message,System.currentTimeMillis()+delayTime);
    }

    public void processDelayedMessage(){
        //reverseRangeByScore 從高到低
        //rangeByScore 從低到高
        Set<String> messages = redisTemplate.opsForZSet().rangeByScore(DELAY_QUEUE_KEY, 0, System.currentTimeMillis());
        for(String message:messages){
            //處理消息
            System.out.println(message);
            redisTemplate.opsForZSet().remove(DELAY_QUEUE_KEY,message);
        }

    }
}

@Component
public class DelayQueueSchedule {
    @Autowired
    private DelayQueueService delayQueueService;


    // 每隔一段時(shí)間進(jìn)行輪詢并處理延遲消息
    @Scheduled(fixedDelay = 1000)
    public void pollAndProcessDelayedMessages() {
        delayQueueService.pollAndProcessDelayedMessages();
    }
}

@Configuration
public class DeadQueueConfig {
    //普通交換機(jī)及隊(duì)列
    public static final String X_EXCHANGE = "X";
    public static final String QUEUE_A = "QA";
    public static final String QUEUE_B = "QB";
    //死信交換機(jī)及隊(duì)列
    public static final String Y_DEAD_LETTER_EXCHANGE = "Y";
    public static final String DEAD_LETTER_QUEUE = "QD";
    //通用隊(duì)列
    public static final String QUEUE_C = "QC";


    // 聲明 xExchange
    @Bean("xExchange")
    public DirectExchange xExchange() {
        return new DirectExchange(X_EXCHANGE);
    }

    //聲明隊(duì)列 A ttl 為 10s 并綁定到對(duì)應(yīng)的死信交換機(jī)
    @Bean("queueA")
    public Queue queueA() {
        Map<String, Object> args = new HashMap<>(3);
        //聲明當(dāng)前隊(duì)列綁定的死信交換機(jī)
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //聲明當(dāng)前隊(duì)列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //聲明隊(duì)列的 TTL
        args.put("x-message-ttl", 10000);
        return QueueBuilder.durable(QUEUE_A).withArguments(args).build();
    }
    //聲明隊(duì)列A綁定X交換機(jī)  路由為XA
    @Bean
    public Binding queueABingX(@Qualifier("queueA") Queue queueA,
                               @Qualifier("xExchange") DirectExchange xExchange){
        return BindingBuilder.bind(queueA).to(xExchange).with("XA");
    }

    //聲明隊(duì)列 B ttl 為 40s 并綁定到對(duì)應(yīng)的死信交換機(jī)
    @Bean("queueB")
    public Queue queueB() {
        Map<String, Object> args = new HashMap<>(3);
        //聲明當(dāng)前隊(duì)列綁定的死信交換機(jī)
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //聲明當(dāng)前隊(duì)列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        //聲明隊(duì)列的 TTL
        args.put("x-message-ttl", 40000);
        return QueueBuilder.durable(QUEUE_B).withArguments(args).build();
    }

    //聲明隊(duì)列 B 綁定 X 交換機(jī)
    @Bean
    public Binding queuebBindingX(@Qualifier("queueB") Queue queue1B,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queue1B).to(xExchange).with("XB");
    }

    //聲明通用隊(duì)列C 不設(shè)ttl，由消費(fèi)者決定ttl
    @Bean("queueC")
    public Queue queueC() {
        Map<String, Object> args = new HashMap<>(3);
        //聲明當(dāng)前隊(duì)列綁定的死信交換機(jī)
        args.put("x-dead-letter-exchange", Y_DEAD_LETTER_EXCHANGE);
        //聲明當(dāng)前隊(duì)列的死信路由 key
        args.put("x-dead-letter-routing-key", "YD");
        return QueueBuilder.durable(QUEUE_C).withArguments(args).build();
    }
    // 聲明隊(duì)列 C 綁定 X 交換機(jī)
    @Bean
    public Binding queuecBindingX(@Qualifier("queueC") Queue queueC,
                                  @Qualifier("xExchange") DirectExchange xExchange) {
        return BindingBuilder.bind(queueC).to(xExchange).with("XC");
    }

    // 聲明 死信隊(duì)列交換機(jī)
    @Bean("yExchange")
    public DirectExchange yExchange() {
        return new DirectExchange(Y_DEAD_LETTER_EXCHANGE);
    }
    //聲明死信隊(duì)列 QD
    @Bean("queueD")
    public Queue queueD() {
        return new Queue(DEAD_LETTER_QUEUE,true);
    }
    //聲明死信隊(duì)列 QD 綁定關(guān)系
    @Bean
    public Binding deadLetterBindingQAD(@Qualifier("queueD") Queue queueD,
                                        @Qualifier("yExchange") DirectExchange yExchange) {
        return BindingBuilder.bind(queueD).to(yExchange).with("YD");
    }

}

@Component
@Slf4j
public class DeadQueueConsumer {

    @RabbitListener(queues = "QD")
    public void receiveD(Message message, Channel channel) throws IOException {
        String msg = new String(message.getBody());
        log.info("當(dāng)前時(shí)間：{},收到死信隊(duì)列信息：{}", new Date().toString(), msg);
    }
}

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @GetMapping("/sendMsg/{message}")
    public String sendMsg(@PathVariable String message){
        log.info("當(dāng)前時(shí)間：{},發(fā)送一條信息給兩個(gè) TTL 隊(duì)列:{}", new Date(), message);
        rabbitTemplate.convertAndSend("X", "XA", "消息來自 ttl 為 10S 的隊(duì)列: " + message);
        rabbitTemplate.convertAndSend("X", "XB", "消息來自 ttl 為 40S 的隊(duì)列: " + message);
        return "finish";
    }

@GetMapping("/send/{message}/{ttlTime}")
    public void sendMsg(@PathVariable String message, @PathVariable String ttlTime) {
        rabbitTemplate.convertAndSend("X", "XC", message, correlationData -> {
            correlationData.getMessageProperties().setExpiration(ttlTime);
            return correlationData;
        });
        log.info("當(dāng)前時(shí)間：{},發(fā)送一條時(shí)長{}毫秒 TTL 信息給隊(duì)列 C:{}", new Date(), ttlTime, message);
    }