Springboot整合Rabbitmq之Confirm和Return机制

2024-09-28 12:45:48

前言

之前专栏中，对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依赖即可：

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

完整的pom依赖如下所示：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.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">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>springboot-rabbitmq</artifactId>
    <version>1.0-SNAPSHOT</version>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.1.4.RELEASE</version>
        <relativePath /> <!-- lookup parent from 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>
        <!-- 引入rabbitmq依赖 -->
            <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: 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 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();
		/**
		 * 参数一：交换机名称；<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
@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：msg0
2022-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  失败原因
 */
@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根据用户的抉择，判断是否需要返回给消息生产者。

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

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