0
0

快速入门消息队列MQ、RabbitMQ

一、MQ简介

微服务一旦拆分，必然涉及到服务之间的相互调用，目前我们服务之间调用采用的都是基于OpenFeign的调用。这种调用中，调用者发起请求后需要等待服务提供者执行业务返回结果后，才能继续执行后面的业务。也就是说调用者在调用过程中处于阻塞状态，因此我们称这种调用方式为同步调用，也可以叫同步通****讯。但在很多场景下，我们可能需要采用异步通讯的方式，为什么呢？

我们先来看看什么是同步通讯和异步通讯。如图：

解读：

  • 同步通讯：就如同打视频电话，双方的交互都是实时的。因此同一时刻你只能跟一个人打视频电话。
  • 异步通讯：就如同发微信聊天，双方的交互不是实时的，你不需要立刻给对方回应。因此你可以多线操作，同时跟多人聊天。

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

所以，如果我们的业务需要实时得到服务提供方的响应，则应该选择同步通讯（同步调用）。而如果我们追求更高的效率，并且不需要实时响应，则应该选择异步通讯（异步调用）。

同步调用的方式我们已经学过了，之前的OpenFeign调用就是。但是：

  • 异步调用又该如何实现？
  • 哪些业务适合用异步调用来实现呢？

** 简而言之就是：之前我们学习的内容OpenFeign是属于同步调用，使用于实时响应的场景；而MQ是属于异步调用，适用于高效率且不需要实时响应的场景。**

1.同步调用

之前说过，我们现在基于OpenFeign的调用都属于是同步调用，那么这种方式存在哪些问题呢？

举个例子，我们以余额支付功能为例来分析，首先看下整个流程：

目前我们采用的是基于OpenFeign的同步调用，也就是说业务执行流程是这样的：

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

三个步骤依次执行。

这其中就存在3个问题：

第一，拓展性差

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

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

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

。。。

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

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

第二，性能下降

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

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

第三，级联失败

由于我们是基于OpenFeign调用交易服务、通知服务。当交易服务、通知服务出现故障时，整个事务都会回滚，交易失败。这看起来似乎没有问题，但是如果我们的整个业务链中还包含着其他业务，比如日志记录等不相干的业务，日志记录失败了，也会导致我们整条业务链全部失败回滚，这肯定是不允许的。

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

但是大家思考一下，我们假设用户余额充足，扣款已经成功，此时我们应该确保支付流水单更新为已支付，确保交易成功。毕竟收到手里的钱没道理再退回去吧。

因此，这里不能因为短信通知、更新订单状态失败而回滚整个事务。

综上，同步调用的方式存在下列问题：

  • 拓展性差
  • 性能下降
  • 级联失败

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

2.异步调用

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

  • 消息发送者：投递消息的人，就是原来的调用方
  • 消息Broker：管理、暂存、转发消息，你可以把它理解成微信服务器
  • 消息接收者：接收和处理消息的人，就是原来的服务提供方

在异步调用中，发送者不再直接同步调用接收者的业务接口，而是发送一条消息投递给消息Broker。然后接收者根据自己的需求从消息Broker那里订阅消息。每当发送方发送消息后，接受者都能获取消息并处理。

这样，发送消息的人和接收消息的人就完全解耦了。

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

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

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

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

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

综上，异步调用的优势包括：

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

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

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

3.技术选型

市面上已经有很多种MQ了，我们这里只需要学习RabbitMQ即可，以下是各个MQ的对比。

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

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

虽然说Kafka的吞吐量很高，但是其应用场景在于大数据分析，不适用于我们的日常项目，就类似大炮打蚊子。

二、RabbitMQ

1.安装

安装说明参考以下博客：RabbitMQ安装教程（超详细）-CSDN博客

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

RabbitMQ对应的架构如图：

其中包含几个概念：

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

2.控制台的使用说明

2.1交换机

此时如果没有队列绑定在交换机上，那发送出去的消息就会丢失

2.2队列

时，我们再次向

  1. amq.fanout

交换机发送一条消息。会发现消息依然没有到达队列！！

怎么回事呢？（注意，交换机只负责转发消息，不存储，就类似外卖小哥一样，只负责送外卖，不不帮你保管，外卖柜就类似队列）

