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

消息队列系列一：RabbitMQ入门讲解

article 2024/11/15 9:57:09

目录

前言

一、案例引入

二、消息队列产品与功能介绍

2.1 同步调用和异步调用的区别

2.1.1 同步调用

2.1.2 异步调用

2.2 消息队列产品对比

2.3 常见的消息队列模式 

2.3.1 简单队列模式

2.3.2 工作队列模式

2.3.3 发布/订阅模式

2.3.4 路由模式

2.3.5 主题模式

三、RabbitMQ使用入门

3.1 RabbitMQ的线上部署

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

3.3 简单消息收发实验

3.4 虚拟主机实现数据隔离实验

四、Java客户端使用RabbitMQ

4.1 SpringAMQP

4.2 SpringAMQP快速入门实验

4.3 RabbitMQ工作队列模式实验

4.4 RabbitMQ交换机

4.5 RMQ_Fanout交换机实验

4.6 RMQ_Direct交换机实验

4.7 RMQ_Topic交换机实验

4.8 基于Java Bean创建队列与交换机

4.8.1 SpringAMQP提供的创建API

4.8.2 创建Fanout交换实验

4.8.3 创建direct交换机实验

4.8.3 总结

4.9 基于注解声明创建队列于交换机

4.10 消息转换器使用

4.10.1 默认JDK转换器实验

4.10.2 使用JSON格式的消息转换器

4.10.3 消费者接收消息改造

五、黑马商城微服务实战改造

5.1 业务改造实验说明

5.2 配置RabbitMQ消息队列

5.3 消息接收配置

5.4 消息发送配置

5.5 启动项目，完成测试

六、RabbitMQ相关知识追问巩固

前言

本文是接着黑马微服务的视频进一步学习消息队列的知识点的。使用文中举的例子包括最后MQ入门的实验都会使用黑马课程中的资料。我会尽量融入自己的理解，拓展更多MQ相关的知识。当然如果你问我这种小型单体项目或者就黑马商城这种入门级别的微服务项目有必要使用消息队列么？我想我会给出以下回答：

有必要，如果你是面向面试编程的话，消息队列还是必不可少的技术栈了。其次，如果你想更好的掌握消息队列的使用，也是很有必要去实践一下的。不然光听不练，你以为你会了，实际上遇到实际的需求你根本不会想到可以使用消息队列。

一、案例引入

【案例引入】：当用户进行登录时，一般我们都希望尽可能的只调用与用户管理相关的接口，对于例如账号风险判断、短信提示服务、甚至是用户积分之类的额外需求服务我们并不想化过多的精力去“了解它们”。

其次，从下面的流程图你发现了，这些个附加的微服务是线性先后执行的，这样一来在性能消耗方面也并不算优秀。

凝练一下上面的需求，我们希望：

  1. 系统解耦，用户微服务不需要在方法内执着于其他服务的调用，而是交给其他实现。
  2. 同步改异步，扩展业务异步执行，提高业务执行效率。

于是方案可以更改成这样：

增加一个 “邮箱”。用户微服务完成自己部分的业务后，不再需要花费精力去了解风控微服务相关方法。而是将“已完成”的信号发送到邮箱，其他服务订阅相关的邮箱，就能取出这个信号。实现了系统解耦这个需求。

与此同时，只要有多个服务订阅该“邮箱”，都能各自取出来，并发执行。实现了异步调用。突破了性能瓶颈，提高业务执行效率。

而这之中的“邮箱”角色，正是我们这节课需要学习的消息队列（Message Queue）MQ

二、消息队列产品与功能介绍

在介绍消息队列产品之前，我们先讲讲异步调用（消息队列） 与 同步调用至今的区别。

2.1 同步调用和异步调用的区别

2.1.1 同步调用

简单理解，同步调用就像打电话一样，双方需要保持稳定的“连接”并即时获取“反馈”。在通话进行的过程中，通话双方都无法再与其他人建立连接，即处于“占线”状态。

  • 优点在于：
    1. 即时反馈：调用者能够立即得到被调用者的响应，便于处理需要即时结果的场景。
    2. 顺序执行：任务按照调用顺序依次执行，易于理解和调试。
    3. 简单直观：对于简单的任务，同步调用的逻辑更为直接和易于理解。
  • 缺点在于：
    1. 阻塞等待：调用者需要等待被调用者完成才能继续执行，这可能导致资源闲置和效率低下。
    2. 并发性差：无法同时处理多个任务，限制了系统的并发处理能力。
    3. 资源占用：长时间占用连接和资源，可能导致系统资源耗尽或瓶颈。
    4. 级联失败：因为是顺序执行的，当链路出现故障时，会波及到整条业务链
同步调用示意图

2.1.2 异步调用

