亲宝软件园·资讯

展开

Springboot整合Rabbitmq

专注写bug 人气:0

前言

之前专栏中,对Springboot整合Rabbitmq都有一系列的配置和说明,但总缺少一些必要的描述信息。导致很多看博客的小伙伴会私信问为什么需要这么配置的问题。

本篇博客重点进行Confirm 机制Return 机制的实现和说明。

为什么会有Confirm

RabbitMq中,针对数据由消息生产者消息队列推送时,通常情况如下所示(以 Routing 方式为例):

在这里插入图片描述

每个Virtual Host 虚拟机中,都会含有各自的ExchangeQueue,需要在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 HostUsernamePasswordHostPort等信息。

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中,由于使用ExchangeQueue进行了绑定,

如果某个队列宕机了,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);

重启项目,访问接口,进行测试:

在这里插入图片描述

消息发送给Exchange成功,但是通过ExchangeQueue中推送数据时 失败,经过ReturnCallback 的 returnedMessage捕获监听!

总结

通过配置ConfirmCallbackReturnCallback,便能实现消息生产者到交换机消息由exchange到queue这个链路的安全性!

在这里插入图片描述

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

相关代码下载

gitee 代码下载地址

加载全部内容

相关教程
猜你喜欢
用户评论