SpringCloud微服务 Day06 - MQ基础：从同步到异步通讯的转变

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

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

 

解读：

• 同步通讯：就如同打视频电话，双方的交互都是实时的。因此同一时刻你只能跟一个人打视频电话。

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

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

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

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

• 异步调用又该如何实现？

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

通过今天的学习你就能明白这些问题了。

1.初识MQ

1.1.同步调用

 

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

举个例子，我们以昨天留给大家作为作业的余额支付功能为例来分析，首先看下整个流程：

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

• 支付服务需要先调用用户服务完成余额扣减

• 然后支付服务自己要更新支付流水单的状态

• 然后支付服务调用交易服务，更新业务订单状态为已支付

三个步骤依次执行。

这其中就存在3个问题：

第一，拓展性差

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

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

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

。。。

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

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

第二，性能下降

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

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

第三，级联失败

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

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

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

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

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

• 拓展性差

• 性能下降

• 级联失败

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

1.2.异步调用

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

• 消息发送者：投递消息的人，就是原来的调用方

• 消息Broker：管理、暂存、转发消息，你可以把它理解成微信服务器

• 消息接收者：接收和处理消息的人，就是原来的服务提供方

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

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

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

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

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

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

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

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

• 耦合度更低

• 性能更好

• 业务拓展性强

• 故障隔离，避免级联失败

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

• 完全依赖于Broker的可靠性、安全性和性能

• 架构复杂，后期维护和调试麻烦

1.3.技术选型

消息Broker，目前常见的实现方案就是消息队列（MessageQueue），简称为MQ.

目比较常见的MQ实现：

• ActiveMQ

• RabbitMQ

• RocketMQ

• Kafka

几种常见MQ的对比：

表格 还在加载中，请等待加载完成后再尝试复制

追求可用性：Kafka、 RocketMQ 、RabbitMQ

追求可靠性：RabbitMQ、RocketMQ

追求吞吐能力：RocketMQ、Kafka

追求消息低延迟：RabbitMQ、Kafka

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

2.RabbitMQ

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

RabbitMQ: One broker to queue them all | RabbitMQ

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

2.1.安装

我们同样基于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

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

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

• 15672：RabbitMQ提供的管理控制台的端口

• 5672：RabbitMQ的消息发送处理接口

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

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

RabbitMQ对应的架构如图：

 

其中包含几个概念：

• publisher：生产者，也就是发送消息的一方

• consumer：消费者，也就是消费消息的一方

• queue：队列，存储消息。生产者投递的消息会暂存在消息队列中，等待消费者处理

• exchange：交换机，负责消息路由。生产者发送的消息由交换机决定投递到哪个队列。

• virtual host：虚拟主机，起到数据隔离的作用。每个虚拟主机相互独立，有各自的exchange、queue

上述这些东西都可以在RabbitMQ的管理控制台来管理，下一节我们就一起来学习控制台的使用。

2.2.收发消息

2.2.1.交换机

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

 

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

 

 

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

2.2.2.队列

我们打开Queues选项卡，新建一个队列：

 

命名为hello.queue1：

 

再以相同的方式，创建一个队列，密码为hello.queue2，最终队列列表如下：

 

此时，我们再次向amq.fanout交换机发送一条消息。会发现消息依然没有到达队列！！

怎么回事呢？

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

2.2.3.绑定关系

点击Exchanges选项卡，点击amq.fanout交换机，进入交换机详情页，然后点击Bindings菜单，在表单中填写要绑定的队列名称：

 

相同的方式，将hello.queue2也绑定到改交换机。

最终，绑定结果如下：

2.2.4.发送消息

再次回到exchange页面，找到刚刚绑定的amq.fanout，点击进入详情页，再次发送一条消息：

 

回到Queues页面，可以发现hello.queue中已经有一条消息了：

 

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

 

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

 

这个时候如果有消费者监听了MQ的hello.queue1或hello.queue2队列，自然就能接收到消息了。

2.3.数据隔离

2.3.1.用户管理

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

 

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

• Name：itheima，也就是用户名

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

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

对于小型企业而言，出于成本考虑，我们通常只会搭建一套MQ集群，公司内的多个不同项目同时使用。这个时候为了避免互相干扰， 我们会利用virtual host的隔离特性，将不同项目隔离。一般会做两件事情：

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

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

比如，我们给黑马商城创建一个新的用户，命名为hmall：

 

你会发现此时hmall用户没有任何virtual host的访问权限：

 

别急，接下来我们就来授权。

2.3.2.virtual host

我们先退出登录：

 

切换到刚刚创建的hmall用户登录，然后点击Virtual Hosts菜单，进入virtual host管理页：

 

可以看到目前只有一个默认的virtual host，名字为 /。

我们可以给黑马商城项目创建一个单独的virtual host，而不是使用默认的/。

 

创建完成后如图：

 

由于我们是登录hmall账户后创建的virtual host，因此回到users菜单，你会发现当前用户已经具备了对/hmall这个virtual host的访问权限了：

此时，点击页面右上角的virtual host下拉菜单，切换virtual host为 /hmall：

 

然后再次查看queues选项卡，会发现之前的队列已经看不到了：

 

这就是基于virtual host 的隔离效果。

3.SpringAMQP

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

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

SpringAmqp的官方地址：

SpringAMQP提供了三个功能：

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

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

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

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

3.1.导入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.快速入门

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

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

也就是：

• publisher直接发送消息到队列

• 消费者监听并处理队列中的消息

注意：这种模式一般测试使用，很少在生产中使用。

为了方便测试，我们现在控制台新建一个队列：simple.queue

 

添加成功：

 

接下来，我们就可以利用Java代码收发消息了。

3.2.1.消息发送

首先配置MQ地址，在publisher服务的application.yml中添加配置：

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

然后在publisher服务中编写测试类SpringAmqpTest，并利用RabbitTemplate实现消息发送：

package com.itheima.publisher.amqp;

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

@SpringBootTest
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, spring amqp!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }
}

打开控制台，可以看到消息已经发送到队列中：

 

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

3.2.2.消息接收

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

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

然后在consumer服务的com.itheima.consumer.listener包中新建一个类SpringRabbitListener，代码如下：

package com.itheima.consumer.listener;

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

@Component
public class SpringRabbitListener {
        // 利用RabbitListener来声明要监听的队列信息
    // 将来一旦监听的队列中有了消息，就会推送给当前服务，调用当前方法，处理消息。
    // 可以看到方法体中接收的就是消息体的内容
    @RabbitListener(queues = "simple.queue")
    public void listenSimpleQueueMessage(String msg) throws InterruptedException {
        System.out.println("spring 消费者接收到消息：【" + msg + "】");
    }
}

3.2.3.测试

启动consumer服务，然后在publisher服务中运行测试代码，发送MQ消息。最终consumer收到消息：

3.3.WorkQueues模型

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

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

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

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

首先，我们在控制台创建一个新的队列，命名为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);
    }
}

3.3.2.消息接收

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

@RabbitListener(queues = "work.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消费者1接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "work.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2........接收到消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}

注意到这两消费者，都设置了Thead.sleep，模拟任务耗时：

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

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

3.3.3.测试

启动ConsumerApplication后，在执行publisher服务中刚刚编写的发送测试方法testWorkQueue。

最终结果如下：