异步调用则像微信聊天，你可以在同一时间内接收和处理来自多个人的消息，并根据需要选择性地回复。当然，微信聊天并不是直接把消息发送给对方，而是会经过微信服务器的处理转发，类似于一个“中间人”的角色。

  • 优点在于：
    1. 非阻塞：调用者无需等待被调用者完成，可以继续执行其他任务，提高系统并发性和效率。
    2. 资源优化：通过减少资源占用和等待时间，优化了系统的资源利用。
    3. 故障隔离：不会影响到整条业务链
    4. 流量控制：这其实是消息队列的作用，你可以控制有多少的请求通过队列，实现一个 ”削峰填谷“ 的效果
  •  缺点在于 
    1. 消息丢失风险：当无人订阅你的消息，中间人不知道你的消息要发送给谁，消息就丢失了。
    2. 时效性差：你的主动权交出去了，别的服务什么时候能完成你无法得知。
    3. 增加风险：如果微信服务器崩了，你的消息就全部丢失了。业务安全指望Broker可靠性

2.2 消息队列产品对比

市面上常见的消息队列产品有RabbitMQ、RocketMQ、kafka、ActiveMQ等。其中前两者是我推荐一定多去看看的，kafka如果有兴趣也可以去了解了解。SpringCloud默认集成的是RabbitMQ，因此本文也会以RabbitMQ为例，讲解消息队列如何在项目中进行使用。

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

总结：

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

【消息队列的业务场景】

  • 解耦：消息队列允许你的服务彼此独立，只需要知道如何与队列进行交互，而无需了解或维护其他服务的详细信息。
  • 异步通信：消息队列提供异步处理机制，允许用户把一个耗时任务放到队列中，然后立即返回，增加系统的吞吐量。
  • 缓冲：消息队列能够起到缓冲的作用，当处理速度不匹配时，可以暂存那些还未处理的消息。
  • 可靠性：在处理过程中，如果一个处理步骤失败，消息队列可以要求重新处理该消息，而不是丢失它。

2.3 常见的消息队列模式 

常见消息队列模式有：简单队列模式、工作队列模式、发布/订阅模式、路由模式、主题模式。（其实不需要分得那么细，简单分两种都能概括）

2.3.1 简单队列模式

一个队列、一个生产者、一个消费者组成的工作模式。生产者负责向队列发送消息，消费者则直接消费消息即可。很简单，不常用。

2.3.2 工作队列模式

工作队列模式就是在简单工作模式的基础上实现了单生产者对多消费者的情况。生产者往某个队列里面发送消息，一个队列可以存储多个生产者的消息，一个队列也可以有多个消费者， 但是消费者之间是竞争关系，即每条消息只能被一个消费者消费。

2.3.3 发布/订阅模式

发布/订阅模型就是在工作队列模式的基础上添加了一个交换机。使得原先一条消息不能被多个消费者共享的问题被解决。打破了部分消费者之间的竞争关系。

具体的，给每一个需要消费消息的消费者单独一个消息队列，然后交由交换机管理。生产者向交换机发送消息，交换机把信息复制发送给所管理的每一个消息队列。

2.3.4 路由模式

路由模式就是在发布/订阅模式的基础上细化了交换机的匹配规则。实现了根据不同的路由键来接受不同的信息。

具体的，我们需要给交换机到消息队列路径上添加特定的标识。当生产者发送带标识的消息后，交换机负责将该消息发送到带有该标识的消息队列上。

2.3.5 主题模式

主题模式则是在路由模式的基础上对通配路径的匹配规则进行了优化。实现了一条路径匹配一系列相同主题的消息。因此成为主题模式。

具体的，你可以利用通配符 #  或 * 来分别代表一些特定含义。

#代指0个或多个单词
*代指一个单词

三、RabbitMQ使用入门

3.1 RabbitMQ的线上部署

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

RabbitMQ: One broker to queue them all | RabbitMQicon-default.png?t=O83Ahttps://www.rabbitmq.com/

我们首先学习一下如何使用Docker部署RabbitMQ。

步骤特别简单，直接拉取Docker仓库中的RabbitMQ镜像，指定网络段即可。我前面说过了，我的项目基于黑马商城微服务，因此我的网络段设置为 zhicong2。

docker run \
 -e RABBITMQ_DEFAULT_USER=admin\
 -e RABBITMQ_DEFAULT_PASS=admin\
 -v mq-plugins:/plugins \
 --name mq \
 --hostname mq \
 -p 15672:15672 \
 -p 5672:5672 \
 --network zhicong2\
 -d \
 rabbitmq:3.8-management
  • -e 指定的是RabbitMQ控制台登录的账号密码
  • -v 数据挂载，RabbitMQ后期需要使用的插件的挂载位置
  • --name 容器名
  • --hostname 主机名
  • 15672 是控制台的访问端口
  • 6572 是RabbitMQ的服务端口
  • -- network 指定网络段
  • -d 后台守护进程启动
  • 镜像版本 

