RabbitMq 基础

ops/2025/1/19 19:18:00/
文章目录
  • 一、初识 MQ
    • 1.1 同步调用:
    • 1.2 异步调用:
  • 二、RabbitMQ
  • 三、SpringAMQP
    • 3.1 依赖和配置文件
    • 3.2 消息发送和接收:
      • 3.2.1 消息发送:
      • 3.2.2 消息接收:
    • 3.3 WorkQueues 模型:
    • 3.4 交换机类型:
      • 3.4.1 Fanout 交换机:
      • 3.4.2 Direct 交换机:
      • 3.4.3 Topic 交换机:
    • 3.5 声明交换机和队列:
      • 3.5.1 使用 API:
        • 3.5.1.1 fanout 示例:
        • 3.5.1.2 direct 示例:
      • 3.5.2 使用注解:
        • 3.5.2.1 fanout 示例:
        • 3.5.2.2 direct 示例:
    • 3.6 消息转化器:

一、初识 MQ

1.1 同步调用:

我们观察下,下面这个余额支付功能的流程图:

image-20241128143225761

如果我们采用的是基于 OpenFeign 的同步调用,也就是说业务执行流程是这样的:

  • 支付服务需要先调用用户服务完成余额扣减。
  • 然后支付服务自己要更新支付流水单的状态。
  • 然后支付服务调用交易服务,更新业务订单状态为已支付。

三个步骤依次执行。

这其中就存在3个问题:

  1. 拓展性差:

我们目前的业务相对简单,但是随着业务规模扩大,产品的功能也在不断完善。

在大多数电商业务中,用户支付成功后都会以短信或者其它方式通知用户,告知支付成功。假如后期产品经理提出这样新的需求,你怎么办?是不是要在上述业务中再加入通知用户的业务?

某些电商项目中,还会有积分或金币的概念。假如产品经理提出需求,用户支付成功后,给用户以积分奖励或者返还金币,你怎么办?是不是要在上述业务中再加入积分业务、返还金币业务?

最终你的支付业务会越来越臃肿:

image-20241128143524312

也就是说每次有新的需求,现有支付逻辑都要跟着变化,代码经常变动,不符合开闭原则,拓展性不好。

  1. 性能下降:

由于我们采用了同步调用,调用者需要等待服务提供者执行完返回结果后,才能继续向下执行,也就是说每次远程调用,调用者都是阻塞等待状态。最终整个业务的响应时长就是每次远程调用的执行时长之和:

image-20241128143701876

假如每个微服务的执行时长都是 50ms,则最终整个业务的耗时可能高达 300ms,性能太差了。

  1. 级联失败:

由于我们是基于 OpenFeign 调用交易服务、通知服务。当交易服务、通知服务出现故障时,整个事务都会回滚,交易失败。

这其实就是同步调用的级联失败问题。

不能因为短信通知、更新订单状态失败而回滚整个事务(这些都不是支付服务的主线任务)。

而要解决这些问题,我们就必须用异步调用的方式来代替同步调用

1.2 异步调用:

异步调用方式其实就是基于消息通知的方式,一般包含三个角色:

  • 消息发送者:投递消息的人,就是原来的调用方。
  • 消息Broker:管理、暂存、转发消息。
  • 消息接收者:接收和处理消息的人,就是原来的服务提供方。

image-20241128144127753

还是以余额支付业务为例:

image-20241128144218877

除了扣减余额、更新支付流水单状态以外,其它调用逻辑全部取消。而是改为发送一条消息到 Broker。而相关的微服务都可以订阅消息通知,一旦消息到达 Broker,则会分发给每一个订阅了的微服务,处理各自的业务。

假如产品经理提出了新的需求,比如要在支付成功后更新用户积分。支付代码完全不用变更,而仅仅是让积分服务也订阅消息即可:

image-20241128144244787

不管后期增加了多少消息订阅者,作为支付服务来讲,执行问扣减余额、更新支付流水状态后,发送消息即可。业务耗时仅仅是这三部分业务耗时,仅仅 100ms,大大提高了业务性能。

另外,不管是交易服务、通知服务,还是积分服务,他们的业务与支付关联度低。现在采用了异步调用,解除了耦合,他们即便执行过程中出现了故障,也不会影响到支付服务。

综上,异步调用的优势包括:

  • 耦合度更低
  • 性能更好
  • 业务拓展性强
  • 故障隔离,避免级联失败