消费者1接收到消息：【hello, message_0】21:06:00.869555300
消费者2........接收到消息：【hello, message_1】21:06:00.884518
消费者1接收到消息：【hello, message_2】21:06:00.907454400
消费者1接收到消息：【hello, message_4】21:06:00.953332100
消费者1接收到消息：【hello, message_6】21:06:00.997867300
消费者1接收到消息：【hello, message_8】21:06:01.042178700
消费者2........接收到消息：【hello, message_3】21:06:01.086478800
消费者1接收到消息：【hello, message_10】21:06:01.087476600
消费者1接收到消息：【hello, message_12】21:06:01.132578300
消费者1接收到消息：【hello, message_14】21:06:01.175851200
消费者1接收到消息：【hello, message_16】21:06:01.218533400
消费者1接收到消息：【hello, message_18】21:06:01.261322900
消费者2........接收到消息：【hello, message_5】21:06:01.287003700
消费者1接收到消息：【hello, message_20】21:06:01.304412400
消费者1接收到消息：【hello, message_22】21:06:01.349950100
消费者1接收到消息：【hello, message_24】21:06:01.394533900
消费者1接收到消息：【hello, message_26】21:06:01.439876500
消费者1接收到消息：【hello, message_28】21:06:01.482937800
消费者2........接收到消息：【hello, message_7】21:06:01.488977100
消费者1接收到消息：【hello, message_30】21:06:01.526409300
消费者1接收到消息：【hello, message_32】21:06:01.572148
消费者1接收到消息：【hello, message_34】21:06:01.618264800
消费者1接收到消息：【hello, message_36】21:06:01.660780600
消费者2........接收到消息：【hello, message_9】21:06:01.689189300
消费者1接收到消息：【hello, message_38】21:06:01.705261
消费者1接收到消息：【hello, message_40】21:06:01.746927300
消费者1接收到消息：【hello, message_42】21:06:01.789835
消费者1接收到消息：【hello, message_44】21:06:01.834393100
消费者1接收到消息：【hello, message_46】21:06:01.875312100
消费者2........接收到消息：【hello, message_11】21:06:01.889969500
消费者1接收到消息：【hello, message_48】21:06:01.920702500
消费者2........接收到消息：【hello, message_13】21:06:02.090725900
消费者2........接收到消息：【hello, message_15】21:06:02.293060600
消费者2........接收到消息：【hello, message_17】21:06:02.493748
消费者2........接收到消息：【hello, message_19】21:06:02.696635100
消费者2........接收到消息：【hello, message_21】21:06:02.896809700
消费者2........接收到消息：【hello, message_23】21:06:03.099533400
消费者2........接收到消息：【hello, message_25】21:06:03.301446400
消费者2........接收到消息：【hello, message_27】21:06:03.504999100
消费者2........接收到消息：【hello, message_29】21:06:03.705702500
消费者2........接收到消息：【hello, message_31】21:06:03.906601200
消费者2........接收到消息：【hello, message_33】21:06:04.108118500
消费者2........接收到消息：【hello, message_35】21:06:04.308945400
消费者2........接收到消息：【hello, message_37】21:06:04.511547700
消费者2........接收到消息：【hello, message_39】21:06:04.714038400
消费者2........接收到消息：【hello, message_41】21:06:04.916192700
消费者2........接收到消息：【hello, message_43】21:06:05.116286400
消费者2........接收到消息：【hello, message_45】21:06:05.318055100
消费者2........接收到消息：【hello, message_47】21:06:05.520656400
消费者2........接收到消息：【hello, message_49】21:06:05.723106700

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

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

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

也就是说消息是平均分配给每个消费者，并没有考虑到消费者的处理能力。导致1个消费者空闲，另一个消费者忙的不可开交。没有充分利用每一个消费者的能力，最终消息处理的耗时远远超过了1秒。这样显然是有问题的。

3.3.4.能者多劳

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

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

再次测试，发现结果如下：

消费者1接收到消息：【hello, message_0】21:12:51.659664200
消费者2........接收到消息：【hello, message_1】21:12:51.680610
消费者1接收到消息：【hello, message_2】21:12:51.703625
消费者1接收到消息：【hello, message_3】21:12:51.724330100
消费者1接收到消息：【hello, message_4】21:12:51.746651100
消费者1接收到消息：【hello, message_5】21:12:51.768401400
消费者1接收到消息：【hello, message_6】21:12:51.790511400
消费者1接收到消息：【hello, message_7】21:12:51.812559800
消费者1接收到消息：【hello, message_8】21:12:51.834500600
消费者1接收到消息：【hello, message_9】21:12:51.857438800
消费者1接收到消息：【hello, message_10】21:12:51.880379600
消费者2........接收到消息：【hello, message_11】21:12:51.899327100
消费者1接收到消息：【hello, message_12】21:12:51.922828400
消费者1接收到消息：【hello, message_13】21:12:51.945617400
消费者1接收到消息：【hello, message_14】21:12:51.968942500
消费者1接收到消息：【hello, message_15】21:12:51.992215400
消费者1接收到消息：【hello, message_16】21:12:52.013325600
消费者1接收到消息：【hello, message_17】21:12:52.035687100
消费者1接收到消息：【hello, message_18】21:12:52.058188
消费者1接收到消息：【hello, message_19】21:12:52.081208400
消费者2........接收到消息：【hello, message_20】21:12:52.103406200
消费者1接收到消息：【hello, message_21】21:12:52.123827300
消费者1接收到消息：【hello, message_22】21:12:52.146165100
消费者1接收到消息：【hello, message_23】21:12:52.168828300
消费者1接收到消息：【hello, message_24】21:12:52.191769500
消费者1接收到消息：【hello, message_25】21:12:52.214839100
消费者1接收到消息：【hello, message_26】21:12:52.238998700
消费者1接收到消息：【hello, message_27】21:12:52.259772600
消费者1接收到消息：【hello, message_28】21:12:52.284131800
消费者2........接收到消息：【hello, message_29】21:12:52.306190600
消费者1接收到消息：【hello, message_30】21:12:52.325315800
消费者1接收到消息：【hello, message_31】21:12:52.347012500
消费者1接收到消息：【hello, message_32】21:12:52.368508600
消费者1接收到消息：【hello, message_33】21:12:52.391785100
消费者1接收到消息：【hello, message_34】21:12:52.416383800
消费者1接收到消息：【hello, message_35】21:12:52.439019
消费者1接收到消息：【hello, message_36】21:12:52.461733900
消费者1接收到消息：【hello, message_37】21:12:52.485990
消费者1接收到消息：【hello, message_38】21:12:52.509219900
消费者2........接收到消息：【hello, message_39】21:12:52.523683400
消费者1接收到消息：【hello, message_40】21:12:52.547412100
消费者1接收到消息：【hello, message_41】21:12:52.571191800
消费者1接收到消息：【hello, message_42】21:12:52.593024600
消费者1接收到消息：【hello, message_43】21:12:52.616731800
消费者1接收到消息：【hello, message_44】21:12:52.640317
消费者1接收到消息：【hello, message_45】21:12:52.663111100
消费者1接收到消息：【hello, message_46】21:12:52.686727
消费者1接收到消息：【hello, message_47】21:12:52.709266500
消费者2........接收到消息：【hello, message_48】21:12:52.725884900
消费者1接收到消息：【hello, message_49】21:12:52.746299900