【查看容器是否正常运行】 

【查看RabbitMQ控制台】

http://192.168.186.140:15672/

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

通过RabbitMQ的控制台，我们能知道一些RabbitMQ的名词，而整体的架构是怎么样的，我觉得下面这张图比较详细。

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

3.3 简单消息收发实验

【实验说明】本实验只需要在控制台完成，掌握RabbitMQ的基本使用，无需额外的代码编写。

【实验内容】在RabbitMQ控制台中完成系列操作：

  • 新建队列hello.queue1和hello.queue2
  • 向默认的amp.fanout交换机发送一条消息
  • 查看消息是否到达hello.queue1和hello.queue2
  • 总结规律

【新建队列】

在Queues中新建队列hello.queue1和hello.queue2

【尝试向交换机发送消息】

点击进入amq.fanout交换机

发送信息，显示消息丢失了

队列没有收到消息

【交换机绑定队列】

【再次尝试发送消息】

成功收到消息

【实验总结】：

  • 生产者发送给交换机的消息，如果没有消费者消费，那么消息会丢失。
  • 交换机只负责传递消息，并没有存储消息的能力。
  • 交换机只会将消息转发到与之绑定的消息队列中。

3.4 虚拟主机实现数据隔离实验

【实验说明】仍然是利用RabbitMQ控制台完成本次实验。

【实验意义】为什么要有不同用户间的数据隔离？

RabbitMQ性能卓越，出于成本考虑，通常只会搭建一套MQ集群支持多个项目一起共用RabbitMQ服务。但是为了避免项目之间互相干扰。RabbitMQ提供了一套虚拟主机virtual host实现数据隔离的方案。具体的，一般分为两步走：

  • 给每个项目创建独立的运维账号，将管理权限分离。

  • 给每个项目创建不同的virtual host，将每个项目的数据隔离。

【实验内容】在RabbitMQ的控制台完成下列操作

  • 新建一个用户hmall
  • 为hmall用户创建一个virtual host
  • 测试不同virtual host之间的数据隔离现象
  • 实验总结

【用户管理说明】

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

  • Name：admin，也就是用户名

  • Tags：administrator，说明admin用户是超级管理员，拥有所有权限

  • Can access virtual host： /，可以访问的virtual host，这里的/是默认的virtual host

【新增hmall用户】添加hmall用户，顺手给个管理员身份

当前创建的用户还没有绑定的虚拟主机

【登录hmall，绑定虚拟主机】

【切换虚拟主机环境】将虚拟主机环境切换到/hmall，观察先前创建的队列

【实验总结】

  • RabbitMQ新建用户支持身份权限管理
  • RabbitMQ支持数据隔离，使用不同的虚拟主机有不同的数据
  • 如何给用户添加虚拟主机权限，只需要切换登录用户，在该用户下创建虚拟主机即可

四、Java客户端使用RabbitMQ

前面学习了如何在RabbitMQ控制台中进行消息收发，但是我们一般不这么用。一来太麻烦，二来也不好适用业务变化。

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

本节中，我们会在Java项目中选取特定的业务场景，修改业务逻辑，利用RabbitMQ将同步调用改为异步调用。为了完成这一优化，我们必须先学习Java如何集成RabbitMQ。

4.1 SpringAMQP

RabbitMQ采用的是AMQP协议，也提供了Java语言的AMQP规范。但是这部分的编码相对较为复杂。因此，Spring官方在RabbitMQ基础上提出了一套消息收发的模板工具，实现了Springboot对其的自动装配。降低了Java客户端使用RabbitMQ编码难度。

SpringAmqp的官方地址：

Spring AMQPicon-default.png?t=O83Ahttps://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能：

  • 自动声明队列、交换机及其绑定关系

  • 基于注解的监听器模式，异步接收消息

  • 封装了RabbitTemplate工具，用于发送消息

接下来我们就来学习如何使用SpringAMQP实现RabbitMQ的消息收发吧！

4.2 SpringAMQP快速入门实验

【实验说明】新建一个项目mq-demo，包含一个父工程和一个生产者、一个消费者子模块的SpringCloud项目。资料在黑马商城微服务中有提供。

【相关依赖】

<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>

【子模块引用父模块】

    <parent>
        <artifactId>mq-demo</artifactId>
        <groupId>cn.itcast.demo</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>

【消息队列模型说明】作为基础入门实验，我们采用简单工作队列模型即可

【控制台新增队列用于测试】

4.2.1 消息发送配置

这一步是指配置生产者（服务调用者）的配置文件，实现将消息发送到RabbitMQ指定的队列中。