当然,异步通信也并非完美无缺,它存在下列缺点:

  • 完全依赖于 Broker 的可靠性、安全性和性能
  • 架构复杂,后期维护和调试麻烦

消息 Broker,目前常见的实现方案就是消息队列(MessageQueue),简称为 MQ。

下面我们就来进行 RabbitMQ 的学习。

二、RabbitMQ

RabbitMQ 是基于 Erlang 语言开发的开源消息通信中间件,官网地址:

https://www.rabbitmq.com/

接下来,我们就学习它的基本概念和基础用法。

安装:

基于 Docker 来安装 RabbitMQ,使用下面的命令即可:

docker run 

-e RABBITMQ_DEFAULT_USER=itheima
-e RABBITMQ_DEFAULT_PASS=123321
-v mq-plugins:/plugins
–name mq
–hostname mq
-p 15672:15672
-p 5672:5672
–network hm-net
-d
rabbitmq:3.8-management

可以看到在安装命令中有两个映射的端口:

  • 15672:RabbitMQ提供的管理控制台的端口
  • 5672:RabbitMQ的消息发送处理接口

安装完成后,我们访问 http://ip地址:15672即可看到管理控制台。首次访问需要登录,默认的用户名和密码在配置文件中已经指定了。

登录后即可看到管理控制台总览页面:

image-20241128144848440

RabbitMQ 对应的架构如图:

image-20241128144930936

其中包含几个概念:

  • publisher:生产者。
  • consumer:消费者。
  • queue:队列,存储消息。生产者投递的消息会暂存在消息队列中,等待消费者处理。
  • exchange:交换机,负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。
  • virtual host:虚拟主机,起到数据隔离的作用。每个虚拟主机相互独立,有各自的 exchange、queue。这个就像一个 mysql 有多个数据库每个数据库有自己的表。

在 MQ 的客户端也可以实现收发消息,比较简单,所以这里跳过。

三、SpringAMQP

将来我们开发业务功能的时候,肯定不会在控制台收发消息,而是应该基于编程的方式。由于RabbitMQ采用了 AMQP 协议,因此它具备跨语言的特性。任何语言只要遵循 AMQP 协议收发消息,都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。

但是,RabbitMQ 官方提供的 Java 客户端编码相对复杂,一般生产环境下我们更多会结合 Spring 来使用。而 Spring 的官方刚好基于RabbitMQ 提供了这样一套消息收发的模板工具:SpringAMQP。并且还基于 SpringBoot 对其实现了自动装配,使用起来非常方便。

SpringAmqp 的官方地址:

https://spring.io/projects/spring-amqp

SpringAMQP 提供了三个功能:

  • 自动声明队列、交换机及其绑定关系。
  • 基于注解的监听器模式,异步接收消息。
  • 封装了 RabbitTemplate 工具,用于发送消息。

3.1 依赖和配置文件

要进行下面的操作,首先我们需要引入相关依赖:

<!--AMQP依赖,包含RabbitMQ-->
<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId>
</dependency>

其次进行配置文件:

application.yml中添加配置:

spring:rabbitmq:host: xxxxx # 你的虚拟机IPport: 5672 # 端口virtual-host: /hmall # 虚拟主机username: hmall # 用户名password: 123 # 密码

注意:不论是消息的发送者还是消息的接收者都需要进行上面的配置。

3.2 消息发送和接收:

3.2.1 消息发送:

利用RabbitTemplate实现消息发送,RabbitTemplate 在我们引入依赖和配置完文件启动项目就自动创建了,我们直接进行依赖注入即可。

下面的代码中,没有实现交换机,因为比较复杂,我们下面讲,这里其实就是感受 RabbitTemplate 的使用。队列需要我们在 MQ 控制台手动创建。

