RabbitMQ可靠消息机制

向右看齐 2023-10-11 14:37 30阅读 0赞

### 发送端确认 ###

**在配置文件中配置以下内容**

#开启发送端确认机制
    spring.rabbitmq.publisher-confirms=true

**编写发送消息到达服务器后的回调setConfirmCallback**  
*注意，这个回调只能说明消息到达了服务器，而不能说明消息发送到了队列*

@Configuration
    public class MyRabbitConfig {
        
    
        @Autowired
        public RabbitTemplate rabbitTemplate;
    
        /**
         * 为容器中的rabbitTemplate设置ConfirmCallback
         * 设置确认回调
         */
        @PostConstruct
        public void initRabbitTemplate(){
        
            System.out.println("initRabbitTemplate.....");
            //当消息发送到broker服务器时回调
            rabbitTemplate.setConfirmCallback((correlationData,ack,cause)->{
        
                /**
                 * correlationData：消息的唯一id
                 * ack:消息是否成功发送
                 * cause:失败的原因
                 */
                System.out.println("消息确认" + correlationData + ":::" + ack + ":::" + cause);
            });
        }

***correlationData：消息的唯一Id，在发送消息时，可以指定这条消息的唯一Id***

rabbitTemplate.convertAndSend("article-exchange","article-key",articleEntity,new CorrelationData(UUID.randomUUID().toString()));

***在发送消息的时候除了给消息指定一条唯一id，还可以将这个唯一id和其对应的消息保存到数据库一份，这样在触发回调时,如果哪个消息发送失败，我们可以根据它的id，去数据库查找到对应的交换机、路由键和消息再次发送***  
![在这里插入图片描述][20200923163745240.png_pic_center]  
***如果消息抵达了服务器，则ack就是true，如果为false，则我们可以获取这个correlationData，再次进行发送***

*  设置了该回调后发送消息查看控制台  
    ![在这里插入图片描述][20200923160615514.png_pic_center]**开启消息入队确认机制**

#开启消息入队确认机制
    spring.rabbitmq.publisher-returns=true
    #当消息抵达队列，以异步方式优先回调returnConfirm
    spring.rabbitmq.template.mandatory=true

**编写消息抵达队列的回调**

### 消费端确认 ###

***消费端默认是自动确认的，只要消息被接受，服务端就会移除，这样导致的问题就是如果消息生产者发送了5条消息，如果消费者只消费了1条，这时消费者宕机了，那就会丢失另外4条消息，因为开启了channle。***

*  配置消费者端手动签收

#消费者端开启消息手动签收
    spring.rabbitmq.listener.simple.acknowledge-mode=manual

*  修改消费端签收的代码，改为手动签收

@RabbitListener(queues = {
        "article-queue"})
        public void handlerMessage(Message message, ApUserEntity apUserEntity, Channel channel){
        
            System.out.println("收到了消息" + apUserEntity);
            /**
             * 手动确认消息
             */
            long deliveryTag = message.getMessageProperties().getDeliveryTag();
            try {
        
                if (deliveryTag % 2 == 0){
        
                    //模拟拒绝签收
                    //long deliveryTag, boolean multiple, boolean requeue
                    channel.basicNack(deliveryTag,false,true); //重新入队发送
                }
                //手动签收
                channel.basicAck(deliveryTag,false);
            } catch (IOException e) {
        
                e.printStackTrace();
            }
        }

手动签收为channle的basicAck方法。

*  此外在运行过程中有可能会出现Listener method could not be invoked with the incoming message异常，这主要是由于发送端和接收端的系列化方式不一致导致的，我们可以给容器中放入一个序列化的bean来解决，注意，两端的序列化方式应一致。

@Bean
        public MessageConverter messageConverter(){
        
            return new Jackson2JsonMessageConverter();
        }

[20200923163745240.png_pic_center]: https://img-blog.csdnimg.cn/20200923163745240.png#pic_center
[20200923160615514.png_pic_center]: https://img-blog.csdnimg.cn/20200923160615514.png#pic_center