springboot学习(十三)：RabbitMQ的使用实现消息延迟消费

旧城等待， 2022-03-27 13:07 796阅读 0赞

# 说明 #

本篇博文主要记录了springboot中rabbitmq的使用。使用rabbitmq实现消息消费处理失败后的延迟重试。

# 正文 #

rabbitmq不支持延迟任务，所以使用rabbitmq的支持的其他特性来实现延迟任务。这里主要使用了[消息的有效期(TTL)][TTL]和[死信路由DLX][DLX]。消息的有效期可以通过队列设置，也可以通过设置消息的相关属性。死信路由的本质是一个Topic路由器，通过在创建缓冲队列时声明"x-dead-letter-exchange"属性，将缓冲队列与DLX绑定，当消息在缓冲队列到期时，会自动通过DLX路由到与消息携带的路由关键字(routingKey)匹配的队列中。  
  简单来说，实现消息的失败重试的总体流程为：消息生产者将消息发送到普通队列，消费者接受到后进行处理，若处理失败，将该消息发送到缓冲队列，消息在队列到期后会自动的被发送到普通队列，再次消费。原理图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70]

## 1.配置类MQConfig ##

配置类用来创建生产者，消费者，创建队列，交换机和绑定关系。

@Configuration
    public class MQConfig {
    
        //创建消费者
        @Bean
        public Consumer consumer() {
            return new Consumer();
        }
    
        //创建生产者
        @Bean
        public Producer producer() {
            return new Producer();
        }
    
        //创建生产者的topic exchange
        @Bean
        public TopicExchange proExchange() {
            return new TopicExchange("proExc");
        }
    
        //创建消费者的fanout exchange
        @Bean
        public FanoutExchange conExchange() {
            return new FanoutExchange("conExc");
        }
    
        //创建缓冲队列的dead letter exchange (死信路由)
        @Bean
        public TopicExchange dlxExchange() {
            return new TopicExchange("dlxExc");
        }
    
        //创建普通(生产者)对列
        @Bean
        public Queue proQueue() {
            return new Queue("proQueue");
        }
    
        //创建缓冲队列, 声明dlx
        @Bean
        public Queue cacheQueue() {
            Map<String, Object> args = new HashMap<>();
            //dlx的名称必须与创建exchange的名称相同
            args.put("x-dead-letter-exchange","dlxExc");
            return QueueBuilder.nonDurable("delayqueue").withArguments(args).build();
        }
    
        //普通队列与生产者的 topic exchage 绑定
        @Bean
        public Binding bindingPtoP(@Qualifier("proQueue") Queue proQueue, @Qualifier("proExchange") TopicExchange proExchange) {
            return BindingBuilder.bind(proQueue).to(proExchange).with("email.#");
        }
    
        //缓冲队列与消息者的 fanout exchange 绑定
        @Bean
        public Binding bindingCtoF(@Qualifier("cacheQueue") Queue cacheQueue, @Qualifier("conExchange") FanoutExchange conExchange) {
            return BindingBuilder.bind(cacheQueue).to(conExchange);
        }
    
        //普通队列与死信路由 dlx 绑定
        @Bean
        public Binding bindingPtoD(@Qualifier("proQueue") Queue proQueue, @Qualifier("dlxExchange") TopicExchange dlxExchange) {
            return BindingBuilder.bind(proQueue).to(dlxExchange).with("email.#");
        }
    
    }

## 2.消费者Consumer ##

在设置消息的有效期时，可以通过MessageProperties来设置expiration,以ms为单位。消息失败被发送到缓冲队列时，这里使用了Fanout Exchange，此交换机会忽略消息携带的routingKey,但是DLX是一个Topic Exchange，到消息到期时，会根据消息携带的routingKey来将消息重新路由到相匹配的队列，所以在失败发送时，要指定routingKey。

public class Consumer {
    
        @Autowired
        private FanoutExchange conExchange;
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @RabbitListener(queues = {"proQueue"})
        public void receiveEmail(String email) throws UnsupportedEncodingException {
            System.out.println("receive a email " + email);
            String routingKey = "email";
            MessageProperties messageProperties = getMessageProperties();
            rabbitTemplate.convertAndSend(conExchange.getName(), routingKey, new Message((email + " try again").getBytes("utf-8"),messageProperties));
        }
    
        /**
         * 设置消息的相关参数
         * @return
         */
        private MessageProperties getMessageProperties() {
            MessageProperties messageProperties = new MessageProperties();
            //设置消息的有效期 ms为单位
            messageProperties.setExpiration("5000");
            messageProperties.setContentType(MessageProperties.CONTENT_TYPE_TEXT_PLAIN);
            return messageProperties;
        }
    }

## 3.生产者Producer ##

生产者将消息发送到一个交换机，这里使用了Topic交换机，也可以使用默认的交换机。Default Exchange是一个Direct类型的交换机，当所有队列被创建时都会自动的用与队列名相同的bindingKey来将队列与默认的交换机绑定。

public class Producer {
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        @Resource(name = "proExchange")
        private TopicExchange proExchange;
    
        public void sendEmail() {
            String message = "this is a email";
            String routingKey = "email";
            rabbitTemplate.convertAndSend(proExchange.getName(),routingKey,message);
            System.out.println("send successfully");
        }
    }

## 4.测试 ##

@RestController
    public class TestController {
    
        @Autowired
        private Producer producer;
    
        @RequestMapping("/send")
        public String test() {
            producer.sendEmail();
            return "test";
        }
    }

![在这里插入图片描述][20190113192905301.png]

这篇博文只是总结记录了springboot使用rabbitmq来实现延迟消费的用法，有关rabbitmq的更多信息，详见[RabbitMQ Tutorials][]。

参考资料：[http://www.rabbitmq.com/getstarted.html][RabbitMQ Tutorials]  
源码地址：[https://github.com/Edenwds/springboot\_study/tree/master/rabbitmq][https_github.com_Edenwds_springboot_study_tree_master_rabbitmq]

[TTL]: http://www.rabbitmq.com/ttl.html
[DLX]: http://www.rabbitmq.com/dlx.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3NpbmF0XzM2NTUzOTEz_size_16_color_FFFFFF_t_70]: /images/20220327/04e9c4cbd22d4945b6a8220a6336403f.png
[20190113192905301.png]: /images/20220327/7d51ceab129c48a381ba4046e10e08db.png
[RabbitMQ Tutorials]: http://www.rabbitmq.com/getstarted.html
[https_github.com_Edenwds_springboot_study_tree_master_rabbitmq]: https://github.com/Edenwds/springboot_study/tree/master/rabbitmq