【配置RabbitMQ连接地址】在publisher服务的application.yml中添加配置：

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

【编写测试类，构建消息】利用RabbitTemplate实现消息发送：

package com.itheima.publisher;

import org.junit.jupiter.api.Test;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.boot.test.context.SpringBootTest;

import javax.annotation.Resource;

@SpringBootTest
class SimpleQueueTest {

    @Resource
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {

        // 1. 队列名称
        String queue = "simple.queue";
        //2. 消息内容
        String message = "hello simple queue";
        // 3. 发送消息
        rabbitTemplate.convertAndSend(queue,message);
    }
}

【控制台查看消息】成功收到信息了

4.2.2 消息接收配置

【配置MQ地址】配置MQ地址，在consumer服务的application.yml中添加配置：

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

【新建监听消息类】在consumer服务的com.itheima.consumer.listener包中新建一个类SpringRabbitListener，代码如下：

1. 注册成组件 2. 添加RabbitListener注解，监听队列消息

package com.itheima.consumer.listener;


import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void receiveMessage(String message){
        System.out.println("收到消息："+message);
    }
}

4.2.3 效果测试

 【启动消费模块，持续接收消息】

4.3 RabbitMQ工作队列模式实验

这个模式相当于我们前面介绍的工作队列模式，实在简单队列的基础上拓展了一对多的关系。

这种模式的好处在于：当生产者的生产效率十分之高时，我们可以创建更多的消费者及时消费信息。

【工作队列准备】在控制台创建一个新的队列，命名为work.queue：

4.3.1 消息发送配置

【编写测试类】

    @Test void testWorkQueue() {
        // 1. 队列名称
        String queue = "work.queue";
        //2. 消息内容
        String message = "wzc learn rabbitmq";
        // 3. 发送消息
        for (int i = 0; i < 50; i++) {
            message = message + " day" + i;
            rabbitTemplate.convertAndSend(queue,message);
        }
    }

4.3.2 消息接收配置

【添加多个消费者】为了模拟多个消费者绑定同一个队列，我们在consumer.SpringRabbitListener中添加2个新的方法：

    @RabbitListener(queues = "work.queue")
    public void workQueueConsumer01(String message) {
        System.out.println("消费者1收到消息："+message);
    }

    @RabbitListener(queues = "work.queue")
    public void workQueueConsumer02(String message) {
        System.err.println("消费者2收到消息："+message);
    }

4.3.3 效果测试

【启动消费者，持续接收消息，查看消息消耗情况】

4.3.4（进阶）消费者消费效率差异实验

上面实验测试的消费者的消费效率是一样的，现在我们对代码进行修改，使得消费效率变得不同。

    @RabbitListener(queues = "work.queue")
    public void workQueueConsumer01(String message)   throws InterruptedException  {
        System.out.println("消费者1收到消息："+message + " {" + LocalTime.now() + "}");
        Thread.sleep(250);
    }

    @RabbitListener(queues = "work.queue")
    public void workQueueConsumer02(String message)  throws InterruptedException   {
        System.err.println("消费者2收到消息："+message + " {" + LocalTime.now() + "}");
        Thread.sleep(25);
    }

两者相差着将近十倍的效率，但是测试结果如下：

可以看到消费者1和消费者2还是每人消费了25条消息：

  • 消费者2很快完成了自己的25条消息

  • 消费者1却在缓慢的处理自己的25条消息。

  • 这样显然是不合理的

4.3.5（优化）多劳多得分配

在spring中有一个简单的配置，可以解决这个问题。我们修改consumer服务的application.yml文件，添加配置：

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

因为限制了每人每次限拿一条消息，处理消息快的就可以继续进行处理

4.3.6 优化效果测试

4.3.7 总结

  • 消费者之间是竞争关系，每个消息只能被处理一次
  • 可以通过prefetch配置实现多劳多得

4.4 RabbitMQ交换机

这一部分在前面消息队列模式介绍时也简单讲解过。简单来说，通过引入交换机，生产者不用再去指定要投递消息到哪一个具体的消息队列，而是同一交给交换机进行消息的分发。

这也是发布/订阅模式 与 工作队列模式的区分点。

角色分析：

  • Publisher：生产者，不再发送消息到队列中，而是发给交换机

  • Exchange：交换机，一方面，接收生产者发送的消息。另一方面，知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

  • Queue：消息队列也与以前一样，接收消息、缓存消息。不过队列一定要与交换机绑定。

  • Consumer：消费者，与以前一样，订阅队列，没有变化

4.5 RMQ_Fanout交换机实验

Fanout在MQ中叫广播，它的作用是像广播一样将消息无差别的分发到每一个订阅它的人。这和我们前面聊的发布/订阅模式是一样的。