可以发现，由于消费者1处理速度较快，所以处理了更多的消息；消费者2处理速度较慢，只处理了6条消息。而最终总的执行耗时也在1秒左右，大大提升。

正所谓能者多劳，这样充分利用了每一个消费者的处理能力，可以有效避免消息积压问题。

3.3.5.总结

Work模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理

• 通过设置prefetch来控制消费者预取的消息数量

3.4.交换机类型

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

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

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

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

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

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

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

交换机的类型有四种：

• Fanout：广播，将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机

• Direct：订阅，基于RoutingKey（路由key）发送给订阅了消息的队列

• Topic：通配符订阅，与Direct类似，只不过RoutingKey可以使用通配符

• Headers：头匹配，基于MQ的消息头匹配，用的较少。

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

3.5.Fanout交换机

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

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

• 1） 可以有多个队列

• 2） 每个队列都要绑定到Exchange（交换机）

• 3） 生产者发送的消息，只能发送到交换机

• 4） 交换机把消息发送给绑定过的所有队列

• 5） 订阅队列的消费者都能拿到消息

我们的计划是这样的：

• 创建一个名为 hmall.fanout的交换机，类型是Fanout

• 创建两个队列fanout.queue1和fanout.queue2，绑定到交换机hmall.fanout

3.5.1.声明队列和交换机

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

在创建一个队列fanout.queue2：

然后再创建一个交换机：

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

3.5.2.消息发送

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

@Test
public void testFanoutExchange() {
    // 交换机名称
    String exchangeName = "hmall.fanout";
    // 消息
    String message = "hello, everyone!";
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}

3.5.3.消息接收

在consumer服务的SpringRabbitListener中添加两个方法，作为消费者：

@RabbitListener(queues = "fanout.queue1")
public void listenFanoutQueue1(String msg) {
    System.out.println("消费者1接收到Fanout消息：【" + msg + "】");
}

@RabbitListener(queues = "fanout.queue2")
public void listenFanoutQueue2(String msg) {
    System.out.println("消费者2接收到Fanout消息：【" + msg + "】");
}

3.5.4.总结

交换机的作用是什么？

• 接收publisher发送的消息

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

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

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

3.6.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发送消息

3.6.1.声明队列和交换机

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

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

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

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

3.6.2.消息接收

在consumer服务的SpringRabbitListener中添加方法：

@RabbitListener(queues = "direct.queue1")
public void listenDirectQueue1(String msg) {
    System.out.println("消费者1接收到direct.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "direct.queue2")
public void listenDirectQueue2(String msg) {
    System.out.println("消费者2接收到direct.queue2的消息：【" + msg + "】");
}

3.6.3.消息发送

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

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "hmall.direct";
    // 消息
    String message = "红色警报！日本乱排核废水，导致海洋生物变异，惊现哥斯拉！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}

由于使用的red这个key，所以两个消费者都收到了消息：

我们再切换为blue这个key：

@Test
public void testSendDirectExchange() {
    // 交换机名称
    String exchangeName = "hmall.direct";
    // 消息
    String message = "最新报道，哥斯拉是居民自治巨型气球，虚惊一场！";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "blue", message);
}

你会发现，只有消费者1收到了消息：

3.6.4.总结

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

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

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

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

3.7.Topic交换机

3.7.1.说明

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

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

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交换机的用法。

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

3.7.2.消息发送

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

/**
 * topicExchange
 */
@Test
public void testSendTopicExchange() {
    // 交换机名称
    String exchangeName = "hmall.topic";
    // 消息
    String message = "喜报！孙悟空大战哥斯拉，胜!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.news", message);
}

3.7.3.消息接收

在consumer服务的SpringRabbitListener中添加方法：

@RabbitListener(queues = "topic.queue1")
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(queues = "topic.queue2")
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}

3.7.4.总结

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

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

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

• #：代表0个或多个词

• *：代表1个词

3.8.声明队列和交换机

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

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

3.8.1.基本API

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

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

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

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

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类型交换机
     */
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("hmall.fanout");
    }

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

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

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

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);
    }
}

3.8.2.fanout示例

在consumer中创建一个类，声明队列和交换机：

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类型交换机
     */
    @Bean
    public DirectExchange directExchange(){
        return ExchangeBuilder.directExchange("hmall.direct").build();
    }

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

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

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

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
    }
}

3.8.3.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类型交换机
     */
    @Bean
    public DirectExchange directExchange(){
        return ExchangeBuilder.directExchange("hmall.direct").build();
    }

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

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithRed(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue1WithBlue(Queue directQueue1, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue1).to(directExchange).with("blue");
    }

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

    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithRed(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("red");
    }
    /**
     * 绑定队列和交换机
     */
    @Bean
    public Binding bindingQueue2WithYellow(Queue directQueue2, DirectExchange directExchange){
        return BindingBuilder.bind(directQueue2).to(directExchange).with("yellow");
    }
}

3.8.4.基于注解声明

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

例如，我们同样声明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 + "】");
}

是不是简单多了。

再试试Topic模式：

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue1"),
    exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
    key = "china.#"
))
public void listenTopicQueue1(String msg){
    System.out.println("消费者1接收到topic.queue1的消息：【" + msg + "】");
}

@RabbitListener(bindings = @QueueBinding(
    value = @Queue(name = "topic.queue2"),
    exchange = @Exchange(name = "hmall.topic", type = ExchangeTypes.TOPIC),
    key = "#.news"
))
public void listenTopicQueue2(String msg){
    System.out.println("消费者2接收到topic.queue2的消息：【" + msg + "】");
}

3.9.消息转换器

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

 

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

只不过，默认情况下Spring采用的序列化方式是JDK序列化。众所周知，JDK序列化存在下列问题：

• 数据体积过大

• 有安全漏洞

• 可读性差

我们来测试一下。

3.9.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）发送消息

我们在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);
}

发送消息后查看控制台：

 

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

3.9.2.配置JSON转换器

显然，JDK序列化方式并不合适。我们希望消息体的体积更小、可读性更高，因此可以使用JSON方式来做序列化和反序列化。

在publisher和consumer两个服务中都引入依赖：

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

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

配置消息转换器，在publisher和consumer两个服务的启动类中添加一个Bean即可：

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

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

此时，我们到MQ控制台删除object.queue中的旧的消息。然后再次执行刚才的消息发送的代码，到MQ的控制台查看消息结构：

3.9.3.消费者接收Object

我们在consumer服务中定义一个新的消费者，publisher是用Map发送，那么消费者也一定要用Map接收，格式如下：

@RabbitListener(queues = "object.queue")
public void listenSimpleQueueMessage(Map<String, Object> msg) throws InterruptedException {
    System.out.println("消费者接收到object.queue消息：【" + msg + "】");
}

4.业务改造

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

如图：

说明：目前没有通知服务和积分服务，因此我们只关注交易服务，步骤如下：

• 定义direct类型交换机，命名为pay.direct

• 定义消息队列，命名为trade.pay.success.queue

• 将trade.pay.success.queue与pay.direct绑定，BindingKey为pay.success

• 支付成功时不再调用交易服务更新订单状态的接口，而是发送一条消息到pay.direct，发送消息的RoutingKey 为pay.success，消息内容是订单id

• 交易服务监听trade.pay.success.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 # 密码