发送到交换机的消息，只会路由到与其绑定的队列，因此仅仅创建队列是不够的，我们还需要将其与交换机绑定。

2.3绑定关系

点击

  1. Exchanges

选项卡，点击

  1. amq.fanout

交换机，进入交换机详情页，然后点击

  1. Bindings

菜单，在表单中填写要绑定的队列名称：

再次回到exchange页面，找到刚刚绑定的

  1. amq.fanout

，点击进入详情页，再次发送一条消息：

回到

  1. Queues

页面，可以发现

  1. hello.queue

中已经有一条消息了：

点击队列名称，进入详情页，查看队列详情，这次我们点击get message：

可以看到消息到达队列了：

这个时候如果有消费者监听了MQ的

  1. hello.queue1

或

  1. hello.queue2

队列，自然就能接收到消息了。

3.AMQP

上述我们都是在控制台中去收发消息，操作起来也十分方便，但是我们以后再开发业务的时候肯定不会在上面操作的，都是基于编程实现的。

由于

  1. RabbitMQ

采用了AMQP协议，因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息，都可以与

  1. RabbitMQ

交互。并且

  1. RabbitMQ

官方也提供了各种不同语言的客户端。

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

以下是AMQP的官网：

Spring AMQPLevel up your Java code and explore what Spring can do for you.https://spring.io/projects/spring-amqp

SpringAMQP提供了三个功能：

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

3.1快速入门

在之前的案例中，我们都是经过交换机发送消息到队列，不过有时候为了测试方便，我们也可以直接向队列发送消息，跳过交换机。

在入门案例中，我们就演示这样的简单模型，如图：

也就是只需要发送和接收即可

第一步就是引依赖：

  1.  <dependency>
  2.             <groupId>org.springframework.boot</groupId>
  3.             <artifactId>spring-boot-starter-amqp</artifactId>
  4. </dependency>

第二步：写配置（在

  1. publisher

服务的

  1. application.yml

中添加配置：）

  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.1 # 你的虚拟机IP
  4.     port: 5672 # 端口
  5.     virtual-host: /scau# 虚拟主机
  6.     username: zzh# 用户名
  7.     password: 123 # 密码

然后在

  1. publisher

服务中编写测试类

  1. SpringAmqpTest

，并利用

  1. RabbitTemplate

实现消息发送：

  1. package com.scau.publisher.amqp;
  3. import org.junit.jupiter.api.Test;
  4. import org.springframework.amqp.rabbit.core.RabbitTemplate;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.boot.test.context.SpringBootTest;
  8. @SpringBootTest
  9. public class SpringAmqpTest {
  11.     @Autowired
  12.     private RabbitTemplate rabbitTemplate;
  14.     @Test
  15.     public void testSimpleQueue() {
  16.         // 队列名称
  17.         String queueName = "simple.queue";
  18.         // 消息
  19.         String message = "hello, spring amqp!";
  20.         // 发送消息
  21.         rabbitTemplate.convertAndSend(queueName, message);
  22.     }
  23. }

主要用到的api就是：

  1. rabbitTemplate.convertAndSend(queueName, message);

接下来，我们再来实现消息接收。

还是一样，对接收者Service，也是引依赖，写配置，这里不过多赘述

然后在

  1. consumer

服务的中新建一个类

  1. SpringRabbitListener

，代码如下：

  1. package com.scau.consumer.listener;
  3. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  4. import org.springframework.stereotype.Component;
  6. @Component
  7. public class SpringRabbitListener {
  8.     // 利用RabbitListener来声明要监听的队列信息
  9.     // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
  10.     // 可以看到方法体中接收的就是消息体的内容
  11.     @RabbitListener(queues = "simple.queue")
  12.     public void listenSimpleQueueMessage(String msg) throws InterruptedException {
  13.         System.out.println("spring 消费者接收到消息：【" + msg + "】");
  14.     }
  15. }

主要是用到@RabbitListener这个注解来接受指定队列的消息

3.2WorkQueues模型

我们在快速入门中所展示的是一对一的队列模型，但如果生产者发送的消息过多，只有一个消费者的话消费不过来，就容易导致消息堆积，这时候就可以用到WorkQueues模型，多个消费者消费消息。