接下来通过实验体会Fanout交换机的使用方式和使用场景。

4.5.1 声明队列和Fanout交换机

创建两个队列 fanout1.queue、fanout2.queue

创建hmall.fanout交换机

将fanout1.queue 和 fanout2.queue添加到hmall.fanout交换机中，进行绑定

4.5.2 消息发送配置

编写测试类，发送广播消息

    @Test
    public void testFanoutExchange() {
        // 1. 交换机名称
        String exchange = "hmall.fanout";
        //2. 消息内容
        String message = "wzc learn Fanout";
        // 3. 发送消息
        rabbitTemplate.convertAndSend(exchange,"",message);
    }

注意：这里的convertAndSend方法需要有三个参数，第二个参数是路由路径，广播交换机配置为空即可，后续会展开讲。

4.5.3 消息接收配置

添加两个绑定队列的消费者

    @RabbitListener(queues = "fanout1.queue")
    public void FanoutQueueConsumer1(String message)  {
        System.out.println("消费者1收到消息："+message);
    }

    @RabbitListener(queues = "fanout2.queue")
    public void FanoutQueueConsumer2(String message)  {
        System.out.println("消费者2收到消息："+message);
    }

4.5.4 测试广播模式结果

4.5.5 总结

交换机的作用是什么？

  • 接收publisher发送的消息

  • 将消息按照规则路由到与之绑定的队列

  • 不能缓存消息，路由失败，消息丢失

  • FanoutExchange的会将消息路由到每个绑定的队列

4.6 RMQ_Direct交换机实验

在Fanout模式中，一条消息，会被所有订阅的队列无差别消费。这时因为交换机只做转发不做校验。而Direct模式就是在Fanout模式基础上对交换机进行了优化，实现了控制消息转发的路由。在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。

Direct交换机就相当于我们前面讲的路由模式。

在Direct模型下：

  • 队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）

  • 消息的发送方在 向 Exchange发送消息时，也必须指定消息的 RoutingKey。

  • Exchange不再把消息交给每一个绑定的队列，而是根据消息的Routing Key进行判断，只有队列的Routingkey与消息的 Routing key完全一致，才会接收到消息

接下来让我们实验学习如何使用Direct交换机吧！

4.6.1 声明队列和Direct交换机

创建两个队列direct1.queue 和 direct2.queue

创建direct类型的交换机hmall.direct

绑定队列到交换机并添加路由路径规则

4.6.2 消息发送配置

编写测试类，发送路由信息【依次测试三条信息】

    @Test
    public void testDirectExchange() {
        //1. 交换机名称
        String exchange = "hmall.direct";
        //2. 消息内容
        String message = "flower color";
        //3. 发送消息
        rabbitTemplate.convertAndSend(exchange,"red",message);
        //rabbitTemplate.convertAndSend(exchange,"blue",message);
        //rabbitTemplate.convertAndSend(exchange,"yellow",message);
    }

4.6.3 消息接收配置

    @RabbitListener(queues = "direct1.queue")
    public void DirectQueueConsumer1(String message)  {
        System.out.println("消费者1收到消息："+message);
        System.out.println("-----------------------------------------------------");
    }

    @RabbitListener(queues = "direct2.queue")
    public void DirectQueueConsumer2(String message)  {
        System.out.println("消费者2收到消息："+message);
        System.out.println("-----------------------------------------------------");
    }

4.6.4 测试路由模式结果

发送消息，指定路由路径为 “red”的队列

发送消息，指定路由路径为 “blue” 的队列

发送消息，指定路由路径为 “yellow” 的队列

4.6.5 总结

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

  • Fanout交换机将消息路由给每一个与之绑定的队列

  • Direct交换机根据RoutingKey判断路由给哪个队列

  • 如果多个队列具有相同的RoutingKey，则与Fanout功能类似。

4.7 RMQ_Topic交换机实验

Topic类型其实和Direct类似，只不过更加全面。它允许在交换机路由规则配置上使用通配符。也就是我们前面讲的主题模式。

通配符规则：

  • #：匹配一个或多个词

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

接下来让我们实验练习如何使用Topic交换机吧

4.7.1 声明队列和Topic交换机

创建队列、交换机，绑定队列，配置通配符路由规则

4.7.2 消息发送配置

@Test
    public void testTopicExchange() {
        // 交换机名称
        String exchange = "hmall.topic";
        // 消息内容
        String message1 = "中国新闻：中国成功发射嫦娥五号！";
        String message2 = "中国天气：北京今天最高温度30度，最低温度20度！";
        String message3 = "A国新闻：特离普当选总统";
        // 发送消息
        rabbitTemplate.convertAndSend(exchange,"china.news",message1);
//        rabbitTemplate.convertAndSend(exchange,"china.weather",message2);
//        rabbitTemplate.convertAndSend(exchange,"A国.news",message3);
    }

