h5网站设计报价,做网站都用什么技术,上海嘉定做网站公司,app商城开发RabbitMQ高级篇请看#xff1a;
RabbitMQ高级篇-CSDN博客
目录
什么是RabbitMQ#xff1f;
MQ 的核心概念
1. RabbitMQ 的核心组件
2. Exchange 的类型
3. 数据流向说明 如何安装RabbitQueue#xff1f;
WorkQueue#xff08;工作队列#xff09;#xff1a;
Fa…RabbitMQ高级篇请看
RabbitMQ高级篇-CSDN博客
目录
什么是RabbitMQ
MQ 的核心概念
1. RabbitMQ 的核心组件
2. Exchange 的类型
3. 数据流向说明 如何安装RabbitQueue
WorkQueue工作队列
Fanout交换机 广播
Direct交换机
Direct 交换机
生产者代码
消费者代码
Topic交换机 SpringAMQP
编辑
1. maven依赖
2. 配置服务端信息
简单示例
注解式声明队列和交换机
编辑
消息转换器
总结如何在IDEA当中使用RabbitMQ 什么是RabbitMQ
它基于AMQP协议Advanced Message Queuing Protocol一种为应用构建消息队列的标准协议。过程中它提供了一些重要模块为消息发送的Producer生产者分发消息的Exchange交换器写入消息的Queue队列和读取消息的Consumer消费者。 MQ 的核心概念
1. 异步处理
问题系统需要执行一些耗时操作如发送邮件、生成报告如果这些操作在主线程执行会影响用户体验或导致系统响应变慢。
MQ 的解决方式
生产者将任务消息放入队列不需要等待任务完成。消费者在后台异步处理任务。
示例
用户下单后系统需要发送订单确认邮件。如果没有 MQ用户可能需要等待邮件发送完成后才能收到订单确认。使用 MQ 后生产者订单服务将“发送邮件”任务放入队列消费者邮件服务异步处理。 2. 系统解耦
问题系统服务之间高度耦合一个服务的变化会导致多个服务需要修改降低开发效率和系统灵活性。
MQ 的解决方式
服务之间通过消息队列通信而不是直接调用。生产者只需要发送消息到 MQ消费者负责处理消息二者互不影响。
示例
用户下单后订单服务需要通知库存服务扣减库存、物流服务生成物流单。如果没有 MQ订单服务需同步调用这些服务的接口导致用户必须等待所有操作完成线性操作响应时间较长且系统耦合度高。使用 MQ 后订单服务将消息发送到消息队列并立即返回“下单成功”的响应。库存服务和物流服务异步订阅消息进行处理各服务独立运行彼此解耦。这样既提升了用户体验也增强了系统的扩展性和稳定性。 3. 削峰填谷
问题在高并发场景下大量请求瞬间涌入可能导致服务过载或崩溃。
MQ 的解决方式
将高并发的请求存入队列消费者按自己的能力逐步处理。队列可以作为缓冲区平衡生产者和消费者之间的处理速度。
示例
秒杀活动中用户请求大量涌入库存系统。没有 MQ库存服务可能因并发过高而宕机。使用 MQ 后所有秒杀请求进入队列库存服务按顺序逐一处理。
好处
防止系统崩溃保障服务稳定性。 4. 数据可靠性
问题数据传输过程中可能因为网络故障、系统宕机等原因导致消息丢失。
MQ 的解决方式
MQ 提供消息持久化功能确保即使系统故障消息也不会丢失。支持消息重试机制确保消息至少被处理一次。
示例
支付系统发送“支付成功”消息给订单系统。如果没有 MQ网络抖动可能导致消息丢失订单状态无法更新。使用 MQ 后消息持久化到磁盘消费者故障恢复后可继续消费消息。 1. RabbitMQ 的核心组件 Virtual Host虚拟主机
定义RabbitMQ 中的逻辑隔离单位类似于一个独立的命名空间。
作用
用于实现不同用户或系统之间的隔离。每个 Virtual Host 下可以有独立的 Exchange交换机、Queue队列和绑定关系。一个 RabbitMQ 服务器可以有多个 Virtual Host。
应用场景
多租户系统例如不同的业务模块可以使用不同的 Virtual Host。 Publisher消息发送者
定义负责向 RabbitMQ 发送消息的生产者应用程序。功能 将消息发送到 Exchange交换机而不是直接发送到 Queue。注意 Publisher 和 Exchange 通过绑定关系决定消息的路由。 Consumer消息消费者
定义负责从 Queue队列中接收消息并处理的应用程序。
功能
消费者直接从队列中读取消息。每个消息只会被一个消费者处理点对点模式。 Queue队列
定义存储消息的缓冲区用于临时保存消息。
功能
消息最终会路由到队列并由消费者从队列中消费。队列可以绑定到多个 Exchange并可根据路由规则接收不同的消息。
特点
持久化队列可以配置为持久化即使 RabbitMQ 服务重启消息也不会丢失。排队顺序消息按照 FIFO先进先出的顺序进行消费。 Exchange交换机
定义负责根据路由规则分发消息的组件。
功能
接收 Publisher 发送的消息并根据路由规则决定将消息发送到哪个 Queue。不直接存储消息消息总是路由到队列中。 2. Exchange 的类型
根据不同的消息路由方式Exchange 有以下几种类型
Direct直连交换机
根据完全匹配的路由键Routing Key将消息发送到指定的队列。适用场景精准匹配例如订单状态更新。
Fanout广播交换机
将消息广播到所有绑定的队列而不考虑路由键。适用场景日志广播、通知推送。
Topic交换机
根据通配符匹配路由键将消息路由到符合条件的队列。适用场景动态路由例如按照“日志级别.模块名”匹配日志消息。 3. 数据流向说明
Publisher - Exchange
消息发送者Publisher将消息发送到 RabbitMQ 的 Exchange交换机。发送时需要指定 Routing Key用于路由消息。
Exchange - Queue
Exchange 根据绑定关系和路由规则将消息分发到一个或多个队列Queue。如果没有匹配的队列消息可能会被丢弃或进入死信队列DLQDead Letter Queue。
Queue - Consumer
消费者Consumer从队列中拉取消息并进行处理。消费者可以是多个每个消息只能被一个消费者消费在同一队列中。 如何安装RabbitQueue
day06-MQ基础 - 飞书云文档 (feishu.cn)
查看以上文档的第二部分即可 WorkQueue工作队列 工作队列的核心理念 任务分发 消息任务由一个生产者Producer发送到队列。多个消费者Consumer从同一个队列中取出消息并处理。 负载均衡 消息会按照一定规则分发给消费者通常每个消费者会处理相同数量的任务轮询机制。不同消费者可以根据其处理能力自行调整消费速率。可以通过配置的prefetch来设置 解耦和异步 生产者和消费者无需直接交互。生产者将任务发送到队列即可消费者独立从队列中获取任务并处理。 Fanout交换机 广播
通过 Fanout Exchange 可以实现消息的广播分发将消息发送给所有绑定的队列。结合 Spring 提供的 RabbitTemplate 工具可以方便地向交换机发送消息极大简化了开发流程。这种模式非常适用于日志收集、通知广播等场景。
发送消息到交换机的 API 示例
Test
public void testFanoutExchange() {// 定义交换机的名称String exchangeName itcast.fanout;// 定义消息内容String message Hello, everyone!;// 发送消息到指定交换机// 参数解释// - exchangeName交换机名称// - routingKey路由键在 Fanout 交换机中会被忽略// - message发送的消息内容rabbitTemplate.convertAndSend(exchangeName, , message);
}Direct交换机
Direct 交换机会根据消息携带的 RoutingKey将消息发送到与交换机绑定且 RoutingKey 完全匹配的队列。 特性精确匹配分发消息广播到所有绑定的队列RoutingKey 是否重要必须匹配不考虑直接广播绑定关系每个队列可以绑定不同的 RoutingKey所有绑定的队列都会接收消息适用场景精确路由如订单状态更新广播通知如日志、系统更新
Direct 交换机
场景订单服务需要将不同类型的订单路由到对应的队列如普通订单和优先订单。
绑定关系
队列 normal_orders 绑定 RoutingKey normal。队列 priority_orders 绑定 RoutingKey priority。
消息发送
消息 RoutingKey normal会被路由到 normal_orders 队列。消息 RoutingKey priority会被路由到 priority_orders 队列。 生产者代码
生产者将消息发送到 Direct 交换机并指定不同的 RoutingKey。
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;Service
public class SimpleProducer {private final RabbitTemplate rabbitTemplate;public SimpleProducer(RabbitTemplate rabbitTemplate) {this.rabbitTemplate rabbitTemplate;}public void sendMessage(String message) {// 直接发送消息到交换机rabbitTemplate.convertAndSend(simple_exchange, simple_routing_key, message);System.out.println(Sent message: message);}
}消费者代码
定义两个消费者分别监听普通队列和优先队列。
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;Service
public class SimpleConsumer {// 监听队列直接声明队列名称RabbitListener(queues simple_queue)public void receiveMessage(String message) {System.out.println(Received message: message);}
}Topic交换机
opic Exchange 是 RabbitMQ 的一种交换机类型它根据消息的路由键RoutingKey和绑定键BindingKey的模式匹配规则将消息路由到一个或多个队列。它是 Direct Exchange 的增强版支持模糊匹配和通配符。 匹配规则精确匹配完全相等模糊匹配支持 * 和 # 通配符RoutingKey 示例order.createdorder.* 或 order.#适用场景简单、明确的路由需求复杂、动态的路由需求扩展性固定路由灵活性低动态路由灵活性高具体匹配示例 RoutingKey 模式消息 RoutingKey是否匹配order.*order.created✔️order.*order.created.new❌ SpringAMQP
Spring AMQP 是 Spring 提供的一个用于与 AMQPAdvanced Message Queuing Protocol高级消息队列协议 通信的模块化框架。它为基于 Spring 的应用程序集成 AMQP 消息中间件例如 RabbitMQ提供了便捷的方法简化了消息的发送、接收和处理。 1. maven依赖 !--AMQP依赖包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency
2. 配置服务端信息
spring:rabbitmq:host: 192.168.88.130 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码listener:simple:prefetch: 1 # 控制消费者预取的消息数量处理完一条再处理如果希望RabbitMQ轮询访问可以不设置这个 什么是 prefetch
prefetch预取数量 是 RabbitMQ 的一个设置用于控制消息消费者Consumer每次从队列中预取消息的数量。
它定义了在消费者确认ACK之前RabbitMQ 可以向消费者发送的未确认消息的最大数量。
场景
假如 prefetch 设置为 1RabbitMQ 会向消费者一次发送 1 条消息只有这条消息被确认后才会发送下一条消息。
如果设置为一个较大的数字例如 10RabbitMQ 会一次性发送多条消息消费者可以并行处理这些消息。
为什么需要设置 prefetch
在多个消费者监听同一个队列的场景下prefetch 设置为 1可以确保消息在消费者之间更均匀地分布。
防止某些消费者处理速度慢但仍然接收大量消息导致处理延迟。
简单示例
发送消息
Autowired
private RabbitTemplate rabbitTemplate;public void sendMessage(String message) {rabbitTemplate.convertAndSend(exchange_name, routing_key, message);
}接收消息
RabbitListener(queues queue_name)
public void receiveMessage(String message) {System.out.println(Received: message);
}注解式声明队列和交换机 RabbitListener
用于监听队列当队列接收到消息时触发对应方法处理消息。
QueueBinding
声明队列与交换机的绑定关系。
包含
Queue声明队列名称和属性。
Exchange声明交换机名称、类型和属性。key指定绑定时使用的路由键。 RabbitListener(bindings QueueBinding(value Queue(namedirect.queue1,durable true),exchange Exchange(namehmall.direct,type ExchangeTypes.DIRECT),key {red,blue}))public void listenDirectQueue(String message) throws InterruptedException {System.err.println(消费者1.....................接收到消息 message, LocalTime.now());}消息转换器
在 RabbitMQ 中消息默认是以字节数组的形式在队列中传输的。如果我们希望以更方便的方式传递和处理对象如 JSON、XML 或 Java 对象就需要使用 消息转换器Message Converter 来完成消息的序列化与反序列化。
消息转换器主要负责
序列化将 Java 对象转换为消息格式如 JSON、XML 或字节数组发送到 RabbitMQ。反序列化将从 RabbitMQ 接收到的消息转换为 Java 对象供消费者处理。
转换器功能SimpleMessageConverter默认的消息转换器支持简单类型如 String、byte[] 和 Serializable 对象。Jackson2JsonMessageConverter使用 Jackson 将 Java 对象转换为 JSON 格式或将 JSON 消息转换为 Java 对象。Jaxb2MarshallerMessageConverter使用 JAXB 将 Java 对象转换为 XML 格式或将 XML 消息转换为 Java 对象。ContentTypeDelegatingMessageConverter根据消息的 content_type 动态选择合适的消息转换器。 导入依赖 dependencygroupIdcom.fasterxml.jackson.core/groupIdartifactIdjackson-databind/artifactId/dependency
记得要给生产者和消费者都需要设置消息转换器。 Beanpublic MessageConverter messageCoverter(){return new Jackson2JsonMessageConverter();}
JSON 格式需要配置 Jackson2JsonMessageConverter。XML 格式需要配置 Jaxb2MarshallerMessageConverter。字符串/字节数组不需要额外配置默认的 SimpleMessageConverter 即可。
总结如何在IDEA当中使用RabbitMQ
开始之前请确定RabbitMQ正在运行哦
引入依赖 !--AMQP依赖包含RabbitMQ--
dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId
/dependency
配置消息转换器Config
Configuration
ConditionalOnClass(RabbitTemplate.class) //该注解表示只有在当前的类路径classpath中存在 RabbitTemplate 类时注解标注的配置类或 Bean 才会生效。
public class MqConfig {/*** 序列化存储JSON字符串* return*/Beanpublic MessageConverter messageCoverter(){Jackson2JsonMessageConverter jjmc new Jackson2JsonMessageConverter();
// jjmc.setCreateMessageIds(true);// 设置消息转换器在创建消息时是否附加唯一的消息 ID。//true 表示会生成消息 ID便于跟踪和调试消息。return jjmc;}
}配置yaml
spring:rabbitmq:host: 192.168.150.101 # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码
之后就可以再需要MQ的地方注入MQ并且实现消费者和生产者的代码了