4.1.接收消息

在trade-service服务中定义一个消息监听类：

 

其代码如下：

package com.hmall.trade.listener;

import com.hmall.trade.service.IOrderService;
import lombok.RequiredArgsConstructor;
import org.springframework.amqp.core.ExchangeTypes;
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;

@Component
@RequiredArgsConstructor
public class PayStatusListener {

    private final IOrderService orderService;

    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(name = "trade.pay.success.queue", durable = "true"),
            exchange = @Exchange(name = "pay.topic"),
            key = "pay.success"
    ))
    public void listenPaySuccess(Long orderId){
        orderService.markOrderPaySuccess(orderId);
    }
}

4.2.发送消息

修改pay-service服务下的com.hmall.pay.service.impl.PayOrderServiceImpl类中的tryPayOrderByBalance方法：

private final RabbitTemplate rabbitTemplate;

@Override
@Transactional
public void tryPayOrderByBalance(PayOrderDTO payOrderDTO) {
    // 1.查询支付单
    PayOrder po = getById(payOrderDTO.getId());
    // 2.判断状态
    if(!PayStatus.WAIT_BUYER_PAY.equalsValue(po.getStatus())){
        // 订单不是未支付，状态异常
        throw new BizIllegalException("交易已支付或关闭！");
    }
    // 3.尝试扣减余额
    userClient.deductMoney(payOrderDTO.getPw(), po.getAmount());
    // 4.修改支付单状态
    boolean success = markPayOrderSuccess(payOrderDTO.getId(), LocalDateTime.now());
    if (!success) {
        throw new BizIllegalException("交易已支付或关闭！");
    }
    // 5.修改订单状态
    // tradeClient.markOrderPaySuccess(po.getBizOrderNo());
    try {
        rabbitTemplate.convertAndSend("pay.direct", "pay.success", po.getBizOrderNo());
    } catch (Exception e) {
        log.error("支付成功的消息发送失败，支付单id：{}， 交易单id：{}", po.getId(), po.getBizOrderNo(), e);
    }
}

5.练习

5.1.抽取共享的MQ配置

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

5.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.登录信息传递优化

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

• 消息发送者在消息体中传递登录用户

• 消费者获取消息体中的登录用户，处理业务

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

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

参考资料：

理解：Day06 - MQ基础

理论理解

在微服务架构中，服务间的通讯方式对系统的性能、扩展性及健壮性具有重要影响。最常见的通讯方式有同步调用和异步调用两种，它们各自有着不同的适用场景和优缺点。

• 同步调用：在同步调用中，调用方在发起请求后需要等待服务方返回响应，才能继续后续操作。该方式简单且直观，但存在一些明显的问题，特别是在高并发场景下。比如，假如支付服务在调用用户服务进行余额扣减时，必须等待用户服务完成扣减并返回结果，才能继续执行后续的支付流程。如果某个服务调用失败，可能会导致整个业务链条的失败，形成级联故障。

• 异步调用：为了避免同步调用带来的性能瓶颈和级联失败问题，异步调用应运而生。在异步调用中，调用方不再等待服务方的响应，而是通过消息队列等中间件发送消息，服务方通过订阅消息来执行后续的业务。这种方式解耦了服务之间的调用，显著提高了系统的性能与扩展性。消息队列（MQ）就是一种典型的异步通讯机制，它在分布式系统中扮演着重要角色。

大厂实战理解

在大规模的分布式系统中，异步调用是提升性能和系统健壮性的重要手段。举个例子，在电商系统中，支付服务往往需要调用多个后端服务（如用户服务、订单服务、通知服务等），这些服务间的依赖关系可以通过消息队列来实现异步通知，从而避免了传统同步调用中的阻塞等待问题。通过使用消息队列，支付服务可以先完成余额扣减的操作并发送成功消息，而不必等待后续服务的响应。这使得系统能够在高并发场景下依然保持高效运行。

例如，某电商平台在用户支付后，会向多个微服务发送异步消息，包括：

1. 扣减用户余额

2. 更新支付流水状态

3. 清理购物车

4. 发送支付成功通知

5. 更新订单状态

如果这些操作都采用同步调用，随着业务量的增大，系统的响应时间会急剧增加，甚至会引发服务的级联故障。而通过异步调用（如RabbitMQ、Kafka等消息队列系统），各个微服务之间的依赖关系得到了解耦，即使某个微服务出现故障，也不会影响到主服务的运行，保障了系统的高可用性。

通过使用MQ，系统实现了解耦、异步化、可扩展性强、故障隔离等多重优势，使得业务流程更加灵活和健壮。进一步地，消息队列的引入不仅仅是为了异步调用，还是处理大规模并发、事务管理、流量控制等复杂场景的有效工具。

在许多大厂中，MQ作为系统间通讯的重要组成部分，应用广泛。例如，阿里巴巴使用Kafka和RocketMQ处理高并发的业务请求，京东也大量依赖消息队列来支持订单系统的高效和稳定。通过精确控制消息流量、优化消息存储和处理机制，这些公司能够确保大规模用户请求的平稳处理。

小结

• 同步调用虽然简单直观，但当涉及到多个服务时会带来性能瓶颈和级联失败问题，特别是在高并发场景下。

• 异步调用通过消息队列解耦服务之间的依赖，提高了系统的性能和可靠性。在处理高并发、事务管理和消息流量控制等方面，异步调用显得尤为重要。

• 常见的消息队列技术如RabbitMQ、Kafka等，在大厂的微服务架构中扮演着核心角色，广泛应用于处理复杂的分布式事务和异步业务流程。

在微服务架构中，正确选择同步与异步调用方式、合理运用消息队列，可以有效提升系统的扩展性、可维护性和容错能力。

 

大厂面试题及参考答案

1. 什么是消息队列？在微服务架构中，使用消息队列的好处是什么？

参考答案：

消息队列（MQ）是一种通信机制，它允许应用程序以异步的方式传递消息。消息被发送到一个队列中，由消费者程序从队列中获取并处理。这种模式解耦了生产者与消费者的关系，允许它们并行运行。

在微服务架构中使用消息队列有以下好处：

• 解耦：生产者和消费者不直接通信，而是通过消息队列传递消息。这样可以使系统的各个模块独立运行，减少彼此间的依赖。

• 异步处理：消息队列支持异步消息传递，可以提高系统的性能，特别是在需要执行大量计算或等待外部资源的情况下。

• 可靠性：消息队列可以持久化消息，即使消费者暂时不可用，消息也不会丢失。

• 负载均衡：多个消费者可以并行消费消息，从而提高处理效率。

• 流量削峰：通过设置消费者的处理能力，可以平滑处理突发的流量。

2. 请简要解释一下RabbitMQ的工作原理及其主要组件。

参考答案：

RabbitMQ 是一个开源的消息代理系统，它实现了高级消息队列协议（AMQP）。它的工作原理基于消息的生产者、消费者和消息队列之间的传递。RabbitMQ的主要组件如下：

• Producer（生产者）：生成并发送消息到消息队列。

• Consumer（消费者）：从消息队列中读取并处理消息。

• Exchange（交换机）：接收来自生产者的消息，并根据预定的规则（路由键、队列类型等）将消息转发到一个或多个队列。

• Queue（队列）：存储消息，直到消费者获取并处理这些消息。

