//桶延遲隊(duì)列
    @Bean(BUCKET_TTL_QUEUE)
    public Queue bucketTtlQueue(){
        Map<String,Object> deadParamsMap = new HashMap<>();
        // 設(shè)置死信隊(duì)列的Exchange
        deadParamsMap.put("x-dead-letter-exchange",BUCKET_DEAD_EXCHANGE);
        //設(shè)置死信隊(duì)列的RouteKey
        deadParamsMap.put("x-dead-letter-routing-key",BUCKET_DEAD_QUEUE);
        // 設(shè)置對(duì)接過(guò)期時(shí)間"x-message-ttl"
        deadParamsMap.put("x-message-ttl",60000*5);//5分鐘
        // 設(shè)置對(duì)接可以存儲(chǔ)的最大消息數(shù)量
        //deadParamsMap.put("x-max-length",10);
        return new Queue(BUCKET_TTL_QUEUE,true,false,false,deadParamsMap);
    }

    //桶延遲交換機(jī)
    @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);
    }

    /**
     * 死信隊(duì)列
     *
     * @return
     */
    @Bean(BUCKET_DEAD_QUEUE)
    public Queue bucketDeadQueue() {
        //屬性參數(shù) 隊(duì)列名稱(chēng) 是否持久化
        return new Queue(BUCKET_DEAD_QUEUE, true);
    }

    /**
     * 死信隊(duì)列交換機(jī)
     *
     * @return
     */
    @Bean(BUCKET_DEAD_EXCHANGE)
    public DirectExchange bucketDeadExchange() {
        return new DirectExchange(BUCKET_DEAD_EXCHANGE,true,false);
    }

    /**
     * 給死信隊(duì)列綁定交換機(jī)
     *
     * @return
     */
    @Bean
    public Binding bucketDeadBinding() {
        return BindingBuilder.bind(bucketDeadQueue()).to(bucketDeadExchange()).with(BUCKET_DEAD_QUEUE);
    }

/**
 * @description:死信隊(duì)列監(jiān)聽(tīng)器，用來(lái)刪除過(guò)期的桶
 * @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);
        // 監(jiān)聽(tīng)隊(duì)列名
        container.setQueueNames(MyMqConfig.BUCKET_DEAD_QUEUE);
        // 當(dāng)前消費(fèi)者數(shù)量 開(kāi)啟幾個(gè)線(xiàn)程去處理數(shù)據(jù) 支持運(yùn)行時(shí)動(dòng)態(tài)修改
        container.setConcurrentConsumers(5);
        // 最大消費(fèi)者數(shù)量  ,  消息堵塞太多的時(shí)候，會(huì)幫我自動(dòng)擴(kuò)展到我的最大消費(fèi)者數(shù)量
        container.setMaxConcurrentConsumers(10);
        // 是否重回隊(duì)列
        container.setDefaultRequeueRejected(true);
        // 手動(dòng)確認(rèn)
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 設(shè)置監(jiān)聽(tīng)器
        container.setMessageListener(new ChannelAwareMessageListener(){
            @Override
            public void onMessage(Message message, Channel channel) throws IOException {
                // 消息的唯一性ID deliveryTag:該消息的index 自增長(zhǎng)
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                byte[] messageBody = message.getBody();
                String s = new String(messageBody);
                System.out.println("消息: " + s);
                System.out.println("消息來(lái)自: "+message.getMessageProperties().getConsumerQueue());
                System.out.println("交換機(jī): "+message.getMessageProperties().getReceivedExchange());
                //刪除桶
                try {
                    minioTemplate.removeBuckets(s, "");
                    channel.basicAck(deliveryTag, false);
                } catch (IOException e) {
                    e.printStackTrace();
                    channel.basicReject(deliveryTag, false);
                }
            }
        });
        return container;
    }
}