Springboot使用RabbitMq延迟队列和死信队列详解

    //桶延迟队列
    @Bean(BUCKET_TTL_QUEUE)
    public Queue bucketTtlQueue(){
        Map<String,Object> deadParamsMap = new HashMap<>();
        // 设置死信队列的Exchange
        deadParamsMap.put("x-dead-letter-exchange",BUCKET_DEAD_EXCHANGE);
        //设置死信队列的RouteKey
        deadParamsMap.put("x-dead-letter-routing-key",BUCKET_DEAD_QUEUE);
        // 设置对接过期时间"x-message-ttl"
        deadParamsMap.put("x-message-ttl",60000*5);//5分钟
        // 设置对接可以存储的最大消息数量
        //deadParamsMap.put("x-max-length",10);
        return new Queue(BUCKET_TTL_QUEUE,true,false,false,deadParamsMap);
    }

    //桶延迟交换机
    @Bean(BUCKET_TTL_EXCHANGE)
    public DirectExchange bucketTtlExchange() {
        return new DirectExchange(BUCKET_TTL_EXCHANGE,true,false);
    }

    // 绑定
    @Bean
    public Binding bucketTtlBinding() {
        return BindingBuilder.bind(bucketTtlQueue())
                .to(bucketTtlExchange())
                .with(BUCKET_TTL_QUEUE);
    }

    /**
     * 死信队列
     *
     * @return
     */
    @Bean(BUCKET_DEAD_QUEUE)
    public Queue bucketDeadQueue() {
        //属性参数 队列名称 是否持久化
        return new Queue(BUCKET_DEAD_QUEUE, true);
    }

    /**
     * 死信队列交换机
     *
     * @return
     */
    @Bean(BUCKET_DEAD_EXCHANGE)
    public DirectExchange bucketDeadExchange() {
        return new DirectExchange(BUCKET_DEAD_EXCHANGE,true,false);
    }

    /**
     * 给死信队列绑定交换机
     *
     * @return
     */
    @Bean
    public Binding bucketDeadBinding() {
        return BindingBuilder.bind(bucketDeadQueue()).to(bucketDeadExchange()).with(BUCKET_DEAD_QUEUE);
    }

/**
 * @description:死信队列监听器，用来删除过期的桶
 * @author manchao
 * @date 2022/2/17 9:52
 */
@Configuration
public class BucketDeadConsumer {

    @Autowired
    private CachingConnectionFactory cachingConnectionFactory;

    @Autowired
    private MinioTemplate minioTemplate;

    @Bean
    public SimpleMessageListenerContainer BucketDeadListenerContainer() {
        SimpleMessageListenerContainer container = new  SimpleMessageListenerContainer(cachingConnectionFactory);
        // 监听队列名
        container.setQueueNames(MyMqConfig.BUCKET_DEAD_QUEUE);
        // 当前消费者数量 开启几个线程去处理数据 支持运行时动态修改
        container.setConcurrentConsumers(5);
        // 最大消费者数量  ,  消息堵塞太多的时候，会帮我自动扩展到我的最大消费者数量
        container.setMaxConcurrentConsumers(10);
        // 是否重回队列
        container.setDefaultRequeueRejected(true);
        // 手动确认
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置监听器
        container.setMessageListener(new ChannelAwareMessageListener(){
            @Override
            public void onMessage(Message message, Channel channel) throws IOException {
                // 消息的唯一性ID deliveryTag:该消息的index 自增长
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                byte[] messageBody = message.getBody();
                String s = new String(messageBody);
                System.out.println("消息: " + s);
                System.out.println("消息来自: "+message.getMessageProperties().getConsumerQueue());
                System.out.println("交换机: "+message.getMessageProperties().getReceivedExchange());
                //删除桶
                try {
                    minioTemplate.removeBuckets(s, "");
                    channel.basicAck(deliveryTag, false);
                } catch (IOException e) {
                    e.printStackTrace();
                    channel.basicReject(deliveryTag, false);
                }
            }
        });
        return container;
    }
}