springboot整合rabbitmq消息确认,rabbitmq confirm机制

　　00-1010序：为什么会有ConfirmSpringboot集成Mq实现确认监听机制？它依赖于添加配置文件、设置连接信息、配置队列和交换机以及绑定它们。写mq消息服务，写消息接口，启动项目，测试正常测试，异常测试。什么是回报？添加ReturnCallback进行监听测试，修改RabbitmqService配置类，测试并总结相关代码下载。

　　在00-1010之前的专栏里，有Springboot集成Rabbitmq的一系列配置和描述，但是一直缺少一些必要的描述信息。这样一来，很多看博客的朋友都会私底下问为什么要这个配置。

　　本博客重点介绍确认机制和返回机制的实现和说明。

前言

　　用于开发和测试的SpringBoot的主要版本是2.1.4.RELEASE

　　此时，您只需要引入指定的amqp依赖关系：

　　依赖关系groupIdorg.springframework.boot/groupId工件Spring-Boot-Starter-AMQP/工件ID/依赖关系完整的pom依赖关系如下：

　　骑乘作物xml版本=1.0 编码=UTF-8 ？项目xmlns= http://maven . Apache . org/POM/4 . 0 . 0 xmlns : xsi= http://www . w3 . org/2001/XML schema-instance xsi : schema location= http://maven . Apache . org/POM/4 . 0 . 0 http://maven.apache.org/xsd/maven-4.0.0.xsd模型版本4 . 0 . 0/模型版本groupIdorg.example/groupId artifactIdspringboot-rabbit MQ/artifactId版本1.0-快照-从中查找父项

repository --> </parent> <properties> <java.version>1.8</java.version> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter</artifactId> </dependency>  <artifactId>spring-boot-starter-amqp</artifactId> <artifactId>spring-boot-starter-web</artifactId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> <artifactId>spring-boot-configuration-processor</artifactId> <optional>true</optional> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.16.20</version> <groupId>org.slf4j</groupId> <artifactId>slf4j-api</artifactId> <version>1.7.26</version> <artifactId>slf4j-log4j12</artifactId> </dependencies></project>

增加配置文件，设定连接信息

增加配置文件，配置使用具体的Virtual Host、Username、Password、Host、Port等信息。

server: port: 80spring: rabbitmq: host: xxxxxx port: 5672 username: xiangjiao password: bunana virtual-host: /xiangjiao publisher-confirms: true #消息发送到转发器确认机制，是都确认回调 publisher-returns: true

配置队列、交换机，以及对其进行绑定

指定交换机名称为：xiangjiao.exchange。队列名称为：xiangjiao.queue。使用Direct 直连模式，其中关联的Routingkey为：xiangjiao.routingKey。

