四、RabbitMQ队列绑定以及死信队列

队列绑定以及死信队列

一、常量

package com.strap.mydemo.constants;

/**
 * <p></p>
 *
 * @author strap
 */
public interface MyDemoConstants {

    interface MQConstants {
        /**
         * 业务队列名称
         */
        String QUEUE_NAME = "demo-queue";

        /**
         * 业务交换器名称
         */
        String EXCHANGE_NAME = "demo-exchange";

        /**
         * 业务路由键名称
         */
        String ROUTING_KEY_NAME = "routing-key.demo";

        /**
         * 死信队列名称
         */
        String DEAD_QUEUE = "dead-queue";

        /**
         * 死信交换器名称
         */
        String DEAD_EXCHANGE = "dead_exchange";

        /**
         * 死信路由键名称
         */
        String DEAD_ROUTING_KEY = "dead_routing_key";

        /**
         * 死信交换器的标志
         */
        String DEAD_EXCHANGE_LABEL = "x-dead-letter-exchange";

        /**
         * 死信路由键的标志
         */
        String DEAD_ROUTING_KEY_LABEL = "x-dead-letter-routing-key";

    }

}

二、队列初始化及绑定死信队列

1. 使用配置类完成

/**
 * <p>mq配置文件</p>
 *
 * @author strap
 */
@Configuration
public class RabbitMQConfig {

    /**
     * 定义一个业务队列并配置死信队列
     */
    @Bean
    public Queue demoQueue() {
        Map<String, Object> args = new HashMap<>();
        // x-dead-letter-exchange：这里声明当前业务队列绑定的死信交换机
        args.put(MyDemoConstants.MQConstants.DEAD_EXCHANGE_LABEL, MyDemoConstants.MQConstants.DEAD_EXCHANGE);
        // x-dead-letter-routing-key：这里声明当前业务队列的死信路由 key
        args.put(MyDemoConstants.MQConstants.DEAD_ROUTING_KEY_LABEL, MyDemoConstants.MQConstants.DEAD_ROUTING_KEY);
        return new Queue(MyDemoConstants.MQConstants.QUEUE_NAME, true, false, false, args);
    }

    /**
     * 定义一个业务队列的交换机
     */
    @Bean
    public TopicExchange demoTopicExchange() {
        return new TopicExchange(MyDemoConstants.MQConstants.EXCHANGE_NAME, true, false);
    }

    /**
     * 定义业务队列的路由键绑定
     */
    @Bean
    public Binding demoBinding() {
        return BindingBuilder.bind(demoQueue()).to(demoTopicExchange()).with(MyDemoConstants.MQConstants.ROUTING_KEY_NAME);
    }


    /**
     * 定义业务队列使用的死信队列
     */
    @Bean
    public Queue deadLetterQueue() {
        return new Queue(MyDemoConstants.MQConstants.DEAD_QUEUE);
    }

    /**
     * 定义死信队列使用的交换机
     */
    @Bean
    public TopicExchange deadLetterTopicExchange() {
        return new TopicExchange(MyDemoConstants.MQConstants.DEAD_EXCHANGE);
    }

    /**
     * 定义死信队列的路由键绑定
     */
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder.bind(deadLetterQueue()).to(deadLetterTopicExchange()).with(MyDemoConstants.MQConstants.DEAD_ROUTING_KEY);
    }

    /**
     * RabbitAdmin 可以用来申明绑定申明的队列以及交换机
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 设置自启动
        rabbitAdmin.setAutoStartup(true);
        // 申明绑定死信交换机和对列
        rabbitAdmin.declareExchange(deadLetterTopicExchange());
        rabbitAdmin.declareQueue(deadLetterQueue());
        // 申明绑定业务交换机和对列
        rabbitAdmin.declareExchange(demoTopicExchange());
        rabbitAdmin.declareQueue(demoQueue());
        return rabbitAdmin;
    }

}

消费者监听使用

/**
 * <p>消费者</p>
 *
 * @author strap
 */
@Component
@Log4j
public class MqConsumer {

    public MqConsumer() {
    }

    @RabbitListener(queues = MyDemoConstants.MQConstants.QUEUE_NAME)
    @RabbitHandler 
    public void consume(@Payload DocMessage docMessage, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        log.info("消费者已拿到消息：" + docMessage);
    }

    @RabbitListener(queues = MyDemoConstants.MQConstants.DEAD_QUEUE)
    @RabbitHandler
    public void consumeDeadLetter(@Payload DocMessage docMessage, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        log.info("死信消息已经收到：" + docMessage.getMessageId());
    }

}

2. 仅使用@RabbitListener完成所有配置

不需要配置类，消费者代码如下

/**
 * <p>消费者</p>
 *
 * @author strap
 */
@Component
@Log4j
public class MqConsumer {

    public MqConsumer() {
    }

    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = MyDemoConstants.MQConstants.QUEUE_NAME, durable = "true",
                            arguments = {
                            @Argument(name = MyDemoConstants.MQConstants.DEAD_EXCHANGE_LABEL, value = MyDemoConstants.MQConstants.DEAD_EXCHANGE),
                            @Argument(name = MyDemoConstants.MQConstants.DEAD_ROUTING_KEY_LABEL, value = MyDemoConstants.MQConstants.DEAD_ROUTING_KEY)
                    }),
                    exchange = @Exchange(name = MyDemoConstants.MQConstants.EXCHANGE_NAME, type = "topic"),
                    key = MyDemoConstants.MQConstants.ROUTING_KEY_NAME
            )
    )
    @RabbitHandler 
    public void consume(@Payload DocMessage docMessage, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        log.info("消费者已拿到消息：" + docMessage);
    }

    /**
     * 处理死信消息
     */
    @RabbitListener(
            bindings = @QueueBinding(
                    value = @Queue(value = MyDemoConstants.MQConstants.DEAD_QUEUE, durable = "true"),
                    exchange = @Exchange(name = MyDemoConstants.MQConstants.DEAD_EXCHANGE, type = "topic"),
                    key = MyDemoConstants.MQConstants.DEAD_ROUTING_KEY
            )
    )
    @RabbitHandler
    public void consumeDeadLetter(@Payload DocMessage docMessage, @Headers Map<String, Object> headers, Channel channel) throws Exception {
        log.info("死信消息已经收到：" + docMessage.getMessageId());
    }

}