当前位置： 首页 > article >正文

RabbitMQ 在 Spring Boot中使用方式

article 2025/2/11 12:03:20

文章目录

    • 作用
    • MQ docker 安装
    • MQ使用
      • RabbitMQ的整体架构及核心概念：
      • RabbitMQ的整体架构及核心概念：
      • 消费者消息推送限制
      • 交换机与队列
      • ## 项目使用MQ
      • Direct: 直连模式
      • Fanout: 广播模式
      • Topic: 主题模式
      • Headers: 头信息模式
    • 使用DEMO地址
    • 异常问题记录

在这里插入图片描述

作用

RabbitMQ 作为一款开源消息队列中间件（基于 AMQP 协议），在项目中主要解决系统间的异步通信、解耦、流量削峰等问题，提升系统的可扩展性和可靠性。以下是其核心作用及典型场景：

  1. 异步处理

    • 场景：耗时操作（如发送邮件、短信、文件处理）异步执行，避免阻塞主流程。
    • 示例：用户注册后，主线程快速返回，通过 RabbitMQ 异步触发邮件发送、数据清洗等任务。
    • 优势：提升响应速度，优化用户体验，提高系统吞吐量。

  2. 应用解耦

    • 场景：系统间通过消息通信，降低直接依赖。
    • 示例：订单系统生成订单后，发送消息到队列，库存系统、物流系统各自订阅消息处理，任一系统故障不影响主流程。
    • 优势：增强系统容错性，模块独立升级维护更灵活。

  3. 流量削峰

    • 场景：应对突发高并发，避免服务过载。
    • 示例：秒杀活动中，请求先写入消息队列，后端服务按处理能力消费，避免数据库被击穿。
    • 优势：平滑流量波动，保护后端资源，结合限流策略提升系统稳定性。

  4. 日志收集与数据处理

    • 场景：分布式系统中收集多节点日志或数据。
    • 示例：多个服务将日志发送到 RabbitMQ，由统一消费者写入 Elasticsearch 或 Hadoop 分析。
    • 优势：集中处理数据，降低对业务系统性能影响。

  5. 分布式系统协调

    • 场景：实现跨服务事务最终一致性。
    • 示例：电商下单后，通过消息队列通知积分系统增加积分，若失败则重试或补偿。
    • 优势：替代同步 RPC 调用，降低分布式事务复杂度。

  6. 跨语言协作

    • 场景：异构系统（不同语言/框架）间通信。
    • 示例：Java 服务与 Python 数据分析服务通过 RabbitMQ 交换数据。
    • 优势：协议标准化，支持多种客户端（Java、Python、Go 等）。

  7. 关键技术机制

    • 可靠性：通过生产者确认（Publisher Confirm）、持久化（Persistent Messages）、消费者手动确认（ACK）确保消息不丢失。
    • 灵活性：支持多种交换机类型（Direct、Topic、Fanout、Headers）实现消息路由。
    • 扩展性：集群与镜像队列提供高可用，横向扩展消费者提升处理能力。
    • 容错：死信队列（DLX）处理失败消息，避免消息无限重试。

注意事项

  • 消息顺序：默认不保证严格顺序，需通过单队列单消费者或业务逻辑处理。
  • 重复消费：网络重试可能导致消息重复，需业务层幂等设计。
  • 监控：使用管理界面或 Prometheus + Grafana 监控队列积压、消费者状态。

典型应用案例

  • 电商系统：订单状态更新、库存扣减、通知推送。
  • 微服务架构：服务间事件驱动通信（如用户注销触发多系统清理）。
  • IoT 数据流：设备上报数据缓冲与分发处理。

例如:
支付服务 ，将核心业务剥离 ，以下 交易服务、通知服务、 积分服务可以通过mq 去处理，无需实时服务调用
在这里插入图片描述

MQ docker 安装

docker pull rabbitmq:latest

docker run \
 -e RABBITMQ_DEFAULT_USER=root \
 -e RABBITMQ_DEFAULT_PASS=123456 \
 -v F:/docker/data/rabbitmq/plugins:/plugins \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:latest
# 运行mq
docker run -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=123456 --name mq --hostname mq1 -p 15672:15672 -p 5672:5672 -d rabbitmq:3.8-management
# 进入容器
docker exec -it rabbitmq bash
# 开启 外部web页面
rabbitmq-plugins enable rabbitmq_management

MQ使用