我们在生产者模拟大量消息堆积现象。

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

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

要模拟多个消费者绑定同一个队列，我们在consumer服务的SpringRabbitListener中添加2个新的方法：

  1. @RabbitListener(queues = "work.queue")
  2. public void listenWorkQueue1(String msg) throws InterruptedException {
  3.     System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
  4.     Thread.sleep(20);
  5. }
  7. @RabbitListener(queues = "work.queue")
  8. public void listenWorkQueue2(String msg) throws InterruptedException {
  9.     System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
  10.     Thread.sleep(200);
  11. }

注意到这两消费者，都设置了

  1. Thead.sleep

，模拟任务耗时：

  • 消费者1 sleep了20毫秒，相当于每秒钟处理50个消息
  • 消费者2 sleep了200毫秒，相当于每秒处理5个消息

测试结果如下：

  1. 消费者1接收到消息：【hello, message_0】21:06:00.869555300
  2. 消费者2........接收到消息：【hello, message_1】21:06:00.884518
  3. 消费者1接收到消息：【hello, message_2】21:06:00.907454400
  4. 消费者1接收到消息：【hello, message_4】21:06:00.953332100
  5. 消费者1接收到消息：【hello, message_6】21:06:00.997867300
  6. 消费者1接收到消息：【hello, message_8】21:06:01.042178700
  7. 消费者2........接收到消息：【hello, message_3】21:06:01.086478800
  8. 消费者1接收到消息：【hello, message_10】21:06:01.087476600
  9. 消费者1接收到消息：【hello, message_12】21:06:01.132578300
  10. 消费者1接收到消息：【hello, message_14】21:06:01.175851200
  11. 消费者1接收到消息：【hello, message_16】21:06:01.218533400
  12. 消费者1接收到消息：【hello, message_18】21:06:01.261322900
  13. 消费者2........接收到消息：【hello, message_5】21:06:01.287003700
  14. 消费者1接收到消息：【hello, message_20】21:06:01.304412400
  15. 消费者1接收到消息：【hello, message_22】21:06:01.349950100
  16. 消费者1接收到消息：【hello, message_24】21:06:01.394533900
  17. 消费者1接收到消息：【hello, message_26】21:06:01.439876500
  18. 消费者1接收到消息：【hello, message_28】21:06:01.482937800
  19. 消费者2........接收到消息：【hello, message_7】21:06:01.488977100
  20. 消费者1接收到消息：【hello, message_30】21:06:01.526409300
  21. 消费者1接收到消息：【hello, message_32】21:06:01.572148
  22. 消费者1接收到消息：【hello, message_34】21:06:01.618264800
  23. 消费者1接收到消息：【hello, message_36】21:06:01.660780600
  24. 消费者2........接收到消息：【hello, message_9】21:06:01.689189300
  25. 消费者1接收到消息：【hello, message_38】21:06:01.705261
  26. 消费者1接收到消息：【hello, message_40】21:06:01.746927300
  27. 消费者1接收到消息：【hello, message_42】21:06:01.789835
  28. 消费者1接收到消息：【hello, message_44】21:06:01.834393100
  29. 消费者1接收到消息：【hello, message_46】21:06:01.875312100
  30. 消费者2........接收到消息：【hello, message_11】21:06:01.889969500
  31. 消费者1接收到消息：【hello, message_48】21:06:01.920702500
  32. 消费者2........接收到消息：【hello, message_13】21:06:02.090725900
  33. 消费者2........接收到消息：【hello, message_15】21:06:02.293060600
  34. 消费者2........接收到消息：【hello, message_17】21:06:02.493748
  35. 消费者2........接收到消息：【hello, message_19】21:06:02.696635100
  36. 消费者2........接收到消息：【hello, message_21】21:06:02.896809700
  37. 消费者2........接收到消息：【hello, message_23】21:06:03.099533400
  38. 消费者2........接收到消息：【hello, message_25】21:06:03.301446400
  39. 消费者2........接收到消息：【hello, message_27】21:06:03.504999100
  40. 消费者2........接收到消息：【hello, message_29】21:06:03.705702500
  41. 消费者2........接收到消息：【hello, message_31】21:06:03.906601200
  42. 消费者2........接收到消息：【hello, message_33】21:06:04.108118500
  43. 消费者2........接收到消息：【hello, message_35】21:06:04.308945400
  44. 消费者2........接收到消息：【hello, message_37】21:06:04.511547700
  45. 消费者2........接收到消息：【hello, message_39】21:06:04.714038400
  46. 消费者2........接收到消息：【hello, message_41】21:06:04.916192700
  47. 消费者2........接收到消息：【hello, message_43】21:06:05.116286400
  48. 消费者2........接收到消息：【hello, message_45】21:06:05.318055100
  49. 消费者2........接收到消息：【hello, message_47】21:06:05.520656400
  50. 消费者2........接收到消息：【hello, message_49】21:06:05.723106700

