0
点赞
收藏
分享

微信扫一扫

如何玩懂RabbitMQ的消息确认机制？

乐百川 2022-04-14 阅读 86
rabbitmq消息确认回调函数

目录

问题引入

一、生产者的消息确认

1.1 准备工作

1.2 回调函数触发规则

1.3 小结

二、消费者的消息确认

2.1 自动确认

2.2 手动确认

2.2.1 确认的方法

2.2.2 basic.nack

2.2.3 basic.reject

2.3 手动确认的配置

2.3.1 监听单个队列

2.3.2 监听多个队列

小结

问题引入

如果不进行特殊配置的话，默认情况下发布操作是不会返回任何信息给生产者的，也就是默认情况下我们的生产者是不知道消息有没有正确到达broker的，如果在消息到达broker之前已经丢失的话，持久化操作也解决不了这个问题，因为消息根本就没到达代理服务器，我们是无法进行持久化的。

消费者收到消息后的处理也是如此，例如还没来得及"消费"它,或者说还没来得及进行业务逻辑处理时,消费者所在的信道或者连接因某种原因断开了，那这条消息岂不是就被无情的抛弃了...而这个问题的解决就需要我们的消息确认机制来进行处理了。

一、生产者的消息确认

1.1 准备工作

（1）创建一个SpringBoot项目

（2）导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>


（3）编写配置文件

application.yml:

server:
  port: 8021
spring:
#给项目起个名字
  application:
    name: rabbitmq-provider
  #配置rabbitMq 服务器
  rabbitmq:
    host: 服务器地址
    port: 5672
    username: yixin
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /
    #消息确认配置项

    #确认消息已发送到交换机(Exchange)
    publisher-confirm-type: correlated
    #确认消息已发送到队列(Queue)
    publisher-returns: true

（4）编写配置相关的消息确认回调函数

package com.yixin.config;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;



@Configuration
public class RabbitConfig {

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

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                System.out.println("ConfirmCallback:     "+"相关数据："+correlationData);
                System.out.println("ConfirmCallback:     "+"确认情况："+ack);
                System.out.println("ConfirmCallback:     "+"原因："+cause);
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("ReturnCallback:     "+"消息："+message);
                System.out.println("ReturnCallback:     "+"回应码："+replyCode);
                System.out.println("ReturnCallback:     "+"回应信息："+replyText);
                System.out.println("ReturnCallback:     "+"交换机："+exchange);
                System.out.println("ReturnCallback:     "+"路由键："+routingKey);
            }
        });

        return rabbitTemplate;
    }

}

此时我们的生产者推送消息的消息确认调用回调函数已经编写完毕。

1.2 回调函数触发规则


可以看到我们在配置类中写了两个回调函数，一个叫 ConfirmCallback ，一个叫 RetrunCallback；
那么以上这两种回调函数都是在什么情况会触发呢？

我们从总体的情况分析，推送消息存在四种情况：

接下来我们进行编写几个接口来分别测试和认证以上4种情况，看看消息确认触发回调函数的触发情况：

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

package com.yixin.controller;

import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@RestController
public class FannoutController {

    @Autowired
    RabbitTemplate rabbitTemplate;  //使用RabbitTemplate,这提供了接收/发送等等方法

    @GetMapping("/TestMessageAck")
    public String TestMessageAck() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: non-existent-exchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        
        //non-existent-exchange这个交换机是不存在的。
       rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,map);
        return "ok";
    }
}

浏览器输入：http://localhost:8021/TestMessageAck 进行调用接口，查看rabbitmq-provider项目的控制台输出情况:

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：false
ConfirmCallback:     原因：channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no exchange 'non-existent-exchange' in vhost '/', class-id=60, method-id=40)

 原因中表示没有找到交换机'non-existent-exchange'。

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

（1）我们需要新增一个交换机，但是不给这个交换机绑定队列，我来简单地在FannoutRabbitMQ里面新增一个Fanout交换机，名叫‘lonelyFanouttExchange’，但没给它做任何绑定配置操作：

package com.yixin.config;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class FannoutRabbitMQ {

    @Bean
    FanoutExchange lonelyFanoutExchange() {
        return new FanoutExchange("lonelyFanoutExchange");
    }
}

