MQ消息可靠抵达

╰半橙微兮° 2023-10-11 15:14 66阅读 0赞

### 发送者设置回调 ###

*  设置消息抵达消息服务器的回调

#开启消息服务器接收回调
    spring.rabbitmq.publisher-confirm-type=correlated

注意：此回调无论是否抵达消息服务器都会回调

//消息成功抵达消息服务器或者未抵达服务器的回调，不是消息进入队列时的回调
            //TODO 注意此方法无论是否抵达消息服务器均回调，可通过 boolean ack 判断是否成功抵达
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        
                    if (ack){
        
                        //发送至消息服务器成功
                        System.out.println("消息发送成功:::消息id是" + correlationData.getId());
                    }else {
        
                        //可能由于未连接服务器成功，或者路由器填写错误
                        System.out.println("消息发送失败:::消息id是" + correlationData.getId());
                        //根据消息id获取数据库中的消息详细信息，可再次发送
                    }
                }
            });

*  设置消息入队的回调

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

这个回调是当消息抵达了消息服务器，但由于路由键错误导致不能抵达队列触发此回调,也可以在此方法中获取消息的id，去数据库中查询消息详细信息，再次发送

//设置消息抵达队列的回调，如果能成功抵达执行队列，则不回调，否则回调
            //例如在发送消息时将 routing-key填写错误，导致抵达了交换器但不能进入队列
           rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
        
               @Override
               public void returnedMessage(ReturnedMessage returnedMessage) {
        
                   System.out.println("未抵达队列的消息："+returnedMessage.getMessage().toString());
               }
           });

### 控制台说明 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70]

### 设置手动签收 ###

*  设置消息接收方手动签收消息，而非自动签收

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

消息接收方在执行完业务代码时要手动签收消息

/**
         *  消息手动签收
         */
        @RabbitListener(queues = {
        "queue-java"})
        public void handlerMessage(Message message, User user, Channel channel) throws IOException {
        
            long messageId = message.getMessageProperties().getDeliveryTag();
            try {
        
                // 收到消息时的业务逻辑处理.....
                System.out.println("接收到消息：" + user);
    
                //如果执行成功
                channel.basicAck(messageId,false); //只签收本条消息
            }catch (Exception e){
        
                channel.basicNack(messageId,false,true); // 出现异常拒收消息，重新入队
                e.printStackTrace();
            }
        }

这样发送方能保证将消息发送至消息队列，接收方在接收完成后手动签收，保证了消息可靠抵达。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20210104092425863.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzQzNzUwNjU2,size_16,color_FFFFFF,t_70