我们发现由于消费者1收消息的速度比较快，于是消费者1收完了消息（每个消费者均分消息）之后就让消费者2慢慢执行，这效率肯定会低很多。

我们现在希望消费者1消费更多消息（毕竟他比较快）

只需要开启一个配置即可，在消费者的yml文件下配置以下内容即可：

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

再次测试，结果如下：

  1. 消费者1接收到消息：【hello, message_0】21:12:51.659664200
  2. 消费者2........接收到消息：【hello, message_1】21:12:51.680610
  3. 消费者1接收到消息：【hello, message_2】21:12:51.703625
  4. 消费者1接收到消息：【hello, message_3】21:12:51.724330100
  5. 消费者1接收到消息：【hello, message_4】21:12:51.746651100
  6. 消费者1接收到消息：【hello, message_5】21:12:51.768401400
  7. 消费者1接收到消息：【hello, message_6】21:12:51.790511400
  8. 消费者1接收到消息：【hello, message_7】21:12:51.812559800
  9. 消费者1接收到消息：【hello, message_8】21:12:51.834500600
  10. 消费者1接收到消息：【hello, message_9】21:12:51.857438800
  11. 消费者1接收到消息：【hello, message_10】21:12:51.880379600
  12. 消费者2........接收到消息：【hello, message_11】21:12:51.899327100
  13. 消费者1接收到消息：【hello, message_12】21:12:51.922828400
  14. 消费者1接收到消息：【hello, message_13】21:12:51.945617400
  15. 消费者1接收到消息：【hello, message_14】21:12:51.968942500
  16. 消费者1接收到消息：【hello, message_15】21:12:51.992215400
  17. 消费者1接收到消息：【hello, message_16】21:12:52.013325600
  18. 消费者1接收到消息：【hello, message_17】21:12:52.035687100
  19. 消费者1接收到消息：【hello, message_18】21:12:52.058188
  20. 消费者1接收到消息：【hello, message_19】21:12:52.081208400
  21. 消费者2........接收到消息：【hello, message_20】21:12:52.103406200
  22. 消费者1接收到消息：【hello, message_21】21:12:52.123827300
  23. 消费者1接收到消息：【hello, message_22】21:12:52.146165100
  24. 消费者1接收到消息：【hello, message_23】21:12:52.168828300
  25. 消费者1接收到消息：【hello, message_24】21:12:52.191769500
  26. 消费者1接收到消息：【hello, message_25】21:12:52.214839100
  27. 消费者1接收到消息：【hello, message_26】21:12:52.238998700
  28. 消费者1接收到消息：【hello, message_27】21:12:52.259772600
  29. 消费者1接收到消息：【hello, message_28】21:12:52.284131800
  30. 消费者2........接收到消息：【hello, message_29】21:12:52.306190600
  31. 消费者1接收到消息：【hello, message_30】21:12:52.325315800
  32. 消费者1接收到消息：【hello, message_31】21:12:52.347012500
  33. 消费者1接收到消息：【hello, message_32】21:12:52.368508600
  34. 消费者1接收到消息：【hello, message_33】21:12:52.391785100
  35. 消费者1接收到消息：【hello, message_34】21:12:52.416383800
  36. 消费者1接收到消息：【hello, message_35】21:12:52.439019
  37. 消费者1接收到消息：【hello, message_36】21:12:52.461733900
  38. 消费者1接收到消息：【hello, message_37】21:12:52.485990
  39. 消费者1接收到消息：【hello, message_38】21:12:52.509219900
  40. 消费者2........接收到消息：【hello, message_39】21:12:52.523683400
  41. 消费者1接收到消息：【hello, message_40】21:12:52.547412100
  42. 消费者1接收到消息：【hello, message_41】21:12:52.571191800
  43. 消费者1接收到消息：【hello, message_42】21:12:52.593024600
  44. 消费者1接收到消息：【hello, message_43】21:12:52.616731800
  45. 消费者1接收到消息：【hello, message_44】21:12:52.640317
  46. 消费者1接收到消息：【hello, message_45】21:12:52.663111100
  47. 消费者1接收到消息：【hello, message_46】21:12:52.686727
  48. 消费者1接收到消息：【hello, message_47】21:12:52.709266500
  49. 消费者2........接收到消息：【hello, message_48】21:12:52.725884900
  50. 消费者1接收到消息：【hello, message_49】21:12:52.746299900