4.7.3 消息接收配置

    @RabbitListener(queues = "topic1.queue")
    public void TopicQueueConsumer1(String message)  {
        System.out.println("消费者1收到消息："+message);
    }
    
    @RabbitListener(queues = "topic2.queue")
    public void TopicQueueConsumer2(String message)  {
        System.out.println("消费者2收到消息："+message);
    }

4.7.4 测试主题模式结果

测试 china.new:

测试 china.weather

测试 A国.news

4.7.5 总结

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

  • Topic交换机接收的消息RoutingKey必须是多个单词，以 . 分割

  • Topic交换机与队列绑定时的bindingKey可以指定通配符

  • #：代表0个或多个词

  • *：代表1个词

4.8 基于Java Bean创建队列与交换机

发现没有，前面学习RabbitMQ的实验中，第一步都是跑到RabbitMQ控制台去创建队列和交换机。这个过程很容易就出错了，会导致java找不到队列和交换机。既然java客户端都能连接到RabbitMQ服务了，那我们能不能直接在Java项目直接创建队列和交换机呢？答案肯定是能，通过这一节，让我们学习如何在Java中创建队列和交换机吧！

4.8.1 SpringAMQP提供的创建API

创建队列的API ：Queue类

继承AbstractDeclarable、实现Cloneable

创建交换机的API：Exchange接口

子类可以细化到不同类型交换机的创建过程

简化创建过程和绑定队列接口： ExchangeBuilder

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

4.8.2 创建Fanout交换实验

在consumer.config中创建一个类，用于声明队列和交换机

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 {

    /**
     * 声明交换机
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange("hmall.fanout.java");
    }

    /**
     * 声明队列1
     */
    @Bean
    public Queue fanoutQueue1() {
        return new Queue("fanout.queue1.java");
    }

    /**
     * 声明队列2
     */
    @Bean
    public Queue fanoutQueue2() {
        return new Queue("fanout.queue2.java");
    }

    /**
     * 确定绑定关系
     */
    @Bean
    public Binding bingQueue1() {
        return BindingBuilder.bind(fanoutQueue1()).to(fanoutExchange());
    }
    
    @Bean
    public Binding bingQueue2() {
        return BindingBuilder.bind(fanoutQueue2()).to(fanoutExchange());
    }

}

 

4.8.3 创建direct交换机实验

package com.itheima.consumer.config;

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 DirectConfig {

    /**
     * 声明交换机
     */
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange("hmall.direct.java");
    }

    /**
     * 声明队列1
     */
    @Bean
    public Queue directQueue1() {
        return new Queue("direct.queue1");
    }

    /**
     * 声明队列2
     */
    @Bean
    public Queue directQueue2() {
        return new Queue("direct.queue2");
    }

/**
     * 绑定队列1到交换机并指定绑定的路由red
     */
    @Bean
    public Binding bindingQueue1AndRed() {
        return BindingBuilder.bind(directQueue1()).to(directExchange()).with("red");
    }

    /**
     * 绑定队列1到交换机并指定绑定的路由blue
     */
    @Bean
    public Binding bindingQueue1AndBlue() {
        return BindingBuilder.bind(directQueue1()).to(directExchange()).with("blue");
    }

    /**
     * 绑定队列2到交换机并指定绑定的路由yellow
     */
    @Bean
    public Binding bindingQueue2AndYellow() {
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("yellow");
    }

    /**
     * 绑定队列2到交换机并指定绑定的路由blue
     */
    @Bean
    public Binding bindingQueue2AndBlue() {
        return BindingBuilder.bind(directQueue2()).to(directExchange()).with("blue");
    }
    

}

4.8.3 总结

剩下的就不演示了，相信你也发现问题了。这种基于JavaBean创建队列和交换机的方式并不是那么好用。特别是对于需要绑定多个路由的队列来说，我们需要写很多个方法。因此我们更常用的方式是基于注解配置。

4.9 基于注解声明创建队列于交换机

基于@Bean的方式声明队列和交换机比较麻烦，Spring还提供了基于注解方式来声明。基于注解创建的话，不需要额外编写Config类了，直接在 原先@RabbitListener上添加参数即可！

4.9.1 基于注解创建Direct交换机实验

把先前写的Config类中的@Configuration注解注掉，再将RMQ控制台中创建出来的队列交换机删除。

然后我们在监听消息的方法注解上添加相关参数