• Binding（绑定）：交换机与队列之间的关联关系，通过路由键（Routing Key）来指定消息如何路由。

• Virtual Host（虚拟主机）：RabbitMQ 支持多个虚拟主机，每个虚拟主机都是一个隔离的环境，可以包含多个交换机、队列和绑定。

工作流程：

1. 生产者将消息发送到交换机。

2. 交换机根据消息的路由规则将消息转发到一个或多个队列。

3. 消费者从队列中获取消息并处理。

3. 什么是分布式事务？如何解决分布式事务中的一致性问题？

参考答案：

分布式事务是指在分布式系统中，跨越多个微服务或数据库的一致性操作。在传统的单体架构中，数据库事务通过ACID（原子性、一致性、隔离性、持久性）保证数据的一致性。然而，分布式事务涉及多个系统和数据库，这使得ACID原则变得难以实现。

解决分布式事务一致性问题的方法有以下几种：

• 两阶段提交（2PC）：每个参与者都在本地执行事务并提交，然后协调者收集所有节点的提交决策。如果所有节点都同意提交，事务就会成功，否则进行回滚。此方法的缺点是性能较差，尤其是在节点数量较多时，且单点故障会影响整个事务的执行。

• 补偿事务：通过执行补偿操作来回滚之前的操作。例如，分布式事务的每个步骤都有一个补偿事务（如手动回滚），当一个操作失败时，触发补偿事务来恢复系统的一致性。

• Saga模式：Saga模式是一种长时间运行的事务模型，它将大事务拆分为多个局部事务，每个局部事务完成后，通过消息或事件通知其他服务继续执行。如果某个局部事务失败，则会执行相应的补偿事务，保证整个事务的一致性。

• Seata框架：Seata是一个分布式事务解决方案，通过事务协调者（TC）和资源管理器（RM）来处理各个微服务的事务。它通过两种模式（AT模式和TCC模式）来解决分布式事务的ACID问题。

4. RabbitMQ与Kafka的区别？

参考答案：

RabbitMQ 和 Kafka 都是常用的消息队列，但它们在设计目标、架构以及使用场景上有一些区别：

• 消息模型：

  • RabbitMQ：基于AMQP协议，支持丰富的消息传递模式（如工作队列、发布订阅、路由模式等），适用于复杂的消息路由和异步消息传递。

  • Kafka：基于发布订阅模型，设计为高吞吐量的流式处理系统，主要用于处理大规模的实时数据流。

• 消息持久化：

  • RabbitMQ：支持消息的持久化，消息在写入队列时可选择持久化到磁盘，但会对性能产生一定影响。

  • Kafka：设计为高吞吐量的数据流平台，所有消息都持久化到磁盘中，且具有强大的日志存储和消费机制。

• 消息消费模式：

  • RabbitMQ：传统的消费者消费模型，支持多种消息传递方式，包括点对点和发布/订阅模式，消息队列中的消息只能被一个消费者消费。

  • Kafka：每个消息可以被多个消费者组消费，且消费者组能够并行消费，提高了消息消费的并发性。

• 性能与吞吐量：

  • RabbitMQ：适合较低延迟、复杂的消息路由场景，性能适中。

  • Kafka：优化了高吞吐量的场景，能够处理大规模的消息流，适用于需要高吞吐量、高可扩展性的场景。

• 使用场景：

  • RabbitMQ：适合需要复杂路由、可靠消息传递的业务场景，典型应用如任务调度系统。

  • Kafka：适用于大数据处理、实时流处理、日志聚合等场景，广泛应用于大规模数据传输和分析。

5. 如何在Spring Cloud中实现服务保护？

参考答案：

在Spring Cloud中，服务保护的核心是确保在面对高并发和异常情况时，系统能够保持高可用性和稳定性。主要技术包括：

1. 限流（Rate Limiting）：限流是控制请求访问的速率，以防止服务过载。常用工具如Hystrix、Resilience4j、Sentinel等。

2. 熔断（Circuit Breaker）：熔断器是一种保护机制，防止故障扩散到整个系统。通过设置阈值（如失败率、响应时间等），一旦达到阈值，熔断器会暂时停止请求，避免进一步恶化服务状态。

3. 请求隔离：通过将不同类型的请求（如长时间运行的请求和短时间运行的请求）隔离，减少对系统资源的竞争。例如，使用线程池隔离或信号量隔离。

4. 重试机制：为失败的请求提供重试功能，避免因偶然的故障导致的请求丢失。可以在重试的次数、间隔等方面进行配置。

5. 服务降级（Fallback）：当服务不可用时，提供备用服务或默认响应，保证系统的健壮性。通过Hystrix的fallback机制或Resilience4j的fallback实现。

6. Sentinel：Sentinel是阿里巴巴开源的服务保护框架，提供了流量控制、熔断、降级、限流等功能，能帮助微服务在高负载时保持稳定。

6. 解释一下分布式锁的实现方式？

参考答案：

分布式锁用于解决在分布式环境下多个节点对同一资源的竞争问题。常见的实现方式有以下几种：

1. 基于数据库的分布式锁：

   • 利用数据库的唯一索引，创建一个锁表。当某个节点需要获得锁时，往锁表中插入一条记录，插入成功即表示获得锁。

   • 缺点：会对数据库产生额外的负载，且需要考虑锁超时、死锁等问题。

2. 基于Redis的分布式锁：

   • 利用Redis的SETNX命令（SET if Not eXists），当锁不存在时，设置锁值并设置过期时间。通过Redis的GET和DEL命令来检查锁状态和释放锁。

   • 使用Redis的优点是速度快，适用于高并发环境，但需要避免死锁和分布式环境中的锁过期等问题。可以使用Redisson、Spring Data Redis等工具简化实现。

3. 基于Zookeeper的分布式锁：

   • Zookeeper通过其顺序节点和Watcher机制来实现分布式锁。每个客户端在Zookeeper中创建一个临时顺序节点，客户端通过监听节点的顺序来判断是否获得锁。

   • Zookeeper锁机制能够确保锁的公平性，但性能相对较低，适用于需要强一致性和协调性的场景。

4. 基于Etcd的分布式锁：

   • Etcd类似于Zookeeper，提供了分布式键值存储。通过compare-and-swap（CAS）机制来确保锁的正确性。

   • 适用于需要强一致性和高可靠性的环境，但与Zookeeper类似，性能较低。

大厂场景题及参考答案

1. 场景：一个电商平台的订单服务存在性能瓶颈，订单查询请求处理时响应时间较长。你作为系统架构师，如何优化这个系统？

参考答案：

针对订单查询性能瓶颈，可以从以下几个方面进行优化：

1. 数据库优化：

   • 数据库索引优化：检查订单表中的查询条件，确保常用的查询字段（如订单ID、用户ID、订单状态等）有适当的索引。使用合适的索引可以大幅提高查询性能。

   • 数据库分表分库：根据订单的量级，将订单表进行水平分割，例如按时间、地域或订单ID范围进行分表，从而避免单表查询性能下降。

   • 查询缓存：对于热点数据（如某些常用的订单查询），可以使用Redis缓存查询结果，减少数据库的负载。