小结：

Work模型的使用：

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

3.3交换机

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

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

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

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

交换机的类型有四种：

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

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

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

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

消息发送代码如下：

  1. @Test
  2. public void testFanoutExchange() {
  3.     // 交换机名称
  4.     String exchangeName = "test.fanout";
  5.     // 消息
  6.     String message = "hello, everyone!";
  7.     rabbitTemplate.convertAndSend(exchangeName, "", message);
  8. }

我们把消息发到了test.fanout这个交换机上，我们想要实现多个消费者通过不同的队列收到这个消息，那消费者的代码如下：

  1. @RabbitListener(queues = "fanout.queue1")
  2. public void listenFanoutQueue1(String msg) {
  3.     System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
  4. }
  6. @RabbitListener(queues = "fanout.queue2")
  7. public void listenFanoutQueue2(String msg) {
  8.     System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
  9. }

小结

交换机的作用是什么？

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

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

在Direct模型下：

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

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

消费者的代码如下：

  1. @RabbitListener(queues = "direct.queue1")
  2. public void listenDirectQueue1(String msg) {
  3.     System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
  4. }
  6. @RabbitListener(queues = "direct.queue2")
  7. public void listenDirectQueue2(String msg) {
  8.     System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
  9. }

消息发送者的代码如下：

  1. @Test
  2. public void testSendDirectExchange() {
  3.     // 交换机名称
  4.     String exchangeName = "xxx.direct";
  5.     // 消息
  6.     String message = "这是一条红色紧急消息！！！";
  7.     // 发送消息
  8.     rabbitTemplate.convertAndSend(exchangeName, "red", message);
  9. }

这时候消费者1和消费者2都能接收到这条消息

  1. @Test
  2. public void testSendDirectExchange() {
  3.     // 交换机名称
  4.     String exchangeName = "xxx.direct";
  5.     // 消息
  6.     String message = "这是一条蓝色级日常消息";
  7.     // 发送消息
  8.     rabbitTemplate.convertAndSend(exchangeName, "blue", message);
  9. }

这时候就只要消费者1收到了这条消息

小结：

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

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

类型的

  1. Exchange

与

  1. Direct

相比，都是可以根据

  1. RoutingKey

把消息路由到不同的队列。

只不过

  1. Topic

类型

  1. Exchange

可以让队列在绑定

  1. BindingKey

的时候使用通配符！

  1. BindingKey

一般都是有一个或多个单词组成，多个单词之间以

  1. .

分割，例如： 

  1. item.insert

通配符规则：

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

举例：

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

假如此时publisher发送的消息使用的

  1. 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. /**
  2.  * topicExchange
  3.  */
  4. @Test
  5. public void testSendTopicExchange() {
  6.     // 交换机名称
  7.     String exchangeName = "test.topic";
  8.     // 消息
  9.     String message = "这是一条国内新闻...";
  10.     // 发送消息
  11.     rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
  12. }

消息接收方代码如下：

  1. @RabbitListener(queues = "topic.queue1")
  2. public void listenTopicQueue1(String msg){
  3.     System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
  4. }
  6. @RabbitListener(queues = "topic.queue2")
  7. public void listenTopicQueue2(String msg){
  8.     System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
  9. }

同时两个队列的Routing Key如下：