/**
     * 基于注解的方式声明Direct交换机和队列 
     */
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue1.zhujie"),
            exchange = @Exchange(name = "direct.exchange.zhujie", type = ExchangeTypes.DIRECT),
            key = {"red","blue"}
    ))
    public void listenDirectQueue1(String message) {
        System.out.println("消费者1收到direct.queue1.zhujie的消息："+message);
    }

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "direct.queue2.zhujie"),
            exchange = @Exchange(name = "direct.exchange.zhujie", type = ExchangeTypes.DIRECT),
            key = {"red","yellow"}
    ))
    public void listenDirectQueue2() {
        System.out.println("消费者2收到direct.queue2.zhujie的消息");
    }

4.10 消息转换器使用

研究一下消息队列的消息是以什么样的形式去传递的。我们进入rabbitTemplate.convertAndSend方法：

在数据传输时，它会把你发送的消息序列化为字节发送给MQ，接收消息的时候，还会把字节反序列化为Java对象。但是它转序列化的方式默认是采用JDK序列化的。这种序列化存在一些弊端如：数据体积过大、可读性极差并且可能伴随安全漏洞。

4.10.1 默认JDK转换器实验

【创建测试队列】

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 {

    /**
     * 创建测试队列
     * @return
     */
    @Bean
    public Queue testObjectQueue() {
        return new Queue("test_object.queue");
    }
}

这里不需要创建消息消费者，我们只需要看信息内容的格式，不涉及消费信息。

【发送消息测试】新增一个消息发送方法，发送一个非String对象

    @Test
    public void testDefaultSendMapMessage()  throws InterruptedException{
        // 准备一个Map消息
        Map<String, Object> msg = new HashMap<>();
        msg.put("name", "wzc");
        msg.put("age", 18);
        
        // 发送消息
        rabbitTemplate.convertAndSend("test_object.queue", msg);
    }

4.10.2 使用JSON格式的消息转换器

最常用的消息格式就是JSON了，体积适中可读性好。而且Java对其的支持也很丰富。所以我们尝试如何配置JSON的消息转换器。

【引入jackson依赖】在publicsher 和 consumer中都引入Jackson依赖

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

注意：如果项目引入了spirng-boot-start-web起步依赖，就已经包含了Jackson依赖了，无需额外再导入.

【配置消息转换器】在publicsher 和 consumer启动类中都配置消息转换器

@SpringBootApplication
public class ConsumerApplication {
    public static void main(String[] args) {
        SpringApplication.run(ConsumerApplication.class, args);
    }


    /**
     * 配置消息转换器
     * @return
     */
    @Bean
    public MessageConverter messageConverter(){
        //1. 定义消息转换器
        Jackson2JsonMessageConverter jackson2JsonMessageConverter = new Jackson2JsonMessageConverter();

        //2. 配置自动创建消息的Id，用于识别不同的信息，
        // 也可以在业务中基于Id判断是否是重复消息从而做到幂等判断
        jackson2JsonMessageConverter.setCreateMessageIds(true);

        //3. 返回Bean对象
        return jackson2JsonMessageConverter;
    }
}

【重新测试消息】删除旧消息，重新发送一次

4.10.3 消费者接收消息改造

如果生产者发送的是Map类型的消息，那消费者也需要使用Map来接收

    @RabbitListener(queues = "test_object.queue")
    public void listenJSONMessageQueue(Map<String, Object> msg)  {
        System.out.println("收到Json格式的Map类型消息："+msg);
    }

五、黑马商城微服务实战改造

到此为止你对RabbitMQ基本的使用方法应该掌握了。知识点并不多，重点在于了解MQ的几种工作模式以及如何在Java客户端中使用RabbitMQ。接下来我们结合RabbitMQ所学，尝试对黑马商城业务进行改造。

5.1 业务改造实验说明

【实验内容说明】

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

【实验步骤说明】

  1. 定义direct类型的交换机pay.direct
  2. 定义消息队列 trade.pay.success.queue
  3. 将trade.pay.success.queue绑定到pay.direct，指定路由为pay.success
  4. 支付成功后发送消息到pay.direct，并指定消息路由为pay.success，消息内容为订单id
  5. 交易服务监听trade.pay.success.queue队列，接收到消息后更新订单状态为已支付

5.2 配置RabbitMQ消息队列

无论是生产者（服务提供者），还是消费者（服务调用者）。都需要配置MQ依赖以及MQ的地址。

【配置MQ依赖】给pay-service 和 trade-service配置消息发送依赖

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

【配置MQ地址】给pay-service 和 trade-service配置文件添加rabbitmq地址信息

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

5.3 消息接收配置

在消息的接收方——trade-service中定义一个监听类，用于监听消息队列信息

package com.hmall.trade.listener;


import com.hmall.trade.service.IOrderService;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

@Component
public class PayStatusListener {

