谷粒商城实战笔记-259-商城业务-消息队列-可靠投递-发送端确认

文章目录

• 一，确认机制简介
• 二，ConfirmCallback
• 三，returnCallback
• 事务消息的问题

一，确认机制简介

RabbitMQ的消息确认机制主要包括以下几种：

1. 发布者确认（Publisher Confirm）：在发布者和代理之间建立一个确认协议。当发布者发送一条消息到代理时，代理会返回一个确认信息给发布者。如果确认信息是positive ack（即确认收到），那么表示消息已经被代理成功接收；如果是negative ack（即拒绝或丢失），则表示消息没有被正确处理。

2. 返回回调（Return Callback）：当发布者发送的消息无法路由到任何队列时，代理会将该消息返回给发布者。

3. 消费者确认（Consumer Acknowledgments）：消费者在处理完消息后向代理发送一个确认信息，告诉代理消息已被正确处理。如果没有收到确认信息，代理会重新投递该消息。

二，ConfirmCallback

生产者的确认（Publisher Confirms）是一种机制，用于确保消息已经成功传递到RabbitMQ broker。当启用此功能时，生产者在发送消息后会收到一个确认，以表明消息已成功到达broker。

以下是启用和使用生产者确认的一些关键点：

1. 配置Spring RabbitMQ以启用确认：
   在application.properties文件中设置spring.rabbitmq.publisher-confirms=true。这将在创建ConnectionFactory时设置PublisherConfirms(true)选项，从而开启confirm callback。

2. CorrelationData：
   CorrelationData是一个用来表示当前消息唯一性的对象。每当生产者发送一条消息时，它应该提供一个唯一的CorrelationData实例，这样在确认回调中就可以识别出哪条消息得到了确认。

3. confirmCallback：
   当消息被broker接收到时，将会调用confirmCallback方法。如果ack参数为true，这意味着消息已被成功接受并处理；如果ack参数为false，这意味着消息未被成功处理，可能是由于某种原因被丢弃。

4. 使用示例：

   
   public void initRabbitTemplate() {
   
        /**
         * 1、只要消息抵达Broker就ack=true
         * correlationData：当前消息的唯一关联数据(这个是消息的唯一id)
         * ack：消息是否成功收到
         * cause：失败的原因
         */
        //设置确认回调
        rabbitTemplate.setConfirmCallback((correlationData,ack,cause) -> {
            System.out.println("confirm...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]");
        });
   
   

通过这种方式，生产者可以在发送消息后立即知道消息是否成功到达RabbitMQ broker，从而实现可靠的通信。

三，returnCallback

returnCallback是在RabbitMQ中使用的一种机制，用于处理无法路由到任何队列的消息。当生产者发送的消息不能被正确路由到任何队列时，RabbitMQ会将该消息返回给生产者。这时，生产者可以通过注册一个returnCallback来监听这些返回的消息。

以下是关于returnCallback的一些要点：

1. 配置Spring RabbitMQ以启用返回回调：
   在application.properties文件中设置spring.rabbitmq.publisher-returns=true。这将在创建ConnectionFactory时设置PublisherReturns(true)选项，从而开启return callback。

2. returnCallback：
   当消息无法路由到任何队列时，RabbitMQ会触发returnCallback。在回调中，你可以获取到消息的内容、原始交换器、路由键和返回代码/描述。

3. 使用示例：

   public void initRabbitTemplate() {
        /**
         * 只要消息没有投递给指定的队列，就触发这个失败回调
         * message：投递失败的消息详细信息
         * replyCode：回复的状态码
         * replyText：回复的文本内容
         * exchange：当时这个消息发给哪个交换机
         * routingKey：当时这个消息用哪个路邮键
         */
        rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> {
            System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]" +
                    "==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]");
        });
    }
   

通过returnCallback，生产者可以及时得知哪些消息无法被正确路由，并采取相应的措施，例如重试发送、记录日志或通知管理员等。这样可以提高系统的健壮性，并减少因消息无法送达而导致的问题。

事务消息的问题

事务消息是指在RabbitMQ中，生产者可以开启一个事务，然后在这个事务中发送多条消息。只有当所有的消息都成功到达队列之后，这个事务才会被提交。如果其中有任何一条消息未能成功到达队列，则整个事务会被回滚，所有的消息都不会被放入队列。

这种机制确保了消息的可靠性，但是同时也带来了性能上的问题。

首先，事务机制引入了大量的同步操作。在开启事务之后，生产者发送的所有消息都需要等待服务器的确认才能继续执行。这就意味着，生产者的发送速度受到了服务器响应速度的限制，而服务器的响应速度又受到当前系统负载的影响。因此，当系统负载较高或者网络延迟较大时，生产者的发送速度就会显著降低。

其次，事务机制增加了数据存储的压力。在事务期间，服务器需要为每个事务维护一个状态，以便在事务结束时决定是否要提交还是回滚。这对于内存和磁盘空间都是一个消耗。

最后，事务机制还增加了网络传输的开销。由于每次发送消息都需要等待服务器的确认，这就导致了大量的网络交互，增加了网络带宽的消耗。

综上所述，虽然事务消息提供了很高的可靠性和一致性，但是在高并发和大数据量的情况下，它的性能表现可能会比较差。因此，在实际应用中，我们需要根据具体的需求来权衡可靠性和性能之间的关系，选择合适的解决方案。