Springboot整合Rabbitmq之Confirm和Return机制

Springboot整合Rabbitmq之Confirm和Return机制

目录前言为什么会有ConfirmSpringboot整合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.boot

spring-boot-starter-amqp

完整的pom依赖如下所示：

xmlns:xsi="http://w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

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-starteRGkhaLikur

spring-boot-starter-amqp

spring-boot-starter-web

spring-boot-starter-test

test

spring-boot-configuration-processor

true

org.projectlombok

lombok

1.16.20

org.slf4j

slf4j-api

1.7.26

slf4j-log4j12

xmlns:xsi="http://w3.org/2001/XMLSchema-instance"

xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

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-starteRGkhaLikur

spring-boot-starter-amqp

spring-boot-starter-web

spring-boot-starter-test

test

spring-boot-configuration-processor

true

org.projectlombok

lombok

1.16.20

org.slf4j

slf4j-api

1.7.26

slf4j-log4j12

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

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

server:

port: 80

spring:

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;

@Configuration

public class MQConfiguration {

//队列名称

public static final String QUEUQ_NAME = "xiangjiao.queue";

//交换器名称

public static final String EXCHANGE = "xiangjiao.exchange";

//路由key

public static final http://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

@Component

public 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;

@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：msg02022-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 exchanRGkhaLikuge '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 失败原因

*/

@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);

}

}

什么是Return？

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

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

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

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

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

增加 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

@Component

public 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 表示消息由交换机 到达不了队列时，会将消息重新返回给生产者

// 如果不设置这个指令，则交换机向队列推送消息失败后，不会触发 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捕获监听！

总结

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

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

相关代码下载

gitee 代码下载地址

版权声明：本文内容由网络用户投稿，版权归原作者所有，本站不拥有其著作权，亦不承担相应法律责任。如果您发现本站中有涉嫌抄袭或描述失实的内容，请联系我们jiasou666@gmail.com 处理，核实后本网站将在24小时内删除侵权内容。