RabbitMQ的整体架构及核心概念：

  • virtual-host：虚拟主机，起到数据隔离的作用
  • publisher：消息发送者
  • consumer：消息的消费者
  • queue：队列，存储消息
  • exchange：交换机，负责路由消息

RabbitMQ的整体架构及核心概念：

virtual-host：虚拟主机，起到数据隔离的作用
publisher：消息发送者
consumer：消息的消费者
queue：队列，存储消息
exchange：交换机，负责路由消息

消费者消息推送限制

默认情况下，RabbitMQ的会将消息依次轮询投递给绑定在队列上的每一个消费者。但这并没有考虑到消费者是否已经处理完消息，可能出现消息堆积。
因此我们需要修改application.yml，设置preFetch值为1，确保同一时刻最多投递给消费者1条消息：

spring:
	rabbitmq:
		listener:
			simple:
				prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

交换机与队列

交换机（Exchange）：
交换机是用来接收生产者发送的消息并将这些消息路由到一个或多个队列中的实体。有不同类型

  • Direct: 直连模式，根据路由键（Routing Key）精确匹配队列。
  • Fanout: 广播模式，将消息广播给所有绑定到该交换机的队列，忽略路由键。
  • Topic: 主题模式，使用通配符对路由键进行模式匹配来决定消息被发送到哪些队列。
  • Headers: 头信息模式，基于消息头属性而非路由键来路由消息。

队列：
队列用于存储消息直到消费者可以处理它们。队列是实际存放消息的地方，消费者从这里获取消息。

绑定（Binding）：
绑定是指将队列连接到交换机的过程。绑定时可以指定一个绑定键（Binding Key），这个键在不同的交换机类型中有不同的意义。

  • 作用：它定义了交换机与队列之间的关系，即通过何种规则将消息从交换机路由到特定的队列。例如，在direct类型的交换机中，绑定键必须与消息的路由键完全匹配才能使消息进入对应的队列；而在topic类型的交换机中，则支持模式匹配。

## 项目使用MQ

maven 引入

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

yaml 配置

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    #虚拟主机
    virtual-host: /hmall
    username: root
    password: 123456

Direct: 直连模式

Direct: 直连模式
配置类定义队列、Direct类型的交换机以及它们之间的绑定关系。

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitConfig {

    // 定义队列
    @Bean
    public Queue myQueue() {
        return new Queue("myQueue", true); // true表示持久化队列
    }

    // 定义Direct交换机
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("directExchange");
    }

    // 绑定队列到交换机上，并指定绑定键
    @Bean
    public Binding binding(Queue myQueue, DirectExchange directExchange) {
        return BindingBuilder.bind(myQueue).to(directExchange).with("routingKey");
    }
}

接收消息

@RabbitListener(queues = "myQueue")
public void receiveDirectMessage(String message) {
	System.out.println(" [x] Received '" + message + "'");
}

创建一个服务类用于向Direct交换机发送消息。

@Test
public void sendDirect(){
	String message = "Direct消息";
	rabbitTemplate.convertAndSend("test.direct", "test1", message);
	System.out.println(" [x] Sent '" + message + "'");
}

Fanout: 广播模式

广播模式通常指的是使用fanout类型的交换机来实现消息的广播。fanout交换机会将接收到的消息广播给所有绑定到该交换机的队列，而不考虑路由键（routing key）
配置Fanout Exchange和Queue
创建一个配置类来定义fanout类型的交换机以及需要绑定到此交换机的队列。

package com.itheima.consumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class RabbitFanoutConfig {

    // 定义第一个队列
    @Bean
    public Queue queue1() {
        return new Queue("queue1", true); // 'true' for durable queue
    }

    // 定义第二个队列
    @Bean
    public Queue queue2() {
        return new Queue("queue2", true);
    }

    // 定义fanout类型的交换机
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("myFanoutExchange");
    }

    // 将queue1绑定到fanout交换机上
    @Bean
    public Binding binding1(Queue queue1, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue1).to(fanoutExchange);
    }

    // 将queue2绑定到fanout交换机上
    @Bean
    public Binding binding2(Queue queue2, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(queue2).to(fanoutExchange);
    }
}

发送消息
发送消息到fanout类型的交换机时，不需要指定路由键，因为fanout交换机会忽略路由键并将消息广播给所有绑定的队列。

 @Test
