RabbitMQ--集成Springboot--3.2--消息确认机制

RabbitMQ–集成Springboot–3.2–消息确认机制

---

代码位置

https://gitee.com/DanShenGuiZu/learnDemo/tree/master/rabbitMq-learn/rabbitMq-01

1、介绍

1. 消息确认有2种，如下所示
   1. 生产者 消息确认
   2. 消费者 消息确认
2. 要使用消息确认，还需要修改配置文件

1.1、修改配置文件

server:
  port: 8080
spring:
  #给项目来个名字
  application:
    name: rabbitmq-learn
  #配置rabbitMq 服务器
  rabbitmq:
    host: 192.168.187.171
    port: 5672
    username: root
    password: root
    #虚拟host 可以不设置,使用server默认host
    virtual-host: test
    #确认消息已发送到队列(Queue)
    publisher-returns: true
    #确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated

2、生产者 消息确认

2.1、配置 消息确认回调函数

/**
 * MQ发送确认
 *
 * @author <a href="920786312@qq.com">周飞</a>
 * @class: ConfirmCallbackImpl
 * @date 2021/9/8 9:51
 * @Verson 1.0 -2021/9/8 9:51
 * @see
 */
public class ConfirmCallbackImpl implements RabbitTemplate.ConfirmCallback {
    Logger logger = LoggerFactory.getLogger(getClass());
    
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        // 消息没有发送成功
        
        if (ack) {
            logger.info("ConfirmCallbackImpl  消息发送成功，ack：" + ack);
            return;
        }
        logger.error("ConfirmCallbackImpl  消息发送失败，ack：" + ack);
        logger.error("ConfirmCallbackImpl  相关数据：" + correlationData);
        logger.error("ConfirmCallbackImpl  确认情况：" + "确认情况：" + ack);
        logger.error("ConfirmCallbackImpl  原因：" + cause);
        
    }
}

/**
 * 发送回调函数，发送消息成功是不会执行这个回调函数的
 *
 * @author <a href="920786312@qq.com">周飞</a>
 * @class: ReturnCallbackImpl
 * @date 2021/9/8 9:51
 * @Verson 1.0 -2021/9/8 9:51
 * @see
 */
public class ReturnCallbackImpl implements RabbitTemplate.ReturnCallback {
    Logger logger = LoggerFactory.getLogger(getClass());
    
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        logger.error("ReturnCallbackImpl 消息：" + message);
        logger.error("ReturnCallbackImpl 回应码：" + replyCode);
        logger.error("ReturnCallbackImpl 回应信息：" + replyText);
        logger.error("ReturnCallbackImpl 交换机：" + exchange);
        logger.error("ReturnCallbackImpl 路由键：" + routingKey);
        
    }
}

/**
 * 配置Rabbit
 **/
@Configuration
public class RabbitConfig {

    /**
     * 配置RabbitMQ 模板
     **/
    @Bean
    public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate();
        rabbitTemplate.setConnectionFactory(connectionFactory);
        // 设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数
        rabbitTemplate.setMandatory(true);

        //配置回调函数
        rabbitTemplate.setConfirmCallback(new ConfirmCallbackImpl());
        rabbitTemplate.setReturnCallback(new ReturnCallbackImpl() );
        return rabbitTemplate;
    }
    
}

到这里，生产者推送消息的消息确认调用回调函数已经完毕。

2.2、消息确认回调函数 的 触发场景

1. 消息推送到server，但是在server里找不到交换机
2. 消息推送到server，找到交换机了，但是没找到队列
3. 消息推送到sever，交换机和队列啥都没找到
4. 消息推送成功

2.2.1、代码

@Configuration
public class ACK_Config {
    
    // 声明队列
    // @param1 队列名称
    // @param2 是否持久化 持久化：RabbitMQ服务器重启，队列还存在；反之，不存在。
    // 持久化的队列中的消息会存盘，不会随着服务器的重启会消失
    // @param3 排他性 是否独占一个队列（一般不会），true:只能被当前创建的连接使用，而且当连接关闭后队列即被删除
    // @param4 是否自动删除 随着最后一个消费者消费消息完毕后，是否自动删除这个队列
    // @param5 携带一些附加信息 供其它消费者获取
    @Bean
    public Queue ack_direct_queue() {
        return new Queue("ack_direct_queue", true, false, false, null);
        
    }
    
    // 声明交换机
    @Bean
    public DirectExchange ack_direct_exchange() {
        return new DirectExchange("ack_direct_exchange");
    }
    
    // 声明交换机与队列之间的关系
    @Bean
    public Binding bindingAck_direct_queue(Queue ack_direct_queue, DirectExchange ack_direct_exchange) {
        return BindingBuilder.bind(ack_direct_queue).to(ack_direct_exchange).with("ack");
    }
    
    @Bean
    // 验证 消息推送到server，找到交换机了，但是没找到队列
    DirectExchange lonelyDirectExchange() {
        return new DirectExchange("lonelyDirectExchange");
    }
}