（2）编写写个测试接口，把消息推送到名为‘lonelyFanoutExchange’的交换机上（这个交换机是没有任何队列配置的）：

    @GetMapping("/TestMessageAck2")
    public String TestMessageAck2() {
        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);
        
       rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,map);
        return "ok";
    }

调用接口，查看rabbitmq-provider项目的控制台输出情况：

ReturnCallback:

ReturnCallback:     消息：(Body:'[serialized object]' MessageProperties [headers={}, contentType=application/x-java-serialized-object, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, deliveryTag=0])
ReturnCallback:     回应码：312
ReturnCallback:     回应信息：NO_ROUTE
ReturnCallback:     交换机：lonelyFanoutExchange
ReturnCallback:     路由键：TestFanoutRouting

ConfirmCallback:

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

分析：

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

这种情况一看就觉得跟①很像，没错 ，③和①情况回调是一致的，所以不做结果说明了。

④消息推送成功

（1）配置中绑定队列和交换机

package com.yixin.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class FannoutRabbitMQ {

    @Bean
    public Queue queueConfirm() {
        return new Queue("fanout.confirm",true);
    }

    @Bean
    FanoutExchange confirmFanouttExchange() {
        return new FanoutExchange("lonelyFanoutExchange");
    }
    @Bean
    Binding bindingExchangeConfirm() {
        return BindingBuilder.bind(queueConfirm()).to(confirmFanouttExchange());
    }

}

（2）编写接口进行调用

    @GetMapping("/TestMessageAck4")
    public String TestMessageAck4() {

        String messageId = String.valueOf(UUID.randomUUID());
        String messageData = "message: lonelyDirectExchange test message ";
        String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        Map<String, Object> map = new HashMap<>();
        map.put("messageId", messageId);
        map.put("messageData", messageData);
        map.put("createTime", createTime);

   rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,map);
        return "ok";
    }


调用接口进行测试，控制台如下：

ConfirmCallback:     相关数据：null
ConfirmCallback:     确认情况：true
ConfirmCallback:     原因：null

1.3 小结

二、消费者的消息确认

消费者的消息确认机制不同，因为消息接收本来就是在监听消息，符合条件的消息就会消费下来。
所以，消息接收的确认机制主要存在两种种模式：

2.1 自动确认

2.2 手动确认

2.2.1 确认的方法

注：

2.2.2 basic.nack

channel.basicNack(deliveryTag, false, true);


第一个参数：依然是当前消息到的数据的唯一id;
第二个参数：指是否针对多条消息；如果是true，也就是说一次性针对当前通道的消息的tagID小于当前这条消息的，都拒绝确认。
第三个参数：是指是否重新入列，也就是指不确认的消息是否重新丢回到队列里面去。

2.2.3 basic.reject

channel.basicReject(deliveryTag, true);  

拒绝消费当前消息，如果第二参数传入true，就是将数据重新丢回队列里，那么下次还会消费这消息。设置false，就是告诉服务器，我已经知道这条消息数据了，因为一些原因拒绝它，而且服务器也把这个消息丢掉就行, 下次不想再消费这条消息了。

注意：

2.3 手动确认的配置

2.3.1 监听单个队列

（1）创建一个SpringBoot的消费者项目

（2）导入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
 
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

（3）编写配置文件

server:
  port: 8022
spring:
  #给项目起个名字
  application:
    name: rabbitmq-consumer
  #配置rabbitMq 服务器
  rabbitmq:
    host: 服务器地址
    port: 5672
    username: yixin
    password: 123456
    #虚拟host 可以不设置,使用server默认host
    virtual-host: /

（4）编写配置类

package com.yixin.config;



import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@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);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        //设置一个队列
        container.setQueueNames("fanout.confirm");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
        //  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;
    }


}

（5）编写消息监听类

package com.yixin.config;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
           
            String msg = message.toString();


            System.out.println("消费者收到消息:"+msg);

            System.out.println("消费的主题消息来自："+message.getMessageProperties().getConsumerQueue());
            channel.basicAck(deliveryTag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }


}

（6）编写调用接口

    @GetMapping("/TestMessageAck5")
    public String TestMessageAck5() {


        String msg="我是一心同学";
        rabbitTemplate.convertAndSend("lonelyFanoutExchange", null,msg);
        return "发送成功 ";
    }

进行调用，投递消息，查看我们的消费者控制台：

2.3.2 监听多个队列

我们在生产者那边多绑定一个队列，其余不变：

