前言为什么会有/confirm/iSpringboot 整合 Mq 实现 Confirm 监听机制
依赖引入增加配置文件，设定连接信息配置队列、交换机，以及对其进行绑定编写mq消息发送服务编写消息发送接口启动项目进行测试
正常测试异常测试 什么是Return？增加 ReturnCallback 监听并测试
修改 RabbitmqService 配置类测试 总结相关代码下载
前言之前专栏中，对Springboot整合Rabbitmq都有一系列的配置和说明，但总缺少一些必要的描述信息。导致很多看博客的小伙伴会私信问为什么需要这么配置的问题。
本篇博客重点进行Confirm 机制和Return 机制的实现和说明。
为什么会有ConfirmRabbitMq中，针对数据由消息生产者向消息队列推送时，通常情况如下所示(以 Routing 方式为例)：
每个Virtual Host 虚拟机中，都会含有各自的Exchange和Queue，需要在rabbitmq web界面中针对可以访问该Virtual Host 虚拟机的用户进行设定。
有点类似数据库的概念，指定用户只能操作指定的数据库。
在使用交换机 Exchange时，消息生产者需要将消息通过Channel 管道将数据发送给MQ，但想过一个问题没有：
Springboot 整合 Mq 实现 Confirm 监听机制 依赖引入如何 确定 消息是否真的发送到了指定的 MQ 中呢？
MQ中，对此问题，提出有Confirm 机制，对其发送数据进行监听，让消息发送者知道消息的发送结果。
开发测试主要的SpringBoot 版本为2.1.4.RELEASE。
此时只需要引入指定的amqp依赖即可：
org.springframework.boot spring-boot-starter-amqp
完整的pom依赖如下所示：
增加配置文件，设定连接信息4.0.0 org.example springboot-rabbitmq 1.0-SNAPSHOT org.springframework.boot spring-boot-starter-parent 2.1.4.RELEASE 1.8 UTF-8 UTF-8 org.springframework.boot spring-boot-starter org.springframework.boot spring-boot-starter-amqp org.springframework.boot spring-boot-starter-web org.springframework.boot spring-boot-starter-test test org.springframework.boot spring-boot-configuration-processor true org.projectlombok lombok 1.16.20 org.slf4j slf4j-api 1.7.26 org.slf4j slf4j-log4j12 1.7.26
增加配置文件，配置使用具体的Virtual Host、Username、Password、Host、Port等信息。
server:
port: 80
spring:
rabbitmq:
host: xxxxxx
port: 5672
username: xiangjiao
password: bunana
virtual-host: /xiangjiao
publisher-/confirm/is: 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;
@Configuration
public 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);
}
//绑定消息队列和交换机
@Bean
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
@Component
public class RabbitmqService implements RabbitTemplate./confirm/iCallback, RabbitTemplate.ReturnCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMessage(String exchange,String routingKey,Object msg) {
// 设置交换机处理失败消息的模式 true 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者
// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallback
rabbitTemplate.setMandatory(true);
//消息消费者确认收到消息后，手动ack回执
rabbitTemplate.set/confirm/iCallback(this);
// 暂时关闭 return 配置
//rabbitTemplate.setReturnCallback(this);
//发送消息
rabbitTemplate.convertAndSend(exchange,routingKey,msg);
}
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
}
@Override
public void /confirm/i(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;
@Controller
public 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] ---- INFO cn.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/i处理机制。
rabbitTemplate.set/confirm/iCallback(this);
@Override
public void /confirm/i(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/i机制，能够更好的保证消息生产者确认消息是否正常到达Exchange中。
但是，在MQ中，由于使用Exchange和Queue进行了绑定，
如果某个队列宕机了，Exchange并未将消息发送匹配 Routing Key 的队列，那么消息就不能到达队列中！！！
mq中，对此情况设有另外一种监听机制：Return机制！
增加 ReturnCallback 监听并测试 修改 RabbitmqService 配置类当消息由Exchange 未能传递到匹配的 queue 中，则会通过ReturnCallback根据用户的抉择，判断是否需要返回给消息生产者。
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
@Component
public class RabbitmqService implements RabbitTemplate./confirm/iCallback, RabbitTemplate.ReturnCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
public void sendMessage(String exchange,String routingKey,Object msg) {
// 设置交换机处理失败消息的模式 true 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者
// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallback
rabbitTemplate.setMandatory(true);
//消息消费者确认收到消息后，手动ack回执
rabbitTemplate.set/confirm/iCallback(this);
// return 配置
rabbitTemplate.setReturnCallback(this);
//发送消息
rabbitTemplate.convertAndSend(exchange,routingKey,msg);
}
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
}
@Override
public void /confirm/i(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 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者 // 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 setReturnCallback rabbitTemplate.setMandatory(true);测试
修改对应的测试类，保证交换机正确，但路由key不存在对应的队列即可。
// xiangjiao.routingKey 存在对应的queue // xiangjiao.routingKey_error 不存在对应的 queue rabbitmqService.sendMessage(MQConfiguration.EXCHANGE, "xiangjiao.routingKey_error", msg);
重启项目，访问接口，进行测试：
总结消息发送给Exchange成功，但是通过Exchange向Queue中推送数据时 失败，经过ReturnCallback 的 returnedMessage捕获监听！
通过配置/confirm/iCallback和ReturnCallback，便能实现消息生产者到交换机和消息由exchange到queue这个链路的安全性！
相关代码下载都是出现问题，或者正常后，给生产者方进行反馈。
gitee 代码下载地址