public class SpringAmqpTest {@Autowiredprivate RabbitTemplate rabbitTemplate;public void testSimpleQueue() {// 队列名称String queueName = "simple.queue";// 消息String message = "hello, spring amqp!";// 发送消息rabbitTemplate.convertAndSend(queueName, message);}
}
3.2.2 消息接收:
@Component
public class SpringRabbitListener {// 利用RabbitListener来声明要监听的队列信息// 将来一旦监听的队列中有了消息,就会推送给当前服务,调用当前方法,处理消息。@RabbitListener(queues = "simple.queue")public void listenSimpleQueueMessage(String msg) throws InterruptedException {System.out.println("spring 消费者接收到消息:【" + msg + "】");}
}

注意:消息接收的方法参数类型要和消息的发送的参数类型一致。例如上面都是 String 类型。

3.3 WorkQueues 模型:

Work queues,任务模型。简单来说就是让多个消费者绑定到一个队列,共同消费队列中的消息

image-20241128155740962

当消息处理比较耗时的时候,可能生产消息的速度会远远大于消息的消费速度。长此以往,消息就会堆积越来越多,无法及时处理。

此时就可以使用work 模型,多个消费者共同处理消息处理,同一条消息只会被一个消费者处理,这样,消息处理的速度就能大大提高了

不过在 RabbitMQ 中该模型还存在一个问题:消息是平均分配给每个消费者,并没有考虑到消费者的处理能力。

这显然是不合理的。

改进:

在spring中有一个简单的配置,可以解决这个问题,在 application.yml 文件中添加配置:

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

这样就能实现能者多劳的效果,充分利用了每一个消费者的处理能力,可以有效避免消息积压问题。

总结:

Work模型的使用:

  • 多个消费者绑定到一个队列,同一条消息只会被一个消费者处理。
  • 通过设置prefetch来控制消费者预取的消息数量。

3.4 交换机类型:

在之前的案例中,都没有交换机,生产者直接发送消息到队列。而一旦引入交换机,消息发送的模式会有很大变化:

image-20241128161013810

可以看到,在订阅模型中,多了一个exchange角色,而且过程略有变化:

  • Publisher:生产者,不再发送消息到队列中,而是发给交换机。
  • Exchange:交换机,一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。
  • Queue:消息队列也与以前一样,接收消息、缓存消息。不过队列一定要与交换机绑定。
  • Consumer:消费者,与以前一样,订阅队列,没有变化。

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

交换机的类型有四种:

  • Fanout:广播,将消息交给所有绑定到交换机的队列。
  • Direct:订阅,基于 RoutingKey(路由 key)发送给订阅了消息的队列。
  • Topic:通配符订阅,与 Direct 类似,只不过 RoutingKey 可以使用通配符。
  • Headers:头匹配,基于 MQ 的消息头匹配,用的较少。

我们这里讲解前面三种交换机。

3.4.1 Fanout 交换机:

在广播模式下,消息发送流程是这样的:

image-20241128162202662

总结:

交换机的特点:

  • 接收 publisher 发送的消息,将消息按照规则路由到与之绑定的队列。
  • 不能缓存消息,路由失败,消息丢失。
  • FanoutExchange 的会将消息路由到每个绑定的队列。
3.4.2 Direct 交换机:

在 Fanout 模式中,一条消息,会被所有订阅的队列都消费。但是,在某些场景下,我们希望不同的消息被不同的队列消费。这时就要用到 Direct 类型的 Exchange。

image-20241128162611237

在 Direct 模型下:

  • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个RoutingKey(路由key)。
  • 消息的发送方在向 Exchange 发送消息时,也必须指定消息的 RoutingKey
  • Exchange 不再把消息交给每一个绑定的队列,而是根据消息的Routing Key进行判断,只有队列的Routingkey与消息的 Routing key完全一致,才会接收到消息。

总结:

描述下 Direct 交换机与 Fanout 交换机的差异?

  • Fanout 交换机将消息路由给每一个与之绑定的队列。
  • Direct 交换机根据 RoutingKey 判断路由给哪个队列。
  • 如果 Direct 交换机绑定的多个队列具有相同的 RoutingKey,则与 Fanout 功能类似。
3.4.3 Topic 交换机:

Topic类型的ExchangeDirect相比,都是可以根据RoutingKey把消息路由到不同的队列。

只不过Topic类型Exchange可以让队列在绑定BindingKey 的时候使用通配符!

**BindingKey 一般都是由一个或多个单词组成,多个单词之间以.分割。**例如: item.insert

通配符规则:

  • #:匹配 0 个或多个词。
  • *:匹配不多不少恰好 1 个词。

image-20241128163534062

总结:

描述下 Direct 交换机与 Topic 交换机的差异?

  • Topic 交换机与队列绑定时的 bindingKey 可以指定通配符。
  • #:代表 0 个或多个词。
  • *:代表 1 个词。

3.5 声明交换机和队列:

在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时,队列和交换机是程序员定义的,将来项目上线,又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来,交给运维。在这个过程中是很容易出现错误的。

因此推荐的做法是由程序启动时检查队列和交换机是否存在,如果不存在自动创建。

**声明交换机和队列一般是消息的接收者来做的。**这里的背景是微服务,消息的发送者和接收者一般是在不同的项目里面。

注意:如果项目中没有消费者(使用 @RabbitListener),可能创建交换机和队列会失败(MQ 上显示不出来,但是 spring 中对象成功创建),至于为什么会这样,我也不清楚,去网上找,问 AI 都没有这方面的资料。这个问题是我在项目中遇到的,解决了很久,算是比较小众的问题,如果有相同情况的友友,可以试试)。

3.5.1 使用 API:

SpringAMQP 提供了一个 Queue 类,用来创建队列:

image-20241128152047559

SpringAMQP 还提供了一个 Exchange 接口,来表示所有不同类型的交换机:

image-20241128152125575

我们可以自己创建队列和交换机,不过 SpringAMQP 还提供了 ExchangeBuilder 来简化这个过程:

image-20241128152158873

而在绑定队列和交换机时,则需要使用 BindingBuilder 来创建 Binding 对象:

image-20241128152250590

3.5.1.1 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 FanoutConfig {/*** 声明交换机* @return Fanout类型交换机*/@Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange("hmall.fanout");}/*** 第1个队列*/@Beanpublic Queue fanoutQueue1(){return new Queue("fanout.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}/*** 第2个队列*/@Beanpublic Queue fanoutQueue2(){return new Queue("fanout.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}
3.5.1.2 direct 示例:

direct 模式由于要绑定多个 KEY,会非常麻烦,每一个 Key 都要编写一个 binding:

package com.itheima.consumer.config;import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class DirectConfig {/*** 声明交换机* @return Direct类型交换机*/@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange("hmall.direct").build();}/*** 第1个队列*/@Beanpublic Queue directQueue1(){return new Queue("direct.queue1");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");}/*** 第2个队列*/@Beanpublic Queue directQueue2(){return new Queue("direct.queue2");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("red");}/*** 绑定队列和交换机*/@Beanpublic Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");}
}

由于 Topic 交换机的声明和 Direct 差不多,大家照着上面的 Direct ,修改一下类型就能成功创建,这里就不演示了。

3.5.2 使用注解:

基于 @Bean 的方式声明队列和交换机比较麻烦,尤其是 direct 交换机,Spring 还提供了基于注解方式来声明。

3.5.2.1 fanout 示例:
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "fanout.queue1"),exchange = @Exchange(name = "hmall.fanout", type = ExchangeTypes.FANOUT)
))
public void listenFanoutQueue1(String msg){System.out.println("消费者1接收到fanout.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "fanout.queue2"),exchange = @Exchange(name = "hmall.fanout", type = ExchangeTypes.FANOUT)
))
public void listenFanoutQueue2(String msg){System.out.println("消费者2接收到fanout.queue2的消息:【" + msg + "】");
}
3.5.2.2 direct 示例:
@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue1"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "blue"}
))
public void listenDirectQueue1(String msg){System.out.println("消费者1接收到direct.queue1的消息:【" + msg + "】");
}@RabbitListener(bindings = @QueueBinding(value = @Queue(name = "direct.queue2"),exchange = @Exchange(name = "hmall.direct", type = ExchangeTypes.DIRECT),key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg){System.out.println("消费者2接收到direct.queue2的消息:【" + msg + "】");
}

3.6 消息转化器:

convertAndSend 方法可以任意类型的消息。

image-20241128170230045

而在数据传输时,它会把你发送的消息序列化为字节发送给 MQ,接收消息的时候,还会把字节反序列化为 Java 对象。

只不过,默认情况下 Spring 采用的序列化方式是 JDK 序列化。JDK 序列化存在下列问题:

  • 数据体积过大
  • 有安全漏洞
  • 可读性差

所以我们需要换一个消息转化器。

这里采用 JSON 方式来做序列化和反序列化。

在发送者和接受者两个服务中都引入依赖。

<dependency><groupId>com.fasterxml.jackson.dataformat</groupId><artifactId>jackson-dataformat-xml</artifactId><version>2.9.10</version>
</dependency>

注意,如果项目中引入了spring-boot-starter-web依赖,则无需再次引入Jackson依赖。

配置消息转换器,在发送者和接受者两个服务的启动类中添加一个 Bean 即可:

@Bean
public MessageConverter messageConverter(){// 1.定义消息转换器Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();// 2.配置自动创建消息id,用于识别不同消息,也可以在业务中基于ID判断是否是重复消息jackson2JsonMessageConverter.setCreateMessageIds(true);return jackson2JsonMessageConverter;
}

消息转换器中添加的 messageId 可以便于我们将来做幂等性判断。

参考文献:

  • 黑马程序员

结语:
其实写博客不仅仅是为了教大家,同时这也有利于我巩固知识点,和做一个学习的总结,由于作者水平有限,对文章有任何问题还请指出,非常感谢。如果大家有所收获的话,还请不要吝啬你们的点赞收藏和关注,这可以激励我写出更加优秀的文章。

在这里插入图片描述


http://www.ppmy.cn/ops/151456.html

相关文章

新星杯-ESP32智能硬件开发--ESP32的I/O组成

本博文内容导读&#x1f4d5;&#x1f389;&#x1f525; ESP32系统的基础外设开发&#xff1a;IO_MUX和GPIO矩阵 IO_MUX和GPIO矩阵 ESP32的I/O组成了与外部世界交互的基础&#xff0c;ESP32芯片有34个物理GPIO引脚。每个引脚都可用作一个通用I/O&#xff0c;或者连接一个内部…

嵌入式我没有解决问题的能力不知道该怎么做,没有这种能力也就没有活着的必要了

如何像程序员一样思考 — 解决问题的经验教训 --- How to think like a programmer — lessons in problem solving 像程序员一样思考——提高解决问题的能力 - 【雨歌】 - 博客园

latin1_swedish_ci(latin1 不支持存储中文、日文、韩文等多字节字符)

文章目录 1、SHOW TABLE STATUS WHERE Name batch_version;2、latin1_swedish_ci使用场景注意事项修改字符集和排序规则修改表的字符集和排序规则修改列的字符集和排序规则修改数据库的默认字符集和排序规则 3、ALTER TABLE batch_version CONVERT TO CHARACTER SET utf8mb4 C…

Python----Python爬虫(Scrapy的应用:CrawlSpider 使用,爬取小说,CrawlSpider版)

一、CrawlSpider 使用 1.1、CrawlSpider CrawSpiders 是 Scrapy 框架中的一个特殊爬虫类&#xff0c;它用于处理需要跟随链接并抓取多个页面的情况。相比于基本的 Spider 类&#xff0c;CrawSpiders 提供了一个更灵活、更强大的方式来定义爬取规则。 在Scrapy中Spider是所有爬…

springboot 项目配置https

当你的前端网页添加了 https 后&#xff0c;那么 由于… 前端调用后端的接口&#xff0c;同样的也需要配置 https。下面以宝塔为例&#xff0c;如何实现&#xff0c;请看下面讲解&#xff1a; 1.准备好 SSL 证书 application.yml源文件&#xff1a; spring:redis:host: 60.20…

自动驾驶汽车目前面临的最大技术挑战是什么?

1. 感知技术的局限性 恶劣天气和复杂环境&#xff1a;当前的传感器在恶劣天气&#xff08;如雨、雾、雪&#xff09;和复杂道路条件下的感知能力仍有待提高。例如&#xff0c;激光雷达在穿透雨雾时的能力有限&#xff0c;易受强光干扰&#xff1b;摄像头在夜间和恶劣天气中的视…

PT2025 单触控双输出 LED 调光IC

1 产品概述 ● PT2025 是一款单键电容式触摸控制 ASIC&#xff0c;支持单通道触摸输入和单路/双路 PWM 输出&#xff0c;可引脚配置4种模式。主要应用于触摸卫浴镜开关盒&#xff0c;具有介质自适应、高抗干扰、宽工作电压范围、灯光无频闪、外围器件少的突出优势。 2 主要特性…

2024最新版JavaScript逆向爬虫教程-------基础篇之Chrome开发者工具学习

目录 一、打开Chrome DevTools的三种方式二、Elements元素面板三、Console控制台面板四、Sources面板五、Network面板六、Application面板七、逆向调试技巧 7.1 善用搜索7.2 查看请求调用堆栈7.3 XHR 请求断点7.4 Console 插桩7.5 堆内存函数调用7.6 复制Console面板输出 工…