package com.yixin.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@Configuration
public class FannoutRabbitMQ {

    @Bean
    public Queue queueConfirm() {
        return new Queue("fanout.confirm",true);
    }

    @Bean
    public Queue queueConfirmMore() {
        return new Queue("fanout.confirmMore",true);
    }

    @Bean
    FanoutExchange confirmFanouttExchange() {
        return new FanoutExchange("lonelyFanoutExchange");
    }
    @Bean
    Binding bindingExchangeConfirm() {
        return BindingBuilder.bind(queueConfirm()).to(confirmFanouttExchange());
    }

    @Bean
    Binding bindingExchangeConfirmMore() {
        return BindingBuilder.bind(queueConfirmMore()).to(confirmFanouttExchange());
    }

}

现在我们交换机已经绑定了fanout.confirm和fanout.confirmMore两个队列。（我们也可以多注册几个交换机，每个交换机绑定各自的队列，这里只是为了方便而已）

消费者配置

配置类：

package com.yixin.config;



import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;


@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);
        container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认，这里改为手动确认消息
        //设置一个队列
      //  container.setQueueNames("fanout.confirm");
        //如果同时设置多个如下： 前提是队列都是必须已经创建存在的
         container.setQueueNames("fanout.confirm","fanout.confirmMore");


        //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues
        //container.setQueues(new Queue("TestDirectQueue",true));
        //container.addQueues(new Queue("TestDirectQueue2",true));
        //container.addQueues(new Queue("TestDirectQueue3",true));
        container.setMessageListener(myAckReceiver);

        return container;
    }


}

消息监听类

手动确认消息监听类，MyAckReceiver.java 就可以同时将上面设置到的队列的消息都消费下来并且我们需要做不用的业务逻辑处理，那么只需要  根据消息来自的队列名进行区分处理即可：

package com.yixin.config;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

@Component
public class MyAckReceiver implements ChannelAwareMessageListener {

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
         
            String msg = message.toString();

            
            if ("fanout.confirm".equals(message.getMessageProperties().getConsumerQueue())) {
                System.out.println("消费的消息来自的队列名为：" + message.getMessageProperties().getConsumerQueue());
                System.out.println("消费者收到消息:"+msg);
                System.out.println("执行fanout.confirm中的消息的业务处理流程......");
            }
            if ("fanout.confirmMore".equals(message.getMessageProperties().getConsumerQueue())){
                System.out.println("消费的消息来自的队列名为："+message.getMessageProperties().getConsumerQueue());
                System.out.println("消费者收到消息:"+msg);
                System.out.println("执行fanout.confirmMore中的消息的业务处理流程......");

            }

            channel.basicAck(deliveryTag, true); //第二个参数，手动确认可以被批处理，当该参数为 true 时，则可以一次性确认 delivery_tag 小于等于传入值的所有消息
//			channel.basicReject(deliveryTag, true);//第二个参数，true会重新放回队列，所以需要自己根据业务逻辑判断什么时候使用拒绝
        } catch (Exception e) {
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }
    }


}

现在我们调用生产者那边的接口进行投递消息，查看消费者这边的控制台：

 可以看到我们的我们不同的队列都被监听了，并且执行了各自的逻辑。

小结

以上就是【一心同学】整理的【RabbitMQ】的【消息确认机制】，我们可以利用【消息确认机制】进而避免我们的消息丢失，从而【保护数据】。

举报

相关推荐

RabbitMQ 消息确认机制

RabbitMQ个人学习vanisha...微服务搭建

RabbitMQ消息确认机制

flink大数据状态清理ttlstate

RabbitMQ消息的发布确认机制详解

策略模式

RabbitMq 消息确认和退回机制

爬虫

重学“消息队列”，详解 RabbitMQ 消息确认机制！

我的日记本Java

RabbitMQ：RabbitMQ之消息确认机制（事务+Confirm）

技术杂谈saasjavaWeb学习java基础程序员Java 杂谈

07. RabbitMQ消息成功确认机制

RabbitMQMQ

RabbitMQ 发布确认机制

rabbitmq分布式

springboot + rabbitmq 用了消息确认机制，感觉掉坑里了

JAVA开发

springboot整合rabbitMQ---生产者消息确认机制

springspringboot消息确认机制rabbitMQ消息发送编程语言
0 条评论