那么显而易见是两个队列都能收到消息

小结：

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

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

3.4 声明交换机和队列

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

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

3.4.1 Fanout示例
  1. package com.scau.consumer.config;
  3. import org.springframework.amqp.core.Binding;
  4. import org.springframework.amqp.core.BindingBuilder;
  5. import org.springframework.amqp.core.FanoutExchange;
  6. import org.springframework.amqp.core.Queue;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  10. @Configuration
  11. public class FanoutConfig {
  12.     /**
  13.      * 声明交换机
  14.      * @return Fanout类型交换机
  15.      */
  16.     @Bean
  17.     public FanoutExchange fanoutExchange(){
  18.         return new FanoutExchange("test.fanout");
  19.     }
  21.     /**
  22.      * 第1个队列
  23.      */
  24.     @Bean
  25.     public Queue fanoutQueue1(){
  26.         return new Queue("fanout.queue1");
  27.     }
  29.     /**
  30.      * 绑定队列和交换机
  31.      */
  32.     @Bean
  33.     public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
  34.         return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
  35.     }
  37.     /**
  38.      * 第2个队列
  39.      */
  40.     @Bean
  41.     public Queue fanoutQueue2(){
  42.         return new Queue("fanout.queue2");
  43.     }
  45.     /**
  46.      * 绑定队列和交换机
  47.      */
  48.     @Bean
  49.     public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
  50.         return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
  51.     }
  52. }
3.4.2 Direct示例
  1. package com.scau.consumer.config;
  3. import org.springframework.amqp.core.*;
  4. import org.springframework.context.annotation.Bean;
  5. import org.springframework.context.annotation.Configuration;
  7. @Configuration
  8. public class DirectConfig {
  10.     /**
  11.      * 声明交换机
  12.      * @return Direct类型交换机
  13.      */
  14.     @Bean
  15.     public DirectExchange directExchange(){
  16.         return ExchangeBuilder.directExchange("test.direct").build();
  17.     }
  19.     /**
  20.      * 第1个队列
  21.      */
  22.     @Bean
  23.     public Queue directQueue1(){
  24.         return new Queue("direct.queue1");
  25.     }
  27.     /**
  28.      * 绑定队列和交换机
  29.      */
  30.     @Bean
  31.     public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
  32.         return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
  33.     }
  34.     /**
  35.      * 绑定队列和交换机
  36.      */
  37.     @Bean
  38.     public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
  39.         return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
  40.     }
  42.     /**
  43.      * 第2个队列
  44.      */
  45.     @Bean
  46.     public Queue directQueue2(){
  47.         return new Queue("direct.queue2");
  48.     }
  50.     /**
  51.      * 绑定队列和交换机
  52.      */
  53.     @Bean
  54.     public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
  55.         return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
  56.     }
  57.     /**
  58.      * 绑定队列和交换机
  59.      */
  60.     @Bean
  61.     public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
  62.         return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
  63.     }
  64. }
3.4.3 基于注解声明

我们从前文可以知道要想声明一个交换机和队列要编写的代码还是很繁琐的，因此AMQP还支持基于注解的形式声明

  1. @RabbitListener(bindings = @QueueBinding(
  2.     value = @Queue(name = "direct.queue1"),
  3.     exchange = @Exchange(name = "test.direct", type = ExchangeTypes.DIRECT),
  4.     key = {"red", "blue"}
  5. ))
  6. public void listenDirectQueue1(String msg){
  7.     System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
  8. }
  10. @RabbitListener(bindings = @QueueBinding(
  11.     value = @Queue(name = "direct.queue2"),
  12.     exchange = @Exchange(name = "test.direct", type = ExchangeTypes.DIRECT),
  13.     key = {"red", "yellow"}
  14. ))
  15. public void listenDirectQueue2(String msg){
  16.     System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
  17. }

topic交换机也是如此：

  1. @RabbitListener(bindings = @QueueBinding(
  2.     value = @Queue(name = "topic.queue1"),
  3.     exchange = @Exchange(name = "test.topic", type = ExchangeTypes.TOPIC),
  4.     key = "china.#"
  5. ))
  6. public void listenTopicQueue1(String msg){
  7.     System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
  8. }
  10. @RabbitListener(bindings = @QueueBinding(
  11.     value = @Queue(name = "topic.queue2"),
  12.     exchange = @Exchange(name = "test.topic", type = ExchangeTypes.TOPIC),
  13.     key = "#.news"
  14. ))
  15. public void listenTopicQueue2(String msg){
  16.     System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
  17. }