public void sendFanout(){
	String message = "Fanout消息";
	rabbitTemplate.convertAndSend("myFanoutExchange", "", message);
	System.out.println(" [x] Sent '" + message + "'");
}

接收消息
为每个队列设置监听器来接收消息。

@RabbitListener(queues = "queue1")
public void receiveMessageFromQueue1(String message) {
	System.out.println("Queue1 [x] Received '" + message + "'");
}

@RabbitListener(queues = "queue2") 
public void receiveMessageFromQueue2(String message) {
	System.out.println("Queue2 [x] Received '" + message + "'");
}

所有绑定到myFanoutExchange交换机的队列都将接收到相同的消息。

Topic: 主题模式

主题模式（Topic Exchange）是一种灵活的消息路由机制。与直接模式（Direct Exchange）不同的是，主题交换机允许使用通配符来匹配消息的路由键（routing key），从而实现更加复杂的消息路由逻辑。下面是如何在Spring Boot中配置和使用主题模式的详细步骤。

在主题模式下，如果你想根据路由键进行个性化的监听，可以使用@RabbitListener的bindings属性，而不是直接指定队列名。这样可以根据路由键动态地监听消息。

  • 在@RabbitListener中，使用了bindings属性来定义如何绑定队列到交换机上。
  • value指定了要使用的队列，如果该队列不存在，则会自动创建。
  • exchange指定了目标交换机及其类型（在这里是topic类型的交换机）。
  • key定义了用于匹配消息路由键的绑定键。在这个例子中，item.*表示任何以item.开头的消息都会被发送到queue1；而#.update则匹配任何以.update结尾的消息并将其发送到queue2。
@RabbitListener(bindings = @QueueBinding(
		value = @Queue(value = "queue1", durable = "true"),
		exchange = @Exchange(value = "topicExchange", type = "topic"),
		key = "item.*"))
public void receiveMessageFromQueue11(String message) {
	System.out.println("Queue1 [x] Received '" + message + "'");
}

@RabbitListener(bindings = @QueueBinding(
		value = @Queue(value = "queue2", durable = "true"),
		exchange = @Exchange(value = "topicExchange", type = "topic"),
		key = "#.update"))
public void receiveMessageFromQueue21(String message) {
	System.out.println("Queue2 [x] Received '" + message + "'");
}

Topic Exchange发送消息。根据不同的路由键，消息会被分发到相应的队列中。

@Test
    public void sendTopic1(){
        String message = "Topic消息";
        rabbitTemplate.convertAndSend("topicExchange", "item.ded", message);
        System.out.println(" [x] Sent '" + message + "'");
    }

    @Test
    public void sendTopic(){
        String message = "Topic消息";
        rabbitTemplate.convertAndSend("topicExchange", "ded.update", message);
        System.out.println(" [x] Sent '" + message + "'");
    }

Headers: 头信息模式

Headers Exchange（头信息交换机）是一种基于消息头属性进行路由的消息传递机制。与直接模式（Direct Exchange）、扇出模式（Fanout Exchange）和主题模式（Topic Exchange）不同，Headers Exchange允许你根据消息的头部属性来决定路由逻辑，而不是简单的字符串匹配。
配置Headers Exchange、Queue及Bindings
创建一个配置类来声明Headers Exchange、队列以及它们之间的绑定关系。这里我们创建两个队列，并将它们通过不同的头部属性绑定到同一个Headers Exchange上。

@RabbitListener(bindings = @QueueBinding(
		value = @Queue(value = "queue6", durable = "true"),
		exchange = @Exchange(value = "headersExchange", type = "headers"),
		arguments = {
				@Argument(name = "type", value = "info"),
				@Argument(name = "format", value = "json")
		}))
public void receiveMessageHeaderFromQueue1(String message) {
	System.out.println("Queue1 [x] Received '" + message + "'");
}

@RabbitListener(bindings = @QueueBinding(
		value = @Queue(value = "queue7", durable = "true"),
		exchange = @Exchange(value = "headersExchange", type = "headers"),
		arguments = {
				@Argument(name = "type", value = "error")
		}))
public void receiveMessageHeaderFromQueue2(String message) {
	System.out.println("Queue2 [x] Received '" + message + "'");
}

发送消息

@Test
public void sendHeader(){
	Map<String, Object> headers = new HashMap<>();
	headers.put("type", "info");
	headers.put("format", "json");
	sendMessage(headers, "This is an info message in JSON format.");
}

