Springboot整合Rabbitmq之Confirm和Return机制 Springboot整合Rabbitmq之Confirm和

前言
为什么会有Confirm
Springboot整合Mq实现Confirm监听机制

依赖引入
增加配置文件，设定连接信息
配置队列、交换机，以及对其进行绑定
编写mq消息发送服务
编写消息发送接口
启动项目进行测试

正常测试
异常测试

什么是Return？

增加ReturnCallback监听并测试

修改RabbitmqService配置类

测试

总结

相关代码下载

前言之前专栏中，对Springboot整合Rabbitmq都有一系列的配置和说明，但总缺少一些必要的描述信息。导致很多看博客的小伙伴会私信问为什么需要这么配置的问题。
本篇博客重点进行Confirm 机制和Return 机制的实现和说明。

为什么会有Confirm RabbitMq中，针对数据由消息生产者向消息队列推送时，通常情况如下所示(以 Routing 方式为例)：

文章图片

每个Virtual Host 虚拟机中，都会含有各自的Exchange和Queue，需要在rabbitmq web界面中针对可以访问该Virtual Host 虚拟机的用户进行设定。

有点类似数据库的概念，指定用户只能操作指定的数据库。

在使用交换机 Exchange时，消息生产者需要将消息通过Channel 管道将数据发送给MQ，但想过一个问题没有：
如何确定消息是否真的发送到了指定的 MQ 中呢？

文章图片

MQ中，对此问题，提出有Confirm 机制，对其发送数据进行监听，让消息发送者知道消息的发送结果。

Springboot 整合 Mq 实现 Confirm 监听机制
依赖引入
开发测试主要的SpringBoot 版本为2.1.4.RELEASE。
此时只需要引入指定的amqp依赖即可：

org.springframework.bootspring-boot-starter-amqp

完整的pom依赖如下所示：

4.0.0org.examplespringboot-rabbitmq1.0-SNAPSHOTorg.springframework.bootspring-boot-starter-parent2.1.4.RELEASE 1.8UTF-8UTF-8org.springframework.bootspring-boot-starterspring-boot-starter-amqpspring-boot-starter-webspring-boot-starter-testtestspring-boot-configuration-processortrueorg.projectlomboklombok1.16.20org.slf4jslf4j-api1.7.26slf4j-log4j12

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

server:port: 80spring:rabbitmq:host: xxxxxxport: 5672username: xiangjiaopassword: bunanavirtual-host: /xiangjiaopublisher-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"; //路由key public static final String ROUTING_KEY = "xiangjiao.routingKey"; //创建队列 @Bean public 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(); /*** 参数一：交换机名称；
* 参数二：是否永久；
* 参数三：是否自动删除；
*/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 {@Autowiredprivate RabbitTemplate rabbitTemplate; public void sendMessage(String exchange,String routingKey,Object msg) {// 设置交换机处理失败消息的模式true 表示消息由交换机到达不了队列时，会将消息重新返回给生产者// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallbackrabbitTemplate.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 路由键*/@Overridepublic 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==truecause="+cause); }else{// 没有接收到log.info("---- confirm ----ack==falsecause="+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 { @Autowired private RabbitmqService rabbitmqService; @RequestMapping("/sendMoreMsgTx") @ResponseBody public 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：msg0
2022-02-28 10:34:58.686 ---- [rabbitConnectionFactory1] ---- INFOcn.linkpower.service.RabbitmqService - ---- confirm ----ack=false
cause=channel error; protocol method: #method(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==truecause="+cause); }else{// 没有接收到log.info("---- confirm ----ack==falsecause="+cause); }}

什么是Return？上面的配置中，采取Confirm机制，能够更好的保证消息生产者确认消息是否正常到达Exchange中。
但是，在MQ中，由于使用Exchange和Queue进行了绑定，

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

文章图片

【Springboot整合Rabbitmq之Confirm和Return机制】
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 {@Autowiredprivate RabbitTemplate rabbitTemplate; public void sendMessage(String exchange,String routingKey,Object msg) {// 设置交换机处理失败消息的模式true 表示消息由交换机到达不了队列时，会将消息重新返回给生产者// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallbackrabbitTemplate.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 路由键*/@Overridepublic 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失败原因*/@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==truecause="+cause); }else{// 没有接收到log.info("---- confirm ----ack==falsecause="+cause); }}}

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

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

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

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

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