2. 系统架构优化：

   • 异步处理：对于一些不需要实时响应的操作（如订单状态更新、短信通知等），可以将其通过消息队列异步处理，减轻主线程的负担，避免阻塞。

   • API网关：通过API网关引入请求限流和负载均衡机制，防止单个服务被过多的查询请求压垮。

   • 微服务拆分：将订单服务拆分成多个微服务，针对订单查询、订单创建、订单支付等不同业务模块进行单独优化和扩展。

3. 前端优化：

   • 懒加载与分页查询：对于订单列表等查询结果，可以使用分页查询的方式，避免一次性加载大量数据。结合懒加载技术，优化用户体验。

   • 缓存前端数据：对于不常更新的订单数据，可以使用浏览器缓存或CDN进行缓存，从而减少对后台系统的压力。

4. 数据存储优化：

   • NoSQL数据库：对于一些简单查询（如用户的历史订单），可以使用NoSQL数据库（如MongoDB、Cassandra等），它们的查询性能比关系型数据库更适合高频读取操作。

2. 场景：你负责一个支付系统的开发，系统需要处理高并发的支付请求，同时保证数据的一致性和可靠性。你如何设计这个系统？

参考答案：

1. 高并发处理设计：

   • 限流与流量控制：在支付高并发场景下，可以使用限流策略（如令牌桶、漏桶算法等）来限制系统的处理能力，防止系统过载。可以结合API网关或应用层限流库实现限流。

   • 队列与异步处理：支付请求不需要完全同步处理，可以将支付请求放入消息队列（如Kafka、RabbitMQ），通过后台异步处理支付逻辑，从而解耦请求的接收和业务处理。

   • 负载均衡：使用负载均衡器（如Nginx、LVS、Kubernetes等）将支付请求均匀分配到多台服务器，避免单点故障和瓶颈。

   • 缓存：使用Redis等缓存技术缓存热点数据（如用户余额），减少数据库压力，提高响应速度。

2. 数据一致性设计：

   • 分布式事务管理：支付业务涉及多个微服务（如支付服务、订单服务、库存服务），需要确保事务一致性。可以使用Seata等分布式事务框架来处理跨服务的事务一致性。

   • 补偿机制：如果支付过程中某个服务失败，应该使用补偿机制来回滚已完成的部分操作，例如通过Saga模式实现事务的补偿。

   • 幂等性设计：支付接口必须保证幂等性，防止用户重复支付。在接口设计时，确保支付请求具有唯一标识（如支付ID），并在处理前验证该支付请求是否已经处理过。

3. 数据可靠性与保障：

   • 事务日志：确保支付过程中的所有操作都被记录到日志中，以便出现问题时能够回滚或重试。

   • 消息队列的可靠性：确保消息队列具有持久化能力，防止消息丢失。消息消费后要及时确认，避免消息被重复处理。

   • 超时与重试机制：支付请求可能会因网络或其他故障而超时，因此在支付过程中应实现超时重试机制。重试策略应当考虑幂等性和避免过度重试。

4. 高可用与灾备设计：

   • 主从数据库：使用主从数据库或数据库集群来保证系统的高可用，读请求可以分摊到从库，减少主库的压力。

   • 分布式部署：系统应当部署在多个数据中心或云平台中，以保证在某个数据中心发生故障时，系统仍然可以继续提供服务。

3. 场景：在一个电商系统中，你需要处理支付后的订单状态更新，如何确保订单状态的一致性？

参考答案：

为了确保订单状态的一致性，可以从以下几个方面着手：

1. 最终一致性：

   • 消息队列与异步处理：支付完成后，可以通过消息队列（如RabbitMQ、Kafka等）将订单更新任务异步化，订单状态的更新由消费消息的服务处理，从而避免直接在支付过程中进行数据库操作，减少数据库压力。

   • 幂等性设计：为了防止因网络问题或系统重试导致的重复更新，订单状态更新操作需要具有幂等性。可以通过唯一标识（如订单ID或支付ID）来判断该订单是否已被处理。

2. 事务保障：

   • 分布式事务：支付过程中涉及多个微服务（如订单服务、支付服务、库存服务），可以使用分布式事务框架（如Seata、TCC等）来确保跨服务的事务一致性。

   • 补偿机制：如果某个服务在订单状态更新过程中失败，可以使用补偿机制来回滚已完成的操作，确保最终一致性。可以采用Saga模式或TCC模式来处理分布式事务。

3. 订单状态机：

   • 状态机设计：在订单服务中，可以设计一个状态机来管理订单的生命周期，保证订单状态的转换是有序且一致的。例如，订单从"待支付"状态转换到"已支付"状态后，才可以进行后续的发货等操作。

   • 状态锁定：在更新订单状态时，可以通过锁机制来避免多线程或多服务并发修改同一订单状态，保证状态更新的一致性。

4. 超时检测与修复：

   • 超时监控：在支付或订单处理过程中，可以设置超时检测机制，如果某个操作在规定时间内未完成，系统会自动进行重试或回滚操作。

   • 定期同步：通过定期任务同步订单状态，确保在系统发生故障或数据不一致时，能够及时修复错误状态。

4. 场景：如何设计一个高可用的电商系统，避免单点故障影响业务正常运行？

参考答案：

设计一个高可用的电商系统，主要通过以下几种方式来避免单点故障：

1. 多节点部署与负载均衡：

   • 多数据中心部署：将系统部署在多个地理位置分散的数据中心，一旦某个数据中心发生故障，其他数据中心可以接管流量，保障业务持续运行。

   • 负载均衡：使用负载均衡器（如Nginx、LVS、F5等）将流量分配到多台服务器，避免单个服务器出现故障导致业务中断。可以通过DNS负载均衡或反向代理实现流量分发。

2. 数据库高可用：

   • 主从复制：数据库采用主从复制架构，主库负责写操作，从库负责读操作。主库故障时，切换到备用主库或从库，确保数据库的高可用性。

   • 数据库分片：对于大规模数据，可以通过数据库分片技术将数据分布到多个数据库实例中，减少单一数据库的压力，提高可用性和扩展性。

   • 数据库故障切换：使用数据库高可用架构（如MySQL Cluster、Galera Cluster）保证在主数据库宕机时能够自动切换到备份数据库。

3. 微服务架构：

   • 微服务拆分：将电商系统拆分成多个微服务，每个微服务独立部署和管理。这样，即使某个服务出现故障，其他服务仍然可以继续运行，不会影响整体业务。

   • 服务发现与负载均衡：通过服务发现框架（如Eureka、Consul等）和负载均衡器动态路由请求，确保请求能够找到可用的服务实例。

4. 容器化与自动化运维：

   • 容器化部署：使用Docker容器化各个微服务，并利用Kubernetes等容器编排工具实现自动扩展和故障恢复。容器化部署能够保证系统的可移植性和弹性。

   • 自动化监控与报警：通过Prometheus、Grafana等工具对系统的健康状态进行实时监控，一旦出现异常，立即报警并触发自动恢复措施，如重启服务、自动扩容等。

5. 数据备份与灾备：

   • 定期数据备份：定期进行全量和增量数据备份，确保数据丢失时可以迅速恢复。

   • 灾备演练：定期进行灾备演练，模拟数据中心故障、系统崩溃等情况，验证灾备方案的有效性，确保在发生故障时能迅速恢复业务。

通过这些措施，可以有效降低电商系统单点故障的风险，确保业务的高可用性和持续性。