@Component
public class ACK_Consumer {
    
    @RabbitListener(queues = { "ack_direct_queue" })
    public void receive(Message message, Channel channel) throws Exception {
        // 消息id
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.println("ack_direct_queue队列消息：" + new String(message.getBody()));
    }
    
}

@RestController
@RequestMapping("/ack")
public class ACK_Producer {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    @RequestMapping("/sendMsg")
    public String sendMsg() {
        String msg = "消息推送到server，但是在server里找不到交换机";
        rabbitTemplate.convertAndSend("non-existent-exchange", "ack", msg);
        return "ok";
    }
    
    @RequestMapping("/sendMsg2")
    public String sendMsg2() {
        String msg = "消息推送到server，找到交换机了，但是没找到队列";
        rabbitTemplate.convertAndSend("lonelyDirectExchange", "ack", msg);
        return "ok";
    }
    
    @RequestMapping("/sendMsg3")
    public String sendMsg3() {
        String msg = "消息推送到sever，交换机和队列啥都没找到";
        rabbitTemplate.convertAndSend("non-existent-exchange", "non-routingkey", msg);
        return "ok";
    }
    
    @RequestMapping("/sendMsg4")
    public String sendMsg4() {
        String msg = "消息推送成功";
        rabbitTemplate.convertAndSend("ack_direct_exchange", "ack", msg);
        return "ok";
    }
}

2.2.2、测试

2.2.2.1、消息推送到server，但是在server里找不到交换机

http://127.0.0.1:8080/ack/sendMsg1

日志

2022-08-27 16:52:12.104 ERROR 72720 --- [68.187.171:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'test', class-id=60, method-id=40)
2022-08-27 16:52:12.105 ERROR 72720 --- [nectionFactory1] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  消息发送失败，ack：false
2022-08-27 16:52:12.105 ERROR 72720 --- [nectionFactory1] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  相关数据：null
2022-08-27 16:52:12.105 ERROR 72720 --- [nectionFactory1] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  确认情况：确认情况：false
2022-08-27 16:52:12.105 ERROR 72720 --- [nectionFactory1] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'test', class-id=60, method-id=40)

1. 报错原因，没有找到交换机’non-existent-exchange’，因为我们就没有配置这个交换机
2. 这种情况触发的是 ConfirmCallback 回调函数。

2.2.2.2、消息推送到server，找到交换机了，但是没找到队列

http://127.0.0.1:8080/ack/sendMsg2

日志

2022-08-27 16:53:53.108  INFO 72720 --- [nectionFactory3] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  消息发送成功，ack：true
2022-08-27 16:53:53.108 ERROR 72720 --- [nectionFactory4] f.z.r.b.a.c.producer.ReturnCallbackImpl  : ReturnCallbackImpl 消息：(Body:'消息推送到server，找到交换机了，但是没找到队列' MessageProperties [headers={}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
2022-08-27 16:53:53.108 ERROR 72720 --- [nectionFactory4] f.z.r.b.a.c.producer.ReturnCallbackImpl  : ReturnCallbackImpl 回应码：312
2022-08-27 16:53:53.108 ERROR 72720 --- [nectionFactory4] f.z.r.b.a.c.producer.ReturnCallbackImpl  : ReturnCallbackImpl 回应信息：NO_ROUTE
2022-08-27 16:53:53.108 ERROR 72720 --- [nectionFactory4] f.z.r.b.a.c.producer.ReturnCallbackImpl  : ReturnCallbackImpl 交换机：lonelyDirectExchange
2022-08-27 16:53:53.109 ERROR 72720 --- [nectionFactory4] f.z.r.b.a.c.producer.ReturnCallbackImpl  : ReturnCallbackImpl 路由键：ack

1. 消息是成功推送到服务器了，所以ConfirmCallback对消息确认情况是true
2. 报错原因：消息是推送到了交换机成功了，但是在路由分发给队列的时候，找不到队列，所以报了错误 NO_ROUTE。
   1. 找不到队列的原因是我们配置 lonelyDirectExchange交换机，就没配置绑定队列
3. 这种情况触发的是 ConfirmCallback和RetrunCallback两个回调函数。

2.2.2.3、消息推送到sever，交换机和队列啥都没找到

http://127.0.0.1:8080/ack/sendMsg3

日志

2022-08-27 16:57:24.035 ERROR 72720 --- [68.187.171:5672] o.s.a.r.c.CachingConnectionFactory       : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'test', class-id=60, method-id=40)
2022-08-27 16:57:24.035 ERROR 72720 --- [nectionFactory6] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  消息发送失败，ack：false
2022-08-27 16:57:24.035 ERROR 72720 --- [nectionFactory6] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  相关数据：null
2022-08-27 16:57:24.035 ERROR 72720 --- [nectionFactory6] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  确认情况：确认情况：false
2022-08-27 16:57:24.035 ERROR 72720 --- [nectionFactory6] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost 'test', class-id=60, method-id=40)