public void sendMessage(Map<String, Object> headers, String payload) {
	MessageProperties messageProperties = new MessageProperties();
	headers.forEach(messageProperties::setHeader);
	Message message = new Message(payload.getBytes(), messageProperties);
	rabbitTemplate.send("headersExchange", "", message);
	System.out.println(" [x] Sent with headers: " + headers + " and payload: " + payload);
}

使用DEMO地址

https://gitee.com/song_of_the_heart/mq-demo

异常问题记录

应该是 最新版的问题非集群， 退回到 3.8版本, docker 测试

org.springframework.amqp.AmqpIOException: java.io.IOException
	at org.springframework.amqp.rabbit.support.RabbitExceptionTranslator.convertRabbitAccessException(RabbitExceptionTranslator.java:70) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:603) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.connection.CachingConnectionFactory.createConnection(CachingConnectionFactory.java:725) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.connection.ConnectionFactoryUtils.createConnection(ConnectionFactoryUtils.java:252) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.core.RabbitTemplate.doExecute(RabbitTemplate.java:2210) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:2183) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.core.RabbitTemplate.execute(RabbitTemplate.java:2163) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.core.RabbitAdmin.getQueueInfo(RabbitAdmin.java:463) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.core.RabbitAdmin.getQueueProperties(RabbitAdmin.java:447) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.attemptDeclarations(AbstractMessageListenerContainer.java:1942) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer.redeclareElementsIfNecessary(AbstractMessageListenerContainer.java:1915) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.initialize(SimpleMessageListenerContainer.java:1384) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer$AsyncMessageProcessingConsumer.run(SimpleMessageListenerContainer.java:1230) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
Caused by: java.io.IOException: null
	at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:129) ~[amqp-client-5.14.2.jar:5.14.2]
	at com.rabbitmq.client.impl.AMQChannel.wrap(AMQChannel.java:125) ~[amqp-client-5.14.2.jar:5.14.2]
	at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:147) ~[amqp-client-5.14.2.jar:5.14.2]
	at com.rabbitmq.client.impl.AMQConnection.start(AMQConnection.java:439) ~[amqp-client-5.14.2.jar:5.14.2]
	at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:1225) ~[amqp-client-5.14.2.jar:5.14.2]
	at com.rabbitmq.client.ConnectionFactory.newConnection(ConnectionFactory.java:1173) ~[amqp-client-5.14.2.jar:5.14.2]
	at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.connectAddresses(AbstractConnectionFactory.java:641) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.connect(AbstractConnectionFactory.java:616) ~[spring-rabbit-2.4.12.jar:2.4.12]
	at org.springframework.amqp.rabbit.connection.AbstractConnectionFactory.createBareConnection(AbstractConnectionFactory.java:566) ~[spring-rabbit-2.4.12.jar:2.4.12]
	... 12 common frames omitted

http://www.kler.cn/a/540616.html

相关文章：

  • 采用分步式无线控制架构实现水池液位自动化管理
  • 12c及以后 ADG主备切换
  • 企业级Mysql实战
  • 【漫话机器学习系列】082.岭回归（或脊回归）中的α值（alpha in ridge regression）
  • 【Nginx + Keepalived 实现高可用的负载均衡架构】
  • 链表和 list
  • 分桶函数的使用
  • 深入理解概率密度函数和概率的关系
  • redis高级数据结构Stream
  • 控制论与信息论：维纳和香农的核心分歧在于对「信息本质」的理解
  • AntDesign X 报错：Cannot read properties of undefined (reading ‘_context‘)
  • 一个树形结构表格数据库建表语句
  • Json-RPC项目框架(二)
  • 宝珀乾坤卡罗素（Blancpain Carrousel of Yin and Yang）：故宫收藏的机械艺术瑰宝
  • 【浏览器多开】Google Chrome 谷歌浏览器分身术
  • 2.4 异步回调参数捕获技巧详解
  • 使用LangChain自定义tools的三种方式
  • 穷举vs暴搜vs深搜vs回溯vs剪枝系列一＞不同路径 III
  • c/c++蓝桥杯经典编程题100道（19）质因数分解
  • 博客项目-day02（登录功能）
  • Django在终端创建项目（pycharm Windows）
  • Ollama+Chatbox本地部署运行deepseek
  • MySQL主从同步+binlog
  • ffmpeg -demuxers
  • 《optee系统架构从入门到精通》
  • 征程 6 相比征程 5 对算子支持扩展的具体案例讲解