1. 场景：在一个分布式文件存储系统中，用户上传文件时，如何保证文件上传的高效性和可靠性？

参考答案：

为了保证文件上传的高效性和可靠性，可以考虑以下几点：

1. 文件分块上传：

   • 将大文件分割为多个小块（例如，分为5MB一块），并允许并行上传每个小块。这样可以大幅度提高文件上传的速度，并减少因为网络波动或连接丢失导致的上传失败的概率。

   • 使用分块上传的方式，每个分块上传后可以立即进行处理和存储，确保文件上传过程中不会产生长时间的阻塞。

2. 分布式存储：

   • 文件存储应使用分布式文件系统（如HDFS、Ceph等），这可以保证文件的冗余存储、快速检索以及高可用性。

   • 分布式存储还可以避免单点故障问题，提高系统的容错性。

3. 消息队列与异步处理：

   • 上传的文件通过消息队列（如Kafka、RabbitMQ等）进行异步处理，上传操作将不再阻塞用户请求，提高用户体验。

   • 文件上传成功后，通过消息队列将文件的处理请求（如转码、缩略图生成、病毒扫描等）推送到后台处理。

4. 幂等性设计：

   • 为了避免因网络问题导致的重复上传，应在系统设计中确保上传操作具有幂等性，确保同一文件不会被重复上传。

   • 文件的唯一标识（如hash值）可以作为上传请求的一部分，判断文件是否已上传过。

5. 上传进度监控：

   • 实现上传进度反馈，用户可以看到文件上传的实时进度，改善用户体验。

---

2. 场景：设计一个高并发的在线教育平台，用户观看视频时，如何设计系统架构确保视频流的稳定播放和高并发支持？

参考答案：

1. 内容分发网络（CDN）：

   • 使用CDN进行视频文件的缓存和分发。将视频文件存储在离用户更近的边缘节点，通过CDN降低服务器压力，提高视频的加载速度和播放的稳定性。

   • CDN能够智能地选择最优路径和节点，减少延迟，优化大规模视频流量的传输。

2. 视频分片与流媒体协议：

   • 将视频文件进行分片存储，通过HTTP Live Streaming (HLS) 或 Dynamic Adaptive Streaming over HTTP (DASH) 协议将视频分段播放。

   • 使用HLS或DASH能够根据网络状况自动调整视频的质量，保证即使在网络较差的情况下也能平稳播放，减少卡顿现象。

3. 负载均衡：

   • 配置负载均衡器（如Nginx、HAProxy等）来均衡流量，确保用户请求被合理分配到多个服务实例，避免单个节点的过载。

   • 使用基于地域的负载均衡策略，确保用户请求被路由到离其最近的服务节点。

4. 异步处理与消息队列：

   • 对于视频播放时可能需要的后台操作（如视频转码、上传记录、用户互动数据统计等），可以使用消息队列（如Kafka、RabbitMQ）异步处理，减少对前端请求的响应时间。

5. 流量控制与限流：

   • 在高并发场景下，使用流量控制和限流机制，防止因请求激增导致系统崩溃。可以使用令牌桶算法或者漏桶算法对视频流的请求进行控制。

---

3. 场景：设计一个高可用的微服务系统，如何处理服务间的通讯，保证高并发场景下的可靠性和数据一致性？

参考答案：

1. 服务间异步通讯：

   • 使用消息队列（如Kafka、RabbitMQ等）来异步处理服务间的通讯，避免因同步调用导致的性能瓶颈。

   • 使用异步消息传递可以减少系统的耦合度，降低请求响应时间。

2. 分布式事务管理：

   • 在微服务之间进行事务管理时，使用分布式事务解决方案，如Seata，Saga模式等来保证多个微服务的业务一致性。

   • 对于不可避免的失败，使用补偿机制（如TCC模式或Saga模式）来确保系统最终一致性。

3. 服务发现与负载均衡：

   • 使用服务注册与发现（如Eureka、Consul、Zookeeper等）来管理微服务实例，确保每个微服务都能够自动发现并访问到其他服务。

   • 配置负载均衡（如Ribbon、Nginx）来处理高并发请求，确保请求能够均匀分布到多台服务实例上，避免单点故障。

4. 断路器与熔断：

   • 使用Hystrix或Resilience4j等断路器来避免级联故障。当某个服务发生故障时，通过断路器的熔断机制阻止故障蔓延到其他服务，保证系统的可用性。

   • 配置熔断策略，失败请求快速失败，减少因依赖故障而拖慢整体系统的时间。

5. 分布式缓存：

   • 对于热点数据，使用分布式缓存（如Redis、Memcached）来提高访问速度，减轻数据库的压力，确保高并发环境下的数据访问性能。

---

4. 场景：设计一个在线广告系统，如何处理广告投放的精确匹配与高效实时分析？

参考答案：

1. 广告投放系统设计：

   • 实时数据流处理：使用Kafka等消息队列将用户行为数据（如点击、浏览等）实时传输至分析系统，进行广告匹配和推荐。

   • 广告匹配引擎：广告系统需要有一个高效的广告匹配引擎，根据用户的实时行为和兴趣（如点击历史、搜索关键词、地理位置等）动态生成广告投放策略，确保广告的精准匹配。

2. 大数据处理与分析：

   • 大数据存储与处理：使用Hadoop、Spark等大数据技术进行大规模数据存储与分析，实时计算用户行为数据、广告效果等。

   • 机器学习推荐算法：通过机器学习算法（如协同过滤、深度学习等）对广告进行精准推荐，根据用户的偏好、历史行为和上下文信息进行实时分析与广告投放。

3. 数据缓存与实时查询：

   • 使用Redis等内存缓存技术来存储热门广告、用户数据、广告投放策略等，提升广告匹配速度。

   • Elasticsearch可以用于实时广告查询，提供高效的搜索和分析能力，快速返回相关广告。

4. 广告投放效果监控与分析：

   • 使用数据可视化工具（如Grafana、Tableau）对广告投放效果进行监控与实时分析，评估广告点击率、转化率等指标，并优化投放策略。

   • A/B 测试：在广告投放中，使用A/B测试来验证不同广告策略的效果，不断优化广告推荐模型。

---

5. 场景：设计一个实时在线聊天系统，如何保证消息传递的低延迟和可靠性？

参考答案：

1. 消息传递的低延迟：

   • WebSocket协议：使用WebSocket协议建立持久连接，确保消息可以在客户端和服务器之间实时双向传输，减少传统HTTP协议带来的请求延迟。

   • 消息队列与缓存：在高并发场景下，使用消息队列（如Kafka、RabbitMQ等）将聊天消息进行异步处理，减少直接在数据库上的操作，提高系统性能。

   • CDN加速：对消息数据进行压缩，通过CDN网络传输，减少网络延迟。

2. 消息的可靠性：

   • 消息持久化：使用消息队列的持久化功能确保消息不会因系统崩溃丢失。

   • 消息确认机制：对每条消息进行确认，保证消息已经成功送达接收方，防止消息丢失。

3. 扩展性与高可用：

   • 微服务架构：将聊天系统拆分为多个微服务，针对不同的业务模块（如消息推送、消息存储、用户管理等）进行单独部署和扩展。

   • 负载均衡：采用负载均衡（如Nginx、HAProxy）分担各个服务器的压力，确保系统的高可用性。

   • 数据库分片与缓存：使用数据库分片技术以及缓存（如Redis）来减少数据库压力，提高读写性能。