1. 报错原因，没有找到交换机’non-existent-exchange’，因为我们就没有配置这个交换机
2. 这种情况触发的是 ConfirmCallback 回调函数。

2.2.2.4、消息推送成功

http://127.0.0.1:8080/ack/sendMsg4

日志

2022-08-27 16:59:58.176  INFO 72720 --- [nectionFactory7] f.z.r.b.a.c.p.ConfirmCallbackImpl        : ConfirmCallbackImpl  消息发送成功，ack：true

3、消费者 消息确认

3.1、确认机制主要存在三种模式

3.1.1、自动确认

1. 默认的消息确认机制
2. RabbitMQ成功将消息发出后(将消息成功写入TCP Socket中)，就认为本次投递已经被正确处理，不管消费者端是否成功处理本次投递。
3. 如果消费端消费逻辑抛出异常，也就是消费端没有处理成功这条消息，那么就相当于丢失了消息。

3.1.2、根据情况确认

这个不做介绍

3.1.3、手动确认

1. 手动ACK
2. 消费者收到消息后，手动调用以下方法后，RabbitMQ收到这些消息后，才认为本次投递成功。
   1. basic.ack
   2. basic.nack
   3. basic.reject

3.2、手动确认确认方法

3.2.1、basic.ack(deliveryTag, false)

1. 表示消息已经被正确处理
2. 第1个参数：消息id
3. 第2个参数：表示是指是否针对多条消息
   1. true：小于 当前这条消息的delivery_tag 的所有消息，都确认
   2. false：只确认当前这条消息

3.2.2、channel.basicNack(deliveryTag, false, true)

1. 用于否定确认
2. 第1个参数：消息id
3. 第2个参数：表示是指是否针对多条消息
   1. true：小于 当前这条消息的delivery_tag 的所有消息，都拒绝确认
   2. false：只拒绝当前这条消息
4. 第3个参数：是否重新入列
   1. true
      1. 消息重新丢回队列里，下次还会消费这消息
      2. 这个配置要谨慎，要考虑 消费-入列-消费-入列 这样循环情况，会导致消息积压。
   2. false
      1. 消息不丢回队列里，服务器丢失这条消息

3.2.3、channel.basicReject(deliveryTag, false)

1. 拒绝消费当前消息
2. 与basic.nack相比有一个限制:一次只能拒绝单条消息
3. 第1个参数：消息id
4. 第2个参数：是否重新入列
   1. true
      1. 消息重新丢回队列里，下次还会消费这消息
      2. 这个配置要谨慎，要考虑 消费-入列-消费-入列 这样循环情况，会导致消息积压。
   2. false
      1. 消息不丢回队列里，服务器丢失这条消息

3.3、代码

@Configuration
public class MessageListenerConfig {
    
    @Autowired
    private CachingConnectionFactory connectionFactory;
    @Autowired
    // 消息接收处理类
    private MyAckReceiver myAckReceiver;
    
    @Bean
    public SimpleMessageListenerContainer simpleMessageListenerContainer() {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        // 设置当前消费者数量
        container.setConcurrentConsumers(1);
        // 设置最大并发消费者数量
        container.setMaxConcurrentConsumers(1);
        // RabbitMQ默认是自动确认，这里改为手动确认消息
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 设置 监听的队列
        container.setQueueNames("ack_direct_queue");
        // 如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        // container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
        
        // 另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        // container.setQueues(new Queue("TestDirectQueue",true));
        // container.addQueues(new Queue("TestDirectQueue2",true));
        // container.addQueues(new Queue("TestDirectQueue3",true));
        
        // 设置监听处理类
        container.setMessageListener(myAckReceiver);
        
        return container;
    }
    
}

/**
 *
 * 监听处理类
 * @author <a href="920786312@qq.com">周飞</a>
 * @class: MyAckReceiver
 * @date 2021/9/7  19:51
 * @Verson 1.0 -2021/9/7  19:51
 * @see
 */

import java.util.HashMap;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

@Component

public class MyAckReceiver implements ChannelAwareMessageListener {
    Logger logger = LoggerFactory.getLogger(getClass());
    
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 获取消息
        String msg = message.toString();
        // 消费者队列
        String consumerQueue = message.getMessageProperties().getConsumerQueue();
        try {
            
            logger.info("消息队列：" + consumerQueue + "，消费的消息：" + msg);
            
            // 手动ACK
            // 第2个参数，手动确认是否可以被批处理
            // 当该参数为 true，小于 当前这条消息的delivery_tag 的所有消息，都确认
            // 当该参数为 false，只确认当前这条消息
            channel.basicAck(deliveryTag, true);
            
        } catch (Exception e) {
            logger.error("消息消费失败，消息队列：" + consumerQueue + "，消费的消息：" + msg);
            // 第2个参数：是否重新入列
            // true：消息重新丢回队列里，下次还会消费这消息
            // false:消息不丢回队列里，服务器丢失这条消息
            
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }
    
}

3.4、测试

http://127.0.0.1:8080/ack/sendMsg4