rabbitmq的消息发布确认机制_rabbitmq手动确认机制

作者：你好赵伟 | 2024-04-08 13:28:59

踩

rabbitmq手动确认机制

1. 消息可靠性投递
前言
在代码里面一定是先操作数据库再发送消息。避免因为数据库回滚导致的数据不一致。但是如果先操作数据，后发送消息，发送消息出了问题，那不是一样会出现业务数据的不一致？

这篇文章我们来分析 RabbitMQ 的可靠性投递，也就是在使用 RabbitMQ 实现异步通信的时候，消息丢了怎么办，消息重复消费怎么办？在 RabbitMQ 里面提供了很多保证消息可靠投递的机制，这个也是 RabbitMQ 的一个特性。

在学习RabbitMQ前，必须要明确一个问题，因为效率与可靠性是无法兼得的，如果要保证每一个环节都成功，势必会对消息的收发效率造成影响。所以如果是一些业务实时一致性要求不是特别高的场合，可以牺牲一些可靠性来换取效率。比如发送通知或者记录日志的这种场景，如果用户没有收到通知，不会造成业务影响，只要再次发送就可以了。

在我们使用 RabbitMQ 收发消息的时候，有几个主要环节：

在这里插入图片描述
1 代表消息从生产者发送到 Broker。
生产者把消息发到 Broker 之后，怎么知道自己的消息有没有被 Broker 成功接收？
2 代表消息从 Exchange 路由到 Queue
Exchange 是一个绑定列表，如果消息没有办法路由到正确的队列，会发生什么事情？应该怎么处理？
3 代表消息在 Queue 中存储
队列是一个独立运行的服务，有自己的数据库（Mnesia），它是真正用来存储消息的。如果还没有消费者来消费，那么消息要一直存储在队列里面。如果队列出了问题，消息肯定会丢失。怎么保证消息在队列稳定地存储呢？
4 代表消费者订阅 Queue 并消费消息
队列的特性是什么？FIFO。队列里面的消息是一条一条的投递的，也就是说，只有上一条消息被消费者接收以后，才能把这一条消息从数据库删掉，继续投递下一条消息。那么问题来了，Broker 怎么知道消费者已经接收了消息呢？

下面我们通过收到ack的方式保证消息的可靠性

1.我们进行配置开启手动确认机制


server.port=9994
 
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
 
# 发送方
# 开启发送确认(未到达MQ服务器)
spring.rabbitmq.publisher-confirms=true
# 开启发送失败退回(未找到对应queue)
spring.rabbitmq.publisher-returns=true
 
#启动时自动启动容器
spring.rabbitmq.listener.auto-startup=true
 
 
# 消费方 开启手动ACK
spring.rabbitmq.listener.direct.acknowledge-mode=manual
spring.rabbitmq.listener.simple.acknowledge-mode=manual

2.创建交换机,队列和路由


@Slf4j
@Configuration
public class Config1 implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;
    
    //后置处理器,先将其他注解都加载完在加载此方法,不然会造成空指针
    @PostConstruct
    public void init(){
        rabbitTemplate.setConfirmCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }
 
    //交换机是否收到消息的确认机制
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (ack){
            log.info("交换机收到生产者发送的消息");
        }else {
            log.error("交换机没有收到生产者发送的消息");
        }
    }
 
    //队列未收到消息会进行消息回调
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
       String mes=new String(message.getBody());
        log.error("队列未收到消息:{}",mes);
    }
 
 
    //交换机
    public static final String CONFIRM_EXCHANGE="confirm.exchange";
    //队列
    public static final String CONFIRM_QUEUE="confirm.queue";
 
    public static final String CONFIRM_ROUTING_KEY="confirm.routing";
 
    @Bean
    public DirectExchange confirmExchange(){
        return new DirectExchange(CONFIRM_EXCHANGE,true,false);
    }
 
    @Bean
    public Queue confirmQueue(){
        return QueueBuilder.durable(CONFIRM_QUEUE).build();
    }
 
    @Bean
    public Binding confirmBinding(){
        return BindingBuilder.bind(confirmQueue()).to(confirmExchange()).with(CONFIRM_ROUTING_KEY);
    }
 
 
 
}

3.生产者发送消息


@Slf4j
@RestController
@RequestMapping("/confirm")
public class ConfirmController {
 
    @Autowired
    private RabbitTemplate rabbitTemplate;
 
 
    @RequestMapping(value = "/send/{message}",method = RequestMethod.GET)
    public void sendMes(@PathVariable("message") String message){
        rabbitTemplate.convertAndSend(Config1.CONFIRM_EXCHANGE,Config1.CONFIRM_ROUTING_KEY,message);
        log.info("生产者发送的消息为:{}",message);
    }
}

4.消费者监听消息(此时还未手动ack)


@Slf4j
@Component
public class ConfirmConsumer {
 
    @RabbitListener(queues = Config1.CONFIRM_QUEUE)
    public void getMes(Message message, Channel channel){
        String mes=new String(message.getBody());
        log.info("消费者收到的消息为:{}",mes);
    }
 
}