---

场景题 1: 电商平台的异步订单处理

题目：
你正在为一个电商平台设计订单支付流程。订单支付成功后，系统需要进行以下操作：

• 扣减用户余额

• 更新支付流水记录

• 通知仓库服务，准备发货

• 发送短信通知用户支付成功

目前系统采用同步调用方式，即每一个操作都等待前一个操作完成后才能执行。然而，你发现系统性能较差，且每当仓库服务或短信通知服务失败时，支付操作会失败，影响用户体验。请描述如何使用RabbitMQ改进这个流程，并解决性能问题和级联失败问题。

答案：
在电商平台的订单支付流程中，我们可以使用RabbitMQ来异步处理这些任务，从而提高系统的性能并减少级联失败的风险。具体实现方式如下：

1. 解耦各个服务：每个操作（扣减余额、更新支付流水、通知仓库和发送短信通知）都可以设计为独立的微服务，彼此之间不直接调用，而是通过RabbitMQ消息队列进行通信。

2. 使用异步通信：

   • 支付服务在完成余额扣减和支付流水更新后，将支付成功的消息发布到RabbitMQ队列。消息内容可以包含订单ID、支付信息等。

   • 仓库服务、短信通知服务等都作为消息消费者，异步地从RabbitMQ队列中获取支付成功消息并进行处理。

3. RabbitMQ的配置：

   • 创建一个交换机（如 payment.success.exchange）用于处理支付成功的消息。

   • 为每个服务创建不同的队列（如 payment.paymentSuccessQueue、payment.warehouseQueue、payment.smsQueue），并将这些队列绑定到交换机。

4. 故障隔离：如果仓库服务或短信服务由于某种原因失败，这些服务不会影响支付服务的执行。支付服务只需要将消息投递到RabbitMQ队列中，其他服务可以在消息投递后异步处理。如果某个服务失败，不会影响到支付流程。

5. 性能提升：所有操作都不需要同步执行，支付服务只需要快速将消息发送到RabbitMQ，减少了请求的响应时间，提高了整体性能。

---

场景题 2: 金融平台的异步支付与账务处理

题目：
在金融平台中，用户支付操作涉及多个环节：

1. 扣减账户余额

2. 更新交易记录

3. 发起账单结算

目前系统采用同步调用，支付操作完成时需要等待每个环节依次完成。假设由于某些原因，余额扣减操作延迟较长，导致整体支付响应时间变慢，且每个环节都依赖前一个环节完成，出现级联失败。如何使用RabbitMQ优化系统性能，并保证支付操作的高可用性？

答案：
为了解决支付操作的性能瓶颈和级联失败问题，可以使用RabbitMQ进行异步处理。具体的解决方案如下：

1. 异步处理支付操作：

   • 在支付服务中，完成余额扣减后，将扣减后的支付成功消息发送到RabbitMQ队列，并通知交易记录服务和结算服务。

   • 交易记录服务和结算服务作为消息消费者，从队列中消费消息并执行相应的处理（如更新交易记录、发起账单结算）。

2. 服务解耦：不同环节（如余额扣减、交易记录更新、结算）的操作通过RabbitMQ实现解耦，避免了直接的同步调用。每个服务独立运行，彼此之间通过消息传递进行协调。

3. 容错与重试机制：RabbitMQ支持消息的可靠投递，并能够在服务故障时进行消息重试或死信队列处理，保证消息不会丢失。若交易记录更新失败，可以将消息放入死信队列进行重新处理。

4. 性能提升：通过异步通信，支付服务完成后无需等待交易记录和结算操作的完成，极大地减少了整体响应时间。此外，多个服务可以并行处理消息，进一步提高系统吞吐量。

---

场景题 3: 物联网设备数据收集与分析

题目：
在一个物联网（IoT）系统中，设备不断生成传感器数据，需要将这些数据传输到云端进行分析处理。设备之间的通信可能非常密集，且需要高效、可靠地传输数据。如何利用RabbitMQ设计一个高效的数据传输系统，并确保数据的可靠性和高效性？

答案：
在物联网系统中，设备生成的数据量巨大且不断增加，使用RabbitMQ可以确保数据传输的可靠性和高效性。解决方案如下：

1. 使用RabbitMQ作为数据传输中间件：

   • 每个设备将传感器数据发送到RabbitMQ队列，队列作为中间缓存，确保数据能够可靠地传输到云端进行分析。

   • 设备可以异步地将数据推送到RabbitMQ，减少设备与云端的直接交互，提高响应速度。

2. 分布式消费者：

   • 云端的分析服务可以作为消息消费者，从RabbitMQ队列中拉取传感器数据进行实时分析。通过增加消费者的数量，可以保证系统能够处理大量的设备数据，提高系统的吞吐量。

3. 队列管理与分配：

   • 每种设备类型可以设置不同的队列，根据设备的种类和优先级将数据分配到不同的队列，以确保数据传输的高效性和正确性。

4. 数据可靠性保障：

   • RabbitMQ支持消息确认机制，确保每条数据都被成功传输。在传输过程中，RabbitMQ会对消息进行持久化，确保设备断电或网络中断时，数据不会丢失。

   • 可配置消息的重试机制和死信队列，确保在某个环节出现故障时，数据可以得到恢复和重新处理。

5. 高效数据流处理：

   • 为了处理海量数据，可以使用RabbitMQ的Fanout交换机将消息广播到多个消费者，并行处理不同的数据分析任务，如实时监控、趋势分析等。

---

场景题 4: 在线社交平台的消息推送

题目：
在一个社交平台中，每当用户发送消息或点赞时，系统需要向被点赞的用户推送通知。为了确保系统的高效性和可靠性，请设计一个基于RabbitMQ的消息推送系统，并解释如何避免高并发下的消息丢失。

答案：
在社交平台的消息推送场景中，使用RabbitMQ可以实现高效、可靠的通知服务。具体设计如下：

1. 消息队列与消费者解耦：

   • 每当用户发送消息或点赞时，系统将消息推送的请求投递到RabbitMQ队列，而不是直接向被点赞的用户推送通知。

   • **消费者（通知服务）**从队列中获取消息，并将通知发送给相应的用户。

2. Fanout交换机实现广播：

   • 使用RabbitMQ的Fanout交换机，确保所有绑定到该交换机的队列都能收到推送消息。

   • 通过将多个队列绑定到Fanout交换机，通知服务可以同时处理多个用户的推送请求，从而提高系统的并发处理能力。

3. 消息持久化与确认机制：

   • 向RabbitMQ发送消息时，可以使用消息持久化机制，确保即使RabbitMQ发生崩溃或重启，消息也不会丢失。

   • 启用消息确认机制，确保消息已经被消费者成功消费。如果消费失败，消息可以重新入队或发送到死信队列进行重试。

4. 高并发下的消息分发：

   • 如果平台的活跃度很高，使用多个消费者实例来分担消息处理任务，通过扩展消费者的数量来提高消息的处理能力。

   • 使用负载均衡的策略，确保每个消费者处理的任务量均匀，避免某个消费者因过多的任务而成为瓶颈。

5.4.改造项目一

思考一下，项目一中的哪些业务可以由同步方式改为异步方式调用？试着改造一下。

举例：短信发送