    @Resource
    private IOrderService orderService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "trade.pay.success.queue",durable = "true"),
            exchange = @Exchange(name = "pay.direct"),
            key = "pay.success"
    ))
    public void listenPaySuccess(Long orderId) {
        // 执行相关逻辑
        // 通知订单服务，订单支付成功
        orderService.markOrderPaySuccess(orderId);
    }
}

5.4 消息发送配置

修改服务提供者pay-service的业务逻辑，向消息队列发送支付状态信息，修改pay-service服务下的com.hmall.pay.service.impl.PayOrderServiceImpl类中的tryPayOrderByBalance方法：

    @Override
    @Transactional
    public void tryPayOrderByBalance(PayOrderFormDTO payOrderFormDTO) {
        // 1.查询支付单
        PayOrder po = getById(payOrderFormDTO.getId());
        // 2.判断状态
        if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){
            // 订单不是未支付，状态异常
            throw new BizIllegalException("交易已支付或关闭！");
        }
        // 3.尝试扣减余额
        userClient.deductMoney(payOrderFormDTO.getPw(), po.getAmount());
        // 4.修改支付单状态
        boolean success = markPayOrderSuccess(payOrderFormDTO.getId(), LocalDateTime.now());
        if (!success) {
            throw new BizIllegalException("交易已支付或关闭！");
        }
        // 5.修改订单状态（向RabbitMQ发送信息）
//        tradeClient.markOrderPaySuccess(po.getBizOrderNo());
        try{
            rabbitTemplate.convertAndSend("pay.direct","pay.success", po.getBizOrderNo());

        }catch (Exception e) {
           log.error("支付成功后修改订单状态失败", e);
        }
    }

5.5 启动项目，完成测试

队列成功创建

输入支付密码123后，跳转支付成功证明消息成功发送。

六、RabbitMQ相关知识追问巩固

  1. 谈谈同步调用和异步调用各自的特点和区别。

  2. 对比消息队列实现的业务和直接调用有何区别？

  3. 请你谈谈什么是消息队列？消息队列的业务场景有哪些？

  4. 列举市面上常见的几种消息队列？你使用过其中的几种？各自有何特点？

  5. 【考察选型】给你一个具体的业务场景，你应该如何选择使用的消息队列？考虑哪些维度？

  6. 介绍一下常见的几种消息队列的工作模式？它们都是为了解决什么问题出现的？

  7. 解释发布/订阅模式为了解决什么问题？工作过程是什么？

  8. 解释主题模式是为了解决什么问题？

  9. RabbitMQ的整体架构是怎么样的？它是如何实现数据隔离的？

  10. 介绍一下SpringAMQP的基本功能有哪些？它是如何帮助我们简化RabbitMQ操作的？

  11. 介绍一下Java项目配置RabbitMQ需要几步骤？

  12. 请你介绍使用JavaBean创建交换机和绑定队列过程中，需要用到哪些类或接口的支持？

  13. 请你介绍基于注解实现自动创建交换机和绑定队列，实现消息的发送与接收，具体需要如何操作？请你描述一下整个过程。

  14. RabbitTemplate提供的默认消息转换器是什么？有什么不好的地方？

  15. 如何在项目中自定义消息转换器？请你描述一下关键步骤。

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

相关文章：

  • 【Excel】ToRow超级查找函数
  • 基于RK3568J多网口电力可信物联网关解决方案
  • 图论-代码随想录刷题记录[JAVA]
  • 时序论文20｜ICLR20 可解释时间序列预测N-BEATS
  • 在JPA和EJB中用乐观锁解决并发问题
  • Ubuntu配置阿里云docker apt源
  • git 提交报错 Error updating changes: bad signature 0x00000000index file corrupt
  • STM32编程遇到的问题随笔【一】
  • 深入探索React合成事件（SyntheticEvent）：跨浏览器的事件处理利器
  • Windowos系统部署mino
  • Spring Boot应用开发实战：构建高效、可维护的Web应用
  • linux startup.sh shutdown.sh （kkFileView）
  • HBase理论_HBase架构组件介绍
  • IEC61850服务分类说明
  • SpringBoot 3.3.5 试用CRaC，启动速度提升3到10倍
  • 【学习率】
  • 时间序列分析——移动平均法、指数平滑法、逐步回归法、趋势外推法等（基于Python实现）
  • SpringBoot（二十二）SpringBoot集成smart-doc自动生成文档
  • gorm使用注意事项
  • 轮转数组
  • 旅行照片 4.0 问题三、四
  • 【自用】0-1背包问题与完全背包问题的Java实现
  • 视频横屏转竖屏播放-使用人脸识别+目标跟踪实现
  • [自然语言处理] [AI]深入理解语言与情感分类：从基础到深度学习的进展
  • Unity自动LOD工具AutoLOD Mesh Decimator的使用
  • HarmonyOS开发 API 13发布首个Beta版本，部分已知的问题建议处理方案