三、在实际业务中添加MQ

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

1.添加依赖

  1.   <!--消息发送-->
  2.   <dependency>
  3.       <groupId>org.springframework.boot</groupId>
  4.       <artifactId>spring-boot-starter-amqp</artifactId>
  5.   </dependency>

2.配置MQ地址

  1. spring:
  2.   rabbitmq:
  3.     host: 192.168.1.1 # 你的虚拟机IP
  4.     port: 5672 # 端口
  5.     virtual-host: /test# 虚拟主机
  6.     username: test# 用户名
  7.     password: 123 # 密码

3.接收信息

先创建一个单独的listener类出来

代码如下：

  1. package com.test.trade.listener;
  3. import com.hmall.trade.service.IOrderService;
  4. import lombok.RequiredArgsConstructor;
  5. import org.springframework.amqp.core.ExchangeTypes;
  6. import org.springframework.amqp.rabbit.annotation.Exchange;
  7. import org.springframework.amqp.rabbit.annotation.Queue;
  8. import org.springframework.amqp.rabbit.annotation.QueueBinding;
  9. import org.springframework.amqp.rabbit.annotation.RabbitListener;
  10. import org.springframework.stereotype.Component;
  12. @Component
  13. @RequiredArgsConstructor
  14. public class PayStatusListener {
  16.     private final IOrderService orderService;
  18.     @RabbitListener(bindings = @QueueBinding(
  19.             value = @Queue(name = "trade.pay.success.queue", durable = "true"),
  20.             exchange = @Exchange(name = "pay.topic"),
  21.             key = "pay.success"
  22.     ))
  23.     public void listenPaySuccess(Long orderId){
  24.         orderService.markOrderPaySuccess(orderId);
  25.     }
  26. }

4.发送消息

修改

  1. pay-service

服务下的

  1. com.test.pay.service.impl.PayOrderServiceImpl

类中的

  1. tryPayOrderByBalance

方法：

  1. private final RabbitTemplate rabbitTemplate;
  3. @Override
  4. @Transactional
  5. public void tryPayOrderByBalance(PayOrderDTO payOrderDTO) {
  6.     // 1.查询支付单
  7.     PayOrder po = getById(payOrderDTO.getId());
  8.     // 2.判断状态
  9.     if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){
  10.         // 订单不是未支付，状态异常
  11.         throw new BizIllegalException("交易已支付或关闭！");
  12.     }
  13.     // 3.尝试扣减余额
  14.     userClient.deductMoney(payOrderDTO.getPw(), po.getAmount());
  15.     // 4.修改支付单状态
  16.     boolean success = markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now());
  17.     if (!success) {
  18.         throw new BizIllegalException("交易已支付或关闭！");
  19.     }
  20.     // 5.修改订单状态
  21.     // tradeClient.markOrderPaySuccess(po.getBizOrderNo());
  22.     try {
  23.         rabbitTemplate.convertAndSend("pay.direct", "pay.success", po.getBizOrderNo());
  24.     } catch (Exception e) {
  25.         log.error("支付成功的消息发送失败，支付单id：{}， 交易单id：{}", po.getId(), po.getBizOrderNo(), e);
  26.     }
  27. }
标签： rabbitmq 分布式 java
本文转载自: https://blog.csdn.net/csdn3043663729/article/details/143900975
版权归原作者 泰山小张只吃荷园 所有， 如有侵权，请联系我们删除。
登录后发布评论

“快速入门消息队列MQ、RabbitMQ”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

网络安全法-网络运行安全

使用selenium/drissionpage时如何阻止chrome自动跳转http到https

docker desktop 里部署的Open WebUI 管理员密码忘记了的处理方法

在ubuntu20.04中搭建onsite比赛运行环境

利用开源的低代码表单设计器FcDesigner高效管理和渲染复杂表单结构

Kafka学习笔记

【前端】浏览器输入url到页面呈现发生了什么？