package cn.linkpower.config;import org.springframework.amqp.core.*;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class MQConfiguration {//队列名称public static final String QUEUQ_NAME = "xiangjiao.queue";//交换器名称public static final String EXCHANGE = "xiangjiao.exchange";//路由keypublic static final String ROUTING_KEY = "xiangjiao.routingKey";//创建队列@Beanpublic Queue getQueue(){ // 另一种方式//QueueBuilder.durable(QUEUQ_NAME).build();return new Queue(QUEUQ_NAME);}//实例化交换机@Bean public DirectExchange getDirectExchange(){//DirectExchange(String name, boolean durable, boolean autoDelete)// 另一种方式：//ExchangeBuilder.directExchange(EXCHANGE).durable(true).build();/** * 参数一：交换机名称；<br> * 参数二：是否永久；<br> * 参数三：是否自动删除；<br> */return new DirectExchange(EXCHANGE, true, false);//绑定消息队列和交换机public Binding bindExchangeAndQueue(DirectExchange exchange,Queue queue){// 将 创建的 queue 和 exchange 进行绑定return BindingBuilder.bind(queue).to(exchange).with(ROUTING_KEY);}

编写mq消息发送服务

在Springboot中，针对MQ消息的发送，采取RabbitTemplate模板进行数据的发送处理操作。

　　手动定义消息发送处理类，对其RabbitTemplate进行其他设置。

package cn.linkpower.service;import lombok.extern.slf4j.Slf4j;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Slf4j@Componentpublic class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback { @Autowired private RabbitTemplate rabbitTemplate; public void sendMessage(String exchange,String routingKey,Object msg) { // 设置交换机处理失败消息的模式 true 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者 // 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallback rabbitTemplate.setMandatory(true); //消息消费者确认收到消息后，手动ack回执 rabbitTemplate.setConfirmCallback(this); // 暂时关闭 return 配置 //rabbitTemplate.setReturnCallback(this); //发送消息 rabbitTemplate.convertAndSend(exchange,routingKey,msg); } /** * 交换机并未将数据丢入指定的队列中时，触发 * channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes()); * 参数三：true 表示如果消息无法正常投递，则return给生产者 ；false 表示直接丢弃 * @param message 消息对象 * @param replyCode 错误码 * @param replyText 错误信息 * @param exchange 交换机 * @param routingKey 路由键 */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" "); * 消息生产者发送消息至交换机时触发，用于判断交换机是否成功收到消息 * @param correlationData 相关配置信息 * @param ack exchange 交换机，判断交换机是否成功收到消息 true 表示交换机收到 * @param cause 失败原因 public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info("---- confirm ----ack="+ack+" cause="+String.valueOf(cause)); log.info("correlationData -->"+correlationData.toString()); if(ack){ // 交换机接收到 log.info("---- confirm ----ack==true cause="+cause); }else{ // 没有接收到 log.info("---- confirm ----ack==false cause="+cause); }}

编写消息发送接口

编写一个Controller，将产生的数据，通过自定义的RabbitmqService发送至指定的Exchange交换机中。

package cn.linkpower.controller;import cn.linkpower.config.MQConfiguration;import cn.linkpower.service.RabbitmqService;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Controller;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.ResponseBody;@Controllerpublic class SendMessageTx {@Autowiredprivate RabbitmqService rabbitmqService;@RequestMapping("/sendMoreMsgTx")@ResponseBodypublic String sendMoreMsgTx(){//发送10条消息for (int i = 0; i < 10; i++) {String msg = "msg"+i;System.out.println("发送消息 msg："+msg);// xiangjiao.exchange 交换机// xiangjiao.routingKey 队列rabbitmqService.sendMessage(MQConfiguration.EXCHANGE, MQConfiguration.ROUTING_KEY, msg);//每两秒发送一次try {Thread.sleep(2000);} catch (InterruptedException e) {e.printStackTrace();}}return "send ok";}}

启动项目进行测试

正常测试

http://localhost/sendMoreMsgTx
　　

从控制台中可以看到消息信息如下所示：

　　发现，消息信息发送，都是ACK 被确认的！

异常测试

异常测试，首先需要保证mq服务中没有对应的exchange交换机。还需要保证消息的发送者exchange信息修改。

　　将controller中对应的消息发送的方式修改如下：

rabbitmqService.sendMessage("xiangjiao.exchangeError", MQConfiguration.ROUTING_KEY, msg);

　　重启项目，重新请求该接口，观察控制台数据信息展示：

　　截取其中的一条信息为例：

发送消息 msg：msg02022-02-28 10:34:58.686 ---- [rabbitConnectionFactory1] ---- INFO cn.linkpower.service.RabbitmqService - ---- confirm ----ack=false cause=channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND -no exchange 'xiangjiao.exchangeError' in vhost '/xiangjiao', class-id=60, method-id=40)
　　

当生产者向Exchange中发送消息，如果消息并未成功发送，则会触发RabbitmqService中设定的confirm处理机制。

rabbitTemplate.setConfirmCallback(this);/** * 消息生产者发送消息至交换机时触发，用于判断交换机是否成功收到消息 * @param correlationData 相关配置信息 * @param ack exchange 交换机，判断交换机是否成功收到消息 true 表示交换机收到 * @param cause 失败原因 */@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info("---- confirm ----ack="+ack+" cause="+String.valueOf(cause)); log.info("correlationData -->"+correlationData.toString()); if(ack){ // 交换机接收到 log.info("---- confirm ----ack==true cause="+cause); }else{ // 没有接收到 log.info("---- confirm ----ack==false cause="+cause); }}

什么是Return？

上面的配置中，采取Confirm机制，能够更好的保证消息生产者确认消息是否正常到达Exchange中。

　　但是，在MQ中，由于使用Exchange和Queue进行了绑定，

如果某个队列宕机了，Exchange并未将消息发送匹配 Routing Key 的队列，那么消息就不能到达队列中！！！
　　

　　mq中，对此情况设有另外一种监听机制：Return机制！

当消息由Exchange 未能传递到匹配的 queue 中，则会通过ReturnCallback根据用户的抉择，判断是否需要返回给消息生产者。
　　

增加 ReturnCallback 监听并测试

修改 RabbitmqService 配置类

package cn.linkpower.service;import lombok.extern.slf4j.Slf4j;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.connection.CorrelationData;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.stereotype.Component;@Slf4j@Componentpublic class RabbitmqService implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback { @Autowired private RabbitTemplate rabbitTemplate; public void sendMessage(String exchange,String routingKey,Object msg) { // 设置交换机处理失败消息的模式 true 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者 // 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallback rabbitTemplate.setMandatory(true); //消息消费者确认收到消息后，手动ack回执 rabbitTemplate.setConfirmCallback(this); // return 配置 rabbitTemplate.setReturnCallback(this); //发送消息 rabbitTemplate.convertAndSend(exchange,routingKey,msg); } /** * 交换机并未将数据丢入指定的队列中时，触发 * channel.basicPublish(exchange_name,next.getKey(), true, properties,next.getValue().getBytes()); * 参数三：true 表示如果消息无法正常投递，则return给生产者 ；false 表示直接丢弃 * @param message 消息对象 * @param replyCode 错误码 * @param replyText 错误信息 * @param exchange 交换机 * @param routingKey 路由键 */ @Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" "); } /** * 消息生产者发送消息至交换机时触发，用于判断交换机是否成功收到消息 * @param correlationData 相关配置信息 * @param ack exchange 交换机，判断交换机是否成功收到消息 true 表示交换机收到 * @param cause 失败原因 */ @Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { log.info("---- confirm ----ack="+ack+" cause="+String.valueOf(cause)); log.info("correlationData -->"+correlationData.toString()); if(ack){ // 交换机接收到 log.info("---- confirm ----ack==true cause="+cause); }else{ // 没有接收到 log.info("---- confirm ----ack==false cause="+cause); } }}

【注意：】设置 setReturnCallback 后，如果需要保证消息未传递到指定的 queue，需要将消息返回生产者时，一定要增加下面配置：

// 设置交换机处理失败消息的模式 true 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallbackrabbitTemplate.setMandatory(true);

测试

修改对应的测试类，保证交换机正确，但路由key不存在对应的队列即可。

// xiangjiao.routingKey 存在对应的queue// xiangjiao.routingKey_error 不存在对应的 queuerabbitmqService.sendMessage(MQConfiguration.EXCHANGE, "xiangjiao.routingKey_error", msg);

重启项目，访问接口，进行测试：

消息发送给Exchange成功，但是通过Exchange向Queue中推送数据时失败，经过ReturnCallback 的 returnedMessage捕获监听！
　　

总结

通过配置ConfirmCallback和ReturnCallback，便能实现消息生产者到交换机和消息由exchange到queue这个链路的安全性！

都是出现问题，或者正常后，给生产者方进行反馈。
　　