RabbitMQ

YangeIT大约 40 分钟基础服务框架MQ同步异步RabbitMQ交换机

RabbitMQ

课程内容

  • 了解同步和异步通讯的优缺点 ❤️ 🍐
  • 了解各种MQ技术的优缺点
  • 能利用SpringAMQP收发消息❤️
  • 能基于@Bean声明队列、交换机、绑定关系❤️
  • 能基于@RabbitListener声明队列、交换机、绑定关系❤️
  • 能配置SpringAMQP的消息转换器

知识储备

  1. 理解同步和异步的概念

1.初识MQ

1.1.同步和异步通讯 ❤️ 🍐

初识MQ

同步和异步通讯

微服务间通讯有同步和异步两种方式：

  • 同步通讯：就像打电话，需要实时响应。
  • 异步通讯：就像发邮件，不需要马上回复。
image-20210717161939695
image-20210717161939695

两种方式各有优劣，打电话可以立即得到响应同步通讯，但是你却不能跟多个人同时通话 。发送邮件可以同时与多个人收发邮件异步通讯，但是往往响应会有延迟。

同步通讯

我们之前学习的Feign调用就属于同步方式，虽然调用可以实时得到结果，但存在下面的问题：

同步通讯存在的问题
同步通讯存在的问题

总结：👇

同步调用的优点：

  • 时效性较强，可以立即得到结果

同步调用的问题：

  • 耦合度高
  • 性能和吞吐能力下降
  • 有额外的资源消耗
  • 有级联失败问题

1.2 消息队列解决方案 🍐

消息队列解决方案

MQ 中文是消息队列（MessageQueue），字面来看就是存放消息的队列。也就是事件驱动架构中的Broker。

比较常见的MQ实现：

  • ActiveMQ
  • RabbitMQ
  • RocketMQ
  • Kafka

几种常见MQ的对比：

RabbitMQActiveMQ👇RocketMQKafka
公司/社区RabbitApache阿里Apache
开发语言ErlangJavaJavaScala&Java
协议支持AMQP，XMPP，SMTP，STOMPOpenWire,STOMP，REST,XMPP,AMQP自定义协议自定义协议
可用性高一般高高
单机吞吐量一般差高非常高
消息延迟微秒级毫秒级毫秒级毫秒以内
消息可靠性高一般高一般

据统计，目前国内消息队列使用最多的还是 RabbitMQ，再加上其各方面都比较均衡，稳定性也好，因此我们课堂上选择 RabbitMQ 来学习。

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ 、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ 、Kafka

总结

  • 同步调用的优点：
    • 时效性较强，可以立即得到结果
  • 同步调用的问题：
    • 耦合度高,违背开闭原则
    • 性能和吞吐能力下降
    • 有额外的资源消耗
    • 有级联失败问题
  • 异步通信的优点：
    • 耦合度低
    • 吞吐量提升
    • 故障隔离
    • 流量削峰
  • 异步通信的缺点：
    • 依赖于Broker的可靠性、安全性、吞吐能力
    • 架构复杂了，业务没有明显的流程线，不好追踪管理

2.RabbitMQ

2.1安装RabbitMQ

1安装RabbitMQ

RabbitMQ 是基于 Erlang 语言开发的开源消息通信中间件，官网地址： https://www.rabbitmq.com/open in new window

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

我们同样基于 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 hmall \
 -d \
 rabbitmq:3.8-management
  • 命令解读：
  • RABBITMQ_DEFAULT_USER: 用户名
  • RABBITMQ_DEFAULT_PASS:密码
  • mq-plugins: 插件的数据卷
  • 15672：RabbitMQ 提供的管理控制台的端口
  • 5672：RabbitMQ 的消息发送处理接口
  • --network hmall :网络

如果拉取镜像困难的话，可以使用课前资料给大家准备的镜像，利用 docker load -i 镜像名称 命令加载：

image
image

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

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

image
image

收发消息实操

1.交换机

我们打开 Exchanges 选项卡，可以看到已经存在很多交换机：

我们点击任意交换机，即可进入交换机详情页面。仍然会利用控制台中的 publish message 发送一条消息：

这里是由控制台模拟了生产者发送的消息。由于没有消费者存在，最终消息丢失了，这样说明交换机没有存储消息的能力 。

总结

课堂作业

  1. 队列的作用什么？为什么发送了消息，刷新几次后，然后存在？🎤
  2. RabbitMQ架构图中核心的概念有哪些？分别是什么意思？🎤

2.3.数据隔离

数据隔离

1.用户管理

点击 Admin 选项卡，首先会看到 RabbitMQ 控制台的用户管理界面：

这里的用户都是 RabbitMQ 的管理或运维人员。目前只有安装 RabbitMQ 时添加的 itheima 这个用户。仔细观察用户表格中的字段，如下：

  • Name：itheima，也就是用户名
  • Tags：administrator，说明 itheima 用户是超级管理员，拥有所有权限
  • Can access virtual host： /，可以访问的 virtual host，这里的 / 是默认的 virtual host

总结

课堂作业

  1. RabbitMQ virtual host(虚拟主机)有什么应用场景？🎤

3.SpringAMQP

3.1 SpringAMQP简介和入门

SpringAMQP简介和入门

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

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

SpringAmqp 的官方地址：

SpringAMQP 提供了三个功能：

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

这一章我们就一起学习一下，如何利用 SpringAMQP 实现对 RabbitMQ 的消息收发。

代码操作

基本步骤

准备工作：导入入门demo工程

  1. 编写代码发送消息
  2. 编写监听器消费消息

导入 Demo 工程

在课前资料给大家提供了一个 Demo 工程，方便我们学习 SpringAMQP 的使用：

将其复制到你的工作空间，然后用 Idea 打开，项目结构如图：

包括三部分：

  • mq-demo：父工程，管理项目依赖
  • publisher：消息的发送者
  • consumer：消息的消费者

在 mq-demo 这个父工程中，已经配置好了 SpringAMQP 相关的依赖：

<?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>cn.itcast.demo</groupId>
    <artifactId>mq-demo</artifactId>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>publisher</module>
        <module>consumer</module>
    </modules>
    <packaging>pom</packaging>

    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.7.12</version>
        <relativePath/>
    </parent>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
        <!--AMQP依赖，包含RabbitMQ-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
        <!--单元测试-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
        </dependency>
    </dependencies>
</project>

因此，子工程中就可以直接使用 SpringAMQP 了。

3.2.WorkQueues 模型

WorkQueues模型

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

image
image

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

此时就可以使用 work 模型，多个消费者共同处理消息处理，消息处理的速度就能大大提高了。

Work queues也被称为（Task queues），任务模型。

简单来说就是让多个消费者绑定到一个队列，共同消费队列中的消息。

业务场景：

  1. 并行任务处理：适用于需要处理大量任务的场景，多个工作者并发处理任务，确保任务在短时间内完成。例如，电子邮件队列处理、图像处理任务等。
  2. 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。此时就可以使用work 模型，多个消费者共同处理消息处理，速度就能大大提高了。
  3. 负载均衡：用于平均分配任务到可用的工作者，确保每个工作者具有相似的负载。

如银行的叫号系统

接下来，我们就来模拟这样的场景。👇

首先，我们在控制台创建一个新的队列，命名为 work.queue：

代码操作

3.3.1.消息发送

这次我们循环发送，模拟大量消息堆积现象。

在 publisher 服务中的 SpringAmqpTest 类中添加一个测试方法：

/**
     * workQueue
     * 向队列中不停发送消息，模拟消息堆积。
     */
@Test
public void testWorkQueue() throws InterruptedException {
    // 队列名称
    String queueName = "simple.queue";
    // 消息
    String message = "hello, message_";
    for (int i = 0; i < 50; i++) {
        // 发送消息，每20毫秒发送一次，相当于每秒发送50条消息
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}

总结

Work 模型的使用：

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

课堂作业

  1. Work 模型适用于什么场景？🎤
  2. 预取有什么作用？

3.2.交换机类型

交换机类型

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

image
image

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

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

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

交换机的类型有四种： 👇

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

课堂中，我们讲解前面的三种交换机模式。 🎯

总结

课堂作业

  1. Exchange（交换机）只负责什么？能存储消息吗？🎤
  2. Fanout：广播有何应用场景？
  3. Direct：订阅模式有何应用场景？
  4. Topic：通配符订阅 有何应用场景？

3.4.Fanout 交换机

前言

Fanout，英文翻译是扇出，我觉得在 MQ 中叫广播更合适。

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

image
image
  • 1） 可以有多个队列
  • 2） 每个队列都要绑定到 Exchange（交换机）
  • 3） 生产者发送的消息，只能发送到交换机
  • 4） 交换机把消息发送给绑定过的所有队列
  • 5） 订阅队列的消费者都能拿到消息

业务场景：

  • 实时通知：用于向多个订阅者广播实时通知、事件或更新。例如，社交媒体平台上的关注者通知，新闻订阅服务。
  • 日志和监控：用于将应用程序的日志和性能数据广播到多个订阅者，以进行分析和监控。

验证案例流程：

image
image
  • 创建一个名为 hmall.fanout 的交换机，类型是 Fanout
  • 创建两个队列 fanout.queue1 和 fanout.queue2，绑定到交换机 hmall.fanout

代码操作

1.声明队列和交换机

在控制台创建队列 fanout.queue1:

在创建一个队列 fanout.queue2：

然后再创建一个交换机：

然后绑定两个队列到交换机：

总结

交换机的作用是什么？

  • 接收 publisher 发送的消息
  • 将消息按照规则路由到与之绑定的队列
  • 不能缓存消息，路由失败，消息丢失
  • FanoutExchange 的会将消息路由到每个绑定的队列

课堂作业

  1. 日志中有哪些信息？🎤

3.5.Direct交换机

Direct交换机

在 Fanout 模式中，一条消息，会被所有订阅的队列都消费。

但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到 Direct 类型的 Exchange。

在Direct模型下：

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

如：医院的叫号系统

案例需求如图：

  1. 声明一个名为 hmall.direct 的交换机
  2. 声明队列 direct.queue1，绑定 hmall.direct，bindingKey 为 blud 和 red
  3. 声明队列 direct.queue2，绑定 hmall.direct，bindingKey 为 yellow 和 red
  4. 在 consumer 服务中，编写两个消费者方法，分别监听 direct.queue1 和 direct.queue2
  5. 在 publisher 中编写测试方法，向 hmall.direct 发送消息

代码操作

1.声明队列和交换机

首先在控制台声明两个队列 direct.queue1 和 direct.queue2，这里不再展示过程：

然后声明一个 direct 类型的交换机，命名为 hmall.direct:

然后使用 red 和 blue 作为 key，绑定 direct.queue1 到 hmall.direct：

同理，使用 red 和 yellow 作为 key，绑定 direct.queue2 到 hmall.direct，步骤略，最终结果：

总结

课堂作业

  1. Direct交换机有哪些应用场景？🎤
  2. 描述下 Direct 交换机与 Fanout 交换机的差异？
    • Fanout 交换机将消息路由给每一个与之绑定的队列
    • Direct 交换机根据 RoutingKey 判断路由给哪个队列
    • 如果多个队列具有相同的 RoutingKey，则与 Fanout 功能类似

3.6.Topic交换机

Topic交换机

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

BindingKey 一般都是有一个或多个单词组成，多个单词之间以 . 分割，例如： item.insert

通配符规则：

  • #：匹配一个或多个词
  • *：匹配不多不少恰好 1 个词

举例：

  • item.#：能够匹配 item.spu.insert 或者 item.spu
  • item.*：只能匹配 item.spu

案例图示：

假如此时 publisher 发送的消息使用的 RoutingKey 共有四种：

  • china.news 代表有中国的新闻消息；
  • china.weather 代表中国的天气消息；
  • japan.news 则代表日本新闻
  • japan.weather 代表日本的天气消息；

解释：

  • topic.queue1：绑定的是 china.# ，凡是以 china. 开头的 routing key 都会被匹配到，包括：
    • china.news
    • china.weather
  • topic.queue2：绑定的是 #.news ，凡是以 .news 结尾的 routing key 都会被匹配。包括:
    • china.news
    • japan.news

接下来，我们就按照上图所示，来演示一下 Topic 交换机的用法。

代码操作

首先，在控制台按照图示例子创建队列、交换机，并利用通配符绑定队列和交换机。此处步骤略。最终结果如下：

总结

课堂作业

  1. 描述下 Direct 交换机与 Topic 交换机的差异？🎤
  • Topic 交换机接收的消息 RoutingKey 必须是多个单词，以 . 分割
  • Topic 交换机与队列绑定时的 bindingKey 可以指定通配符
  • #：代表 0 个或多个词
  • *：代表 1 个词？

3.7.声明队列和交换机

声明队列和交换机

在之前我们都是基于 RabbitMQ 控制台来创建队列、交换机。

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

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

操作

1.基本 API

SpringAMQP 提供了一个 Queue 类，用来创建队列：

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

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

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

3.9.消息转换器

消息转换器

Spring 的消息发送代码接收的消息体是一个 Object：

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

只不过，默认情况下 Spring 采用的序列化方式是 JDK 序列化。

众所周知，JDK 序列化存在下列问题：

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

我们来测试一下。

代码操作

1.测试默认转换器

1️⃣ 1）创建测试队列

首先，我们在 consumer 服务中声明一个新的配置类：

利用 @Bean 的方式创建一个队列，

具体代码：

package com.itheima.consumer.config;

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

@Configuration
public class MessageConfig {

    @Bean
    public Queue objectQueue() {
        return new Queue("object.queue");
    }
}

注意，这里我们先不要给这个队列添加消费者，我们要查看消息体的格式。

重启 consumer 服务以后，该队列就会被自动创建出来了：

2️⃣ 2）发送消息

我们在 publisher 模块的 SpringAmqpTest 中新增一个消息发送的代码，发送一个 Map 对象：

@Test
public void testSendMap() throws InterruptedException {
    // 准备消息
    Map<String,Object> msg = new HashMap<>();
    msg.put("name", "柳岩");
    msg.put("age", 21);
    // 发送消息
    rabbitTemplate.convertAndSend("object.queue", msg);
}

发送消息后查看控制台：

可以看到消息格式非常不友好。

总结

课堂作业

  1. 消息转换器的有什么作用？🎤

4.业务改造

业务改造

案例需求：改造余额支付功能，将支付成功后基于 OpenFeign 的交易服务的更新订单状态接口的同步调用，改为基于 RabbitMQ 的异步通知。

如图：

说明，我们只关注交易服务，步骤如下：

  • 定义 topic 类型交换机，命名为 pay.topic
  • 定义消息队列，命名为 mark.order.pay.queue
  • 将 mark.order.pay.queue 与 pay.topic 绑定，BindingKey 为 pay.success
  • 支付成功时不再调用交易服务更新订单状态的接口，而是发送一条消息到 pay.topic，发送消息的 RoutingKey 为 pay.success，消息内容是订单 id
  • 交易服务监听 mark.order.pay.queue 队列，接收到消息后更新订单状态为已支付

代码操作

4.1.配置 MQ

不管是生产者还是消费者，都需要配置 MQ 的基本信息。分为两步：

1）添加依赖：

<!--消息发送-->
  <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-amqp</artifactId>
  </dependency>

2）配置 MQ 地址：

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

总结

课堂作业

  1. 为什么支付业务要用到mq🎤

5.作业

5.1.作业1

抽取共享的 MQ 配置 ·za ggghmnmngzsaAAASTYTRQ啊发噶CM.,wsssaA

A P[908] 将 MQ 配置抽取到 Nacos 中管理，微服务中直接使用共享配置。

5.2. 作业2

改造下单功能

改造下单功能，将基于 OpenFeign 的清理购物车同步调用，改为基于 RabbitMQ 的异步通知：

  • 定义 topic 类型交换机，命名为 trade.topic
  • 定义消息队列，命名为 cart.clear.queue
  • 将 cart.clear.queue 与 trade.topic 绑定，BindingKey 为 order.create
  • 下单成功时不再调用清理购物车接口，而是发送一条消息到 trade.topic，发送消息的 RoutingKey 为 order.create，消息内容是下单的具体商品、当前登录用户信息
  • 购物车服务监听 cart.clear.queue 队列，接收到消息后清理指定用户的购物车中的指定商品

5.3.作业3

登录信息传递优化

某些业务中，需要根据登录用户信息处理业务，而基于 MQ 的异步调用并不会传递登录用户信息。前面我们的做法比较麻烦，至少要做两件事：

  • 消息发送者在消息体中传递登录用户
  • 消费者获取消息体中的登录用户，处理业务

这样做不仅麻烦，而且编程体验也不统一，毕竟我们之前都是使用 UserContext 来获取用户。

大家思考一下：有没有更优雅的办法传输登录用户信息，让使用 MQ 的人无感知，依然采用 UserContext 来随时获取用户。

参考资料：https://b11et3un53m.feishu.cn/wiki/PAxAw3A04ihHa0ko1nTcXxHUndgopen in new window

焱哥的博客
Copyright © 2025 YangeIT