---
title: MQ 面试
date: 2022-02-17 22:34:30
categories:
- 分布式
- 分布式通信
- MQ
- MQ综合
tags:
- Java
- 中间件
- MQ
- 面试
permalink: /pages/3510f99a/
---
# MQ 面试
> 消息队列（Message Queue，简称 MQ）技术是**应用间交换信息**的一种技术。
>
> 消息队列主要解决异步处理、应用间耦合，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。
>
> 目前主流的 MQ 有：Kafka、RabbitMQ、RocketMQ、ActiveMQ，而部分数据库如 Redis、MySQL 以及 phxsql 也可实现消息队列的功能。
>
> 注意：_为了简便，下文中除了文章标题，一律使用 MQ 简称_。
## MQ 简介
### 【简单】什么是 MQ？
**MQ（Message Queue，消息队列）** 是一种**异步通信机制**，用于在不同服务、应用或系统组件之间**可靠地传递消息**。它的核心思想是**解耦生产者和消费者**，通过缓冲消息来提高系统的可靠性、扩展性和可维护性。
**MQ 的核心概念**
- **生产者（Producer）**：发送消息的应用或服务。
- **消费者（Consumer）**：接收并处理消息的应用或服务。
- **消息（Message）**：传输的数据单位，可以是文本、JSON、二进制等格式。
- **队列（Queue）**：存储消息的缓冲区，遵循 **FIFO（先进先出）** 或优先级策略。
- **Broker（消息代理）**：负责接收、存储和转发消息的中间件（如 RabbitMQ、Kafka）。
消息队列主要解决应用耦合，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。是大型分布式系统不可缺少的中间件。
MQ 是消费-生产者模型的一个典型的代表，一端往消息队列中不断写入消息，而另一端则可以读取队列中的消息。消息发布者只管把消息发布到 MQ 中而不用管谁来取，消息使用者只管从 MQ 中取消息而不管是谁发布的。这样发布者和使用者都不用知道对方的存在。
MQ 的数据可驻留在内存或磁盘上，直到它们被应用程序读取。通过 MQ，应用程序可独立地执行，它们不需要知道彼此的位置，不需要等待接收程序接收此消息。在分布式计算环境中，为了集成分布式应用，开发者需要对异构网络环境下的分布式应用提供有效的通信手段。为了管理需要共享的信息，对应用提供公共的信息交换机制是重要的。
目前主流的 MQ 有：Kafka、RabbitMQ、RocketMQ、ActiveMQ。
### 【简单】为什么需要 MQ？
**MQ 的典型应用场景**
- **异步处理**
- **系统解耦**
- **流量削峰**
- **系统间通信**
- **传输缓冲**
- **最终一致性**
::: info 异步处理
:::
MQ 可以将系统间的处理流程异步化，减少等待响应的时间，从而提高整体并发吞吐量。一般，MQ 异步处理应用于非核心流程，例如：短信/邮件通知、数据推送、上报数据到监控中心、日志中心等。
假设这样一个场景，用户向系统 A 发起请求，系统 A 处理计算只需要 `10ms`，然后通知系统 BCD 写库，系统 BCD 写库耗时分别为：`100ms`、`200ms`、`300ms`。最终总耗时为： `10ms+100ms+200ms+300ms=610ms`。此外，加上请求和响应的网络传输时间，从用户角度看，可能要等待将近 `1s` 才能得到结果。
如果使用 MQ，系统 A 接到请求后，耗时 `10ms` 处理计算，然后向系统 BCD 连续发送消息，假设耗时 `5ms`。那么 这一过程的总耗时为 `3ms + 5ms = 8ms`，这相比于 `610 ms`，大大缩短了响应时间。至于系统 BCD 的写库操作，只要自行消费 MQ 后处理即可，用户无需关注。
::: info 系统解耦
:::
通过 MQ，可以消除系统间的强耦合。它的好处在于：
- 消息的消费者系统可以随意增加，无需修改生产者系统的代码。
- 生产者系统、消费者系统彼此不会影响对方的流程。
- 如果生产者系统宕机，消费者系统收不到消息，就不会有下一步的动作。
- 如果消费者系统宕机，生产者系统让然可以正常发送消息，不影响流程。
不同系统如果要建立通信，传统的做法是：调用接口。
如果需要和新的系统建立通信或删除已建立的通信，都需要修改代码，这种方案显然耦合度很高。
如果使用 MQ，系统间的通信只需要通过发布/订阅（Pub/Sub）模型即可，彼此没有直接联系，也就不需要相互感知，从而达到 **解耦**。
::: info 流量削峰
:::
当 **上下游系统** 处理能力存在差距的时候，利用 MQ 做一个 “**漏斗**” 模型，进行 **流控**。把 MQ 当成可靠的 **消息缓冲池**，进行一定程度的 **消息堆积**；在下游有能力处理的时候，再发送消息。
MQ 的流量削峰常用于高并发场景（例如：秒杀、团抢等业务场景），它是缓解瞬时暴增流量的核心手段之一。
如果没有 MQ，两个系统之间通过 **协商**、**滑动窗口**、**限流**/**降级**/**熔断** 等复杂的方案也能实现 **流控**。但 **系统复杂性** 指数级增长，势必在上游或者下游做存储，并且要处理 **定时**、**拥塞** 等一系列问题。而且每当有 **处理能力有差距** 的时候，都需要 **单独** 开发一套逻辑来维护这套逻辑。
假设某个系统读写数据库的稳定性能为每秒处理 1000 条数据。平常情况下，远远达不到这么大的处理量。假设，因为因为做活动，系统的瞬时请求量剧增，达到每秒 10000 个并发请求，数据库根本承受不了，可能直接就把数据库给整崩溃了，这样系统服务就不可用了。
如果使用 MQ，每秒写入 10000 条请求，但是系统 A 每秒只从 MQ 中消费 1000 条请求，然后写入数据库。这样，就不会超过数据库的承受能力，而是把请求积压在 MQ 中。只要高峰期一过，系统 A 就会很快把积压的消息给处理掉。
::: info 系统间通信
:::
消息队列一般都内置了 **高效的通信机制**，因此也可以用于单纯的 **消息通讯**，比如实现 **点对点消息队列** 或者 **聊天室** 等。
**生产者/消费者** 模式，只需要关心消息是否 **送达队列**，至于谁希望订阅和需要消费，是 **下游** 的事情，无疑极大地减少了开发和联调的工作量。
::: info 传输缓冲
:::
（1）MQ 常被用于做海量数据的传输缓冲。
例如，Kafka 常被用于做为各种日志数据、采集数据的数据中转。然后，Kafka 将数据转发给 Logstash、Elasticsearch 中，然后基于 Elasticsearch 来做日志中心，提供检索、聚合、分析日志的能力。开发者可以通过 Kibana 集成 Elasticsearch 数据进行可视化展示，或自行进行定制化开发。
（2）MQ 也可以被用于流式处理。
例如，Kafka 几乎已经是流计算的数据采集端的标准组件。而流计算通过实时数据处理能力，提供了更为快捷的聚合计算能力，被大量应用于链路监控、实时监控、实时数仓、实时大屏、风控、推荐等应用领域。
::: info 最终一致性
:::
**最终一致性** 不是 **消息队列** 的必备特性，但确实可以依靠 **消息队列** 来做 **最终一致性** 的事情。
- **先写消息再操作**，确保操作完成后再修改消息状态。**定时任务补偿机制** 实现消息 **可靠发送接收**、业务操作的可靠执行，要注意 **消息重复** 与 **幂等设计**。
- 所有不保证 `100%` **不丢消息** 的消息队列，理论上无法实现 **最终一致性**。
> 像 `Kafka` 一类的设计，在设计层面上就有 **丢消息** 的可能（比如 **定时刷盘**，如果掉电就会丢消息）。哪怕只丢千分之一的消息，业务也必须用其他的手段来保证结果正确。
### 【中等】引入 MQ 带来哪些问题？
任何技术都会有利有弊，MQ 给整体系统架构带来很多好处，但也会付出一定的代价。
MQ 主要引入了以下问题：
- **系统可用性降低**：引入了 MQ 后，通信需要基于 MQ 完成，如果 MQ 宕机，则服务不可用。因此，MQ 要保证是高可用的。
- **系统复杂度提高**：使用 MQ，需要关注一些新的问题：
- 如何保证消息没有 **重复消费**？
- 如何处理 **消息丢失** 的问题？
- 如何保证传递 **消息的顺序性**？
- 如何处理大量 **消息积压** 的问题？
- **一致性问题**：假设系统 A 处理完直接返回成功的结果给用户，用户认为请求成功。但如果此时，系统 BCD 中只要有任意一个写库失败，那么数据就不一致了。这种情况如何处理？
### 【中等】MQ 有哪些通信模型？
MQ（消息队列）常见的通信模型主要有以下几种，适用于不同的业务场景：
**点对点（Point-to-Point / Queue）**
- **特点**：消息由 **一个生产者** 发送到 **一个队列**，**只有一个消费者** 能消费该消息（竞争消费），消息被消费后即从队列删除。
- **适用场景**：任务分发（如订单处理、异步任务）。
- **示例**：RabbitMQ 的普通队列、ActiveMQ 的 Queue。
**发布/订阅（Publish/Subscribe / Topic）**
- **特点**：消息由 **一个生产者** 发送到 **Topic（主题）**，**多个消费者**（订阅者）可同时接收同一消息，消息会广播给所有订阅者。
- **适用场景**：事件通知（如系统日志广播、实时数据同步）。
- **示例**：RabbitMQ 的 Exchange + Fanout 模式、Kafka 的 Topic。
**请求/响应（Request/Reply）**
- **特点**：生产者发送消息后，消费者处理并返回响应（类似 RPC），通常需要 **临时队列** 存储响应。
- **适用场景**：需要同步结果的异步调用（如支付状态查询）。
- **示例**：RabbitMQ 的 RPC 模式、Kafka 的 Request-Reply 扩展。
**扇出（Fanout）**
- **特点**：消息发送到 Exchange 后，**无条件广播** 给所有绑定的队列（无路由规则），类似发布/订阅，但更简单。
- **适用场景**：实时通知多个系统（如价格变动、库存更新）。
- **示例**：RabbitMQ 的 Fanout Exchange。
**路由（Routing / Direct）**
- **特点**：消息根据 **Routing Key** 被精准投递到匹配的队列，消费者只接收符合规则的消息。
- **适用场景**：条件过滤（如错误日志分级处理）。
- **示例**：RabbitMQ 的 Direct Exchange。
**主题路由（Topic）**
- **特点**：基于 **通配符**（如 `user.*`）匹配 Routing Key，实现灵活订阅，比 Direct 更灵活，比 Fanout 更精准。
- **适用场景**：复杂事件分发（如物联网设备消息分类）。
- **示例**：RabbitMQ 的 Topic Exchange、MQTT 的 Topic。
**总结**
| 模型 | 生产者-消费者关系 | 典型应用场景 |
| ------------- | ----------------- | ---------------------- |
| **点对点** | 1:1（竞争消费） | 任务队列、订单处理 |
| **发布/订阅** | 1:N（广播） | 日志广播、实时通知 |
| **请求/响应** | 1:1（带响应） | 异步 RPC、结果回调 |
| **扇出** | 1:N（无条件广播） | 多系统数据同步 |
| **路由** | 1:1（精准匹配） | 条件过滤、优先级队列 |
| **主题路由** | 1:N（通配符匹配） | 复杂事件分类（如 IoT） |
### 【中等】MQ 推拉模式各有什么利弊，如何选择？
消息引擎（MQ）获取消息的模式主要分为 **Push（推）** 和 **Pull（拉）** 两种，不同消息队列中间件采用不同的策略，部分系统还支持 **混合模式**。
**Push 模式（服务端推送）**
- **特点**：
- **消息由 Broker 主动推送给消费者**，消费者被动接收。
- 实时性高，减少消费者轮询开销。
- **优点**：低延迟，适合实时性要求高的场景（如即时通讯）。
- **缺点**：可能造成消费者过载（需背压机制控制流速）。
- **典型实现**：RabbitMQ、ActiveMQ、RocketMQ（默认长轮询模拟 Push）。
**Pull 模式（客户端拉取）**
- **特点**：
- **消费者主动从 Broker 拉取消息**，按需获取。
- 消费者控制消费速率。
- **优点**：避免消息堆积冲击消费者，适合高吞吐场景（如日志处理）。
- **缺点**：存在空轮询开销（可通过长轮询优化）。
- **典型实现**：Kafka、Pulsar（原生 Pull）、RocketMQ（支持显式 Pull）。
**长轮询（Long Polling）**
- **特点**：**Push 和 Pull 的折中方案**。消费者发起请求后，Broker 若无消息则保持连接，直到有消息或超时才返回。
- **优点**：减少无效轮询，平衡实时性与服务端压力。
- **典型实现**：RocketMQ（默认模式）、HTTP 长轮询（如 WebSocket）。
**混合模式（Push + Pull）**
- **特点**：
- 关键消息用 Push 保证实时性，批量数据用 Pull 提高吞吐。
- 消费者可动态切换模式。
- **典型实现**：Pulsar（支持多模式）、部分自研 MQ 系统。
**对比总结**
| 模式 | 实时性 | 服务端压力 | 消费者控制力 | 适用场景 |
| ---------- | ------ | ---------- | ------------ | -------------------------- |
| **Push** | 高 | 高 | 低 | 即时通讯、事件通知 |
| **Pull** | 低 | 低 | 高 | 大数据处理、日志收集 |
| **长轮询** | 中 | 中 | 中 | 平衡实时性与性能（如电商） |
**选择建议**
- 需要低延迟 → **Push**（如 RabbitMQ）。
- 需要高吞吐 → **Pull**（如 Kafka）。
- 平衡场景 → **长轮询**（如 RocketMQ）。
## MQ 可靠传输
### 【困难】如何保证 MQ 消息不丢失？
要保证 MQ 中的消息不丢失，需从 **生产端、MQ 服务端、消费端** 三个环节进行可靠性设计。一言以蔽之，**生产端确认+服务端持久化+消费端手动 ACK+监控补偿** 是保证消息不丢失的核心逻辑，需根据业务场景权衡性能与可靠性。
::: info 生产端防丢失
:::
| | Kafka | RocketMQ |
| ------------ | ---------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------ |
| **发送方式** | 支持同步、异步、异步回调发送方式
同步性能低；异步无视成功与否；异步回调比较合适 | 支持同步、异步回调发送方式
同步性能低；异步回调比较合适 |
| **生产 ACK** | `acks` 参数确保消息被多少副本写入成功后才返回确认
`min.insync.replicas` 配合 `acks` 使用，设定最少写入副本数 | （配合刷盘机制）若配置为同步刷盘，消息必须先成功写入磁盘，发送方才能收到写成功的确认 |
| **失败重试** | `retries` 参数控制失败重试次数 | `retryTimesWhenSendFailed` 参数控制失败重试次数 |
| **事务** | 通过事务+生产幂等，实现 exactly once 语义，非真正的分布式事务 | 可实现分布式事务 |
::: info MQ 服务端防丢失
:::
| | Kafka | RocketMQ |
| ------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------- |
| **持久化** | 消息写入页缓存（PageCache），即可返回生产 ACK，由 OS 负责刷盘（`fsync`）
可设置自动刷盘的间隔时间或消息数阈值
可以通过AdminClient手动刷盘 | 默认采用异步刷盘，类似 Kafka，先写入页缓存（PageCache），由 OS 负责刷盘（`fsync`）
也支持同步刷盘，消息必须先成功写入磁盘，发送方才能收到写成功的确认 |
| **副本机制** | 通过副本机制保证冗余，避免单点故障
副本的粒度是针对分区
`replication.factor` 设置分区副本数
`min.insync.replicas` 控制写入到多少个副本才算是“已提交” | 通过副本机制保证冗余，避免单点故障
副本的粒度是针对节点（Broker） |
| **故障检测** | 所有Broker会向zk的`/borker`路径写临时节点，而Controller会监听该目录
一旦有Broker故障或失联，zk会话中断，zk自动删除临时节点，并通知Controller
Controller将Broker视为下线，并将该Broker上的分区Leader置为无效 | 主节点定期向从节点发送心跳以续活
超时未收到消息，从节点视主节点为下线 |
| **故障恢复** | Controller 负责选新的分区 Leader，优先从ISR中选
通过 `unclean.leader.election.enable` 可设置是否允许从非ISR中选Leader，但丢失数据风险增大 | 采用 Raft 来选主 |
::: info 消费端防丢失
:::
| | **Kafka** | **RocketMQ** |
| --------------- | ------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------- |
| **消费 ACK** | 支持自动、手动提交Offset（`enable.auto.commit=false`配置）
关闭自动提交，处理完整事务后，再手动提交Offset | 支持自动、手动提交Offset（consumer.setAutoCommit配置）
关闭自动提交，处理完整事务后，再手动提交Offset |
| **重置 Offset** | `auto.offset.reset` 可以重置消费的 Offset | |
| **失败重试** | 不完善：自动提交不支持重试；手动提交续自行处理重试逻辑 | 默认重试16次
多次消费失败后会被投递到**死信队列**，供后续特殊处理 |
::: info 监控与补偿
:::
- **消息堆积告警**：监控队列长度，及时发现异常。
- **定期对账**：对比生产与消费的记录，修复差异（如定时扫描数据库补发）。
### 【困难】如何保证 MQ 消息不重复？
::: info MQ 为什么会出现重复消息？
:::
- 生产端重试（网络抖动时自动重发）
- 消费端超时后 MQ 重新投递（如 RabbitMQ 未及时 ACK）
- 消息队列集群脑裂（如 Kafka 副本切换）
以 Kafka 举例，Kafka 每个 Partition 都是一个有序的、不可变的记录序列，不断追加到结构化的提交日志中。Partition 中为每条记录分配一个连续的 id 号，称为偏移量（Offset），用于唯一标识 Partition 内的记录。
Kafka 的客户端和 Broker 都会保存 Offset。客户端消费消息后，每隔一段时间，就把已消费的 Offset 提交给 Kafka Broker，表示已消费。
在这个过程中，如果客户端应用消费消息后，因为宕机、重启等情况而没有提交已消费的 Offset 。当系统恢复后，会继续消费消息，由于 Offset 未提交，就会出现重复消费的问题。
::: info 重复消息通用解决方案
:::
处理重复消息 = **“业务幂等为基础，缓存/DB 去重为辅助，监控兜底保万一”**。
处理 MQ 重复消息的核心思路是 **幂等性设计** + **去重机制**，确保即使消息被多次消费，业务结果也不会出错。
（1）**业务层幂等设计**
- **唯一标识**：每条消息携带唯一业务 ID（如订单号、支付流水号），处理前先查库判断是否已执行。
```sql
SELECT status FROM orders WHERE order_id = '123';
-- 若已处理则直接跳过
```
- **状态机控制**：业务状态严格流转（如「已支付」订单不允许重复扣款）。
（2）**去重表/缓存**
- **数据库去重表**：消费前先 `INSERT` 唯一键（消息 ID），利用主键冲突避免重复处理。
```sql
INSERT INTO message_processed(id) VALUES ('msg_123') ON DUPLICATE KEY IGNORE;
```
- **Redis 去重**：用 `SETNX` 设置消息 ID 过期时间（适合高频场景）。
```bash
SETNX msg_123 1 EX 3600 # 1 小时内不重复处理
```
::: info 主流 MQ 处理
:::
| 消息队列 | 重复触发场景 | 推荐方案 |
| ------------ | ----------------------------- | ----------------------------- |
| **Kafka** | 消费者重启导致 offset 回滚 | 业务幂等 + 本地 offset 持久化 |
| **RabbitMQ** | 未 ACK 导致重新入队 | 手动 ACK + 死信队列监控 |
| **RocketMQ** | 消息重试机制（16 次后进死信） | 消费日志 + 人工干预 |
极端情况兜底：
- **对账系统**：定时扫描业务数据与消息记录，修复不一致（如定时补发短信）。
- **人工告警**：监控重复消息频率（如 1 分钟同消息 ID 出现 3 次以上则报警）。
::: info 方案选型
:::
- **低频业务**：数据库唯一索引（简单可靠）。
- **高频业务**：Redis + 过期时间（高性能）。
- **金融级场景**：幂等 + 对账 + 人工审核（强一致）。
### 【困难】如何保证 MQ 消息的顺序性？
要保证 MQ 消息的顺序性，需从 **生产、存储、消费** 三个环节控制。
核心思路是：**“同一业务 ID 锁定同一队列 + 单线程消费”**，需结合业务需求选择局部顺序或全局顺序方案。
::: info 生产端保序
:::
- **单生产者+单线程发送**：同一业务 ID（如订单 ID）的消息由 **同一生产者线程** 顺序发送，避免多线程并发乱序。
```java
// 示例：相同 orderId 的消息由同一线程发送
mqProducer.send(msg, orderId); // 通过 hash 选择分区
```
- **禁用异步发送重试**：异步发送失败时可能乱序，需同步发送或关闭重试（如 Kafka 配置 `max.in.flight.requests.per.connection=1`）。
::: info MQ 服务端保序
:::
- **单分区/队列有序**
- **Kafka/RocketMQ**：同一业务 ID 的消息发送到 **同一分区**（Partition）。
```java
// 根据 orderId 哈希选择分区
int partition = orderId.hashCode() % partitionNum;
```
- **RabbitMQ**：使用单队列（或一致性哈希交换器绑定唯一队列）。
- **关闭分区/队列并行**：避免服务端多分区/多副本间的顺序混乱（如 Kafka 的 `unclean.leader.election.enable=false`）。
::: info 消费端保序
:::
- **单消费者串行消费**
- 同一队列/分区由 **单消费者线程** 处理（如 Kafka 单线程消费或 `max.poll.records=1`）。
- 多消费者时，相同业务 ID 的消息路由到同一消费者（如 RocketMQ 的 `MessageQueueSelector`）。
- **内存队列排序**（复杂场景）：消费者拉取消息后，按业务 ID 分组存入内存队列，由不同线程分别串行处理。
::: info 特殊场景处理
:::
- **全局严格顺序**：牺牲性能，全链路单线程（生产→MQ→消费），仅适合低吞吐场景（如 Binlog 同步）。
- **局部顺序**：仅保证同一业务 ID 的顺序（如订单的创建→支付→退款），允许不同订单并发。
::: info 主流 MQ 处理
:::
| | **Kafka** | **RocketMQ** |
| ------------------- | ------------------------------------------------------------ | ------------------------------------------------------------ |
| **单分区/队列有序** | 单分区追加写入，天然有序 | 单队列有序 |
| **哈希路由** | 采用哈希路由，相同key固定发往同一分区
如果不指定key，则采用轮询方式选择分区 | 使用 `MessageQueueSelector` 指定队列 |
| **单线程生产** | `max.in.flight.requests.per.connection=1` 控制 | |
| **单线程消费** | 确保一个分区仅由一个消费者线程处理（Kafka 默认规则） | 使用 `MessageListenerOrderly` |
| **业务保证** | 避免多线程并发消费同一队列
分布式锁（如 Redis）
关键业务操作加锁，防止并发执行乱序 | 避免多线程并发消费同一队列
分布式锁（如 Redis）
关键业务操作加锁，防止并发执行乱序 |
::: info 注意事项
:::
- **性能权衡**：顺序性越高，并发性能越低（需根据业务容忍度平衡）。
- **错误处理**：消费失败时需暂停当前分区消费（如 Kafka 的 `pause()`），避免跳过消息导致乱序。
- **监控**：定期检查消息积压和顺序偏移（如 Kafka 的 `consumer.position()`）。
### 【困难】如何处理 MQ 消息积压？
处理 MQ 消息积压的核心思路是 **“快速消费存量+优化生产速率”**，需结合监控、扩容、降级等手段综合治理。
大致可以归纳为：
- **短期**：扩容+降级，优先恢复服务。
- **长期**：优化消费逻辑+自动化运维，预防再次积压。
- **口诀**：**监控早发现，扩容扛流量，消费改批量，生产限流速**。
::: info 快速消费积压消息
:::
- **增加消费者实例**：横向扩展消费者服务（如 Kubernetes 动态扩容 Pod），注意分区数限制（Kafka 需提前规划足够分区）。
- **提升消费并行度**：
- 调整消费者并发参数（如 Kafka 的 `max.poll.records`、RabbitMQ 的 `prefetch_count`）。
- 多线程消费（需保证无顺序要求的场景）。
- **临时降级**：非核心业务暂停消费（如日志处理），集中资源处理核心业务消息。
::: info 优化消费能力
:::
- **批量处理**：合并多条消息一次处理（如数据库批量插入）。
- **异步化+削峰**：消费者将消息存入内存队列，后台线程异步处理，避免同步阻塞。
- **跳过非关键逻辑**：临时关闭日志记录、数据校验等非必要操作。
::: info 控制生产端流量
:::
- **限流**：生产端启用速率限制（如 Kafka 的 `quota`、Redis 令牌桶）。
- **削峰填谷**：消息先写入缓存层（如 Redis List），再匀速写入 MQ。
- **业务降级**：高峰期关闭非核心功能的消息生产（如暂停推荐系统更新）。
::: info 监控与告警
:::
- **实时监控指标**：
- 队列堆积量（如 Kafka 的 `lag`）、消费速率（TPS）、消费者状态。
- 设置阈值告警（如积压超过 10W 条触发短信通知）。
- **根因分析工具**：
- 日志分析（消费者卡顿、GC 问题）。
- 链路追踪（如 SkyWalking 定位慢消费）。
::: info 长期预防措施
:::
- **容量规划**：根据业务峰值预先扩容分区/队列（如 Kafka 分区数 = 消费者数 × 1.5）。
- **死信队列+重试机制**：处理失败的消息转入死信队列，避免阻塞正常消费。
- **自动化扩缩容**：基于积压指标动态调整消费者数量（如 K8s HPA）。
::: info 主流 MQ 处理
:::
| 消息队列 | 关键操作 |
| ------------ | --------------------------------------- |
| **Kafka** | 增加分区+消费者，调整 `fetch.max.bytes` |
| **RabbitMQ** | 镜像队列扩容，提高 `prefetch_count` |
| **RocketMQ** | 消费组扩容，启用定时消息延迟消费 |
## MQ 高可用
### 【困难】如何保证 MQ 的高可用？
不同 MQ 实现高可用的原理各不相同。因为 Kafka 比较具有代表性，所以这里以 Kafka 为例。
::: info Kafka 的核心概念
:::
了解 Kafka，必须先了解 Kafka 的核心概念：
- **Broker** - Kafka 集群包含一个或多个节点，这种节点被称为 Broker。
- **Topic** - 每条发布到 Kafka 集群的消息都有一个类别，这个类别被称为 Topic。（不同 Topic 的消息是物理隔离的；同一个 Topic 的消息保存在一个或多个 Broker 上，但用户只需指定消息的 Topic 即可生产或消费数据而不必关心数据存于何处）。对于每一个 Topic， Kafka 集群都会维持一个分区日志。
- **Partition** - 了提高 Kafka 的吞吐率，每个 Topic 包含一个或多个 Partition，每个 Partition 在物理上对应一个文件夹，该文件夹下存储这个 Partition 的所有消息和索引文件。
- Kafka 日志的分区（Partition）分布在 Kafka 集群的节点上。每个节点在处理数据和请求时，共享这些分区。每一个分区都会在已配置的节点上进行备份，确保容错性。
::: info Kafka 的副本机制
:::
Kafka 是如何实现高可用的呢？
Kafka 在 0.8 以前的版本中，如果一个 Broker 宕机了，其上面的 Partition 都不能用了，这自然不是高可用的。
为了实现高可用，Kafka 引入了复制功能，简单来说，就是副本机制（ Replicate ）：
**每个 Partition 都有一个 Leader，零个或多个 Follower**。Leader 和 Follower 都是 Broker，每个 Broker 都会成为某些分区的 Leader 和某些分区的 Follower，因此集群的负载是平衡的。
- **Leader 处理一切对 Partition （分区）的读写请求**；
- **而 Follower 只需被动的同步 Leader 上的数据**。
**同一个 Topic 的不同 Partition 会分布在多个 Broker 上，而且一个 Partition 还会在其他的 Broker 上面进行备份**，Producer 在发布消息到某个 Partition 时，先找到该 Partition 的 Leader，然后向这个 Leader 推送消息；每个 Follower 都从 Leader 拉取消息，拉取消息成功之后，向 Leader 发送一个 ACK 确认。
> FAQ
>
> 问：为什么让 Leader 处理一切对对 Partition （分区）的读写请求？
>
> 答：因为如果允许所有 Broker 都可以处理读写请求，就可能产生数据一致性问题。
::: info Kafka 选举 Leader
:::
由上文可知，Partition 在多个 Broker 上存在副本。
如果某个 Follower 宕机，啥事儿没有，正常工作。
如果 Leader 宕机了，会从 Follower 中**重新选举**一个新的 Leader。
## MQ 架构
### 【困难】Kafka、ActiveMQ、RabbitMQ、RocketMQ 有什么优缺点？
::: info ActiveMQ
:::
`ActiveMQ` 是由 `Apache` 出品，`ActiveMQ` 是一个完全支持`JMS1.1` 和 `J2EE 1.4` 规范的 `JMS Provider` 实现。它非常快速，支持 **多种语言的客户端** 和 **协议**，而且可以非常容易的嵌入到企业的应用环境中，并有许多高级功能。
**(a) 主要特性**
1. **服从 JMS 规范**：`JMS` 规范提供了良好的标准和保证，包括：**同步** 或 **异步** 的消息分发，一次和仅一次的消息分发，**消息接收** 和 **订阅** 等等。遵从 `JMS` 规范的好处在于，不论使用什么 `JMS` 实现提供者，这些基础特性都是可用的；
2. **连接灵活性**：`ActiveMQ` 提供了广泛的 **连接协议**，支持的协议有：`HTTP/S`，`IP` **多播**，`SSL`，`TCP`，`UDP` 等等。对众多协议的支持让 `ActiveMQ` 拥有了很好的灵活性；
3. **支持的协议种类多**：`OpenWire`、`STOMP`、`REST`、`XMPP`、`AMQP`；
4. **持久化插件和安全插件**：`ActiveMQ` 提供了 **多种持久化** 选择。而且，`ActiveMQ` 的安全性也可以完全依据用户需求进行 **自定义鉴权** 和 **授权**；
5. **支持的客户端语言种类多**：除了 `Java` 之外，还有：`C/C++`，`.NET`，`Perl`，`PHP`，`Python`，`Ruby`；
6. **代理集群**：多个 `ActiveMQ` **代理** 可以组成一个 **集群** 来提供服务；
7. **异常简单的管理**：`ActiveMQ` 是以开发者思维被设计的。所以，它并不需要专门的管理员，因为它提供了简单又使用的管理特性。有很多中方法可以 **监控** `ActiveMQ` 不同层面的数据，包括使用在 `JConsole` 或者在 `ActiveMQ` 的 `Web Console` 中使用 `JMX`。通过处理 `JMX` 的告警消息，通过使用 **命令行脚本**，甚至可以通过监控各种类型的 **日志**。
**(b) 部署环境**
`ActiveMQ` 可以运行在 `Java` 语言所支持的平台之上。使用 `ActiveMQ` 需要：
- `Java JDK`
- `ActiveMQ` 安装包
**(c) 优点**
1. **跨平台** (`JAVA` 编写与平台无关，`ActiveMQ` 几乎可以运行在任何的 `JVM` 上）；
2. 可以用 `JDBC`：可以将 **数据持久化** 到数据库。虽然使用 `JDBC` 会降低 `ActiveMQ` 的性能，但是数据库一直都是开发人员最熟悉的存储介质；
3. 支持 `JMS` 规范：支持 `JMS` 规范提供的 **统一接口**;
4. 支持 **自动重连** 和 **错误重试机制**；
5. 有安全机制：支持基于 `shiro`，`jaas` 等多种 **安全配置机制**，可以对 `Queue/Topic` 进行 **认证和授权**；
6. 监控完善：拥有完善的 **监控**，包括 `Web Console`，`JMX`，`Shell` 命令行，`Jolokia` 的 `RESTful API`；
7. 界面友善：提供的 `Web Console` 可以满足大部分情况，还有很多 **第三方的组件** 可以使用，比如 `hawtio`；
**(d) 缺点**
1. 社区活跃度不及 `RabbitMQ` 高；
2. 根据其他用户反馈，会出莫名其妙的问题，会 **丢失消息**；
3. 目前重心放到 `activemq 6.0` 产品 `Apollo`，对 `5.x` 的维护较少；
4. 不适合用于 **上千个队列** 的应用场景；
::: info RabbitMQ
:::
`RabbitMQ` 于 `2007` 年发布，是一个在 `AMQP` (**高级消息队列协议**) 基础上完成的，可复用的企业消息系统，是当前最主流的消息中间件之一。
**(a) 主要特性**
1. **可靠性**：提供了多种技术可以让你在 **性能** 和 **可靠性** 之间进行 **权衡**。这些技术包括 **持久性机制**、**投递确认**、**发布者证实** 和 **高可用性机制**；
2. **灵活的路由**：消息在到达队列前是通过 **交换机** 进行 **路由** 的。`RabbitMQ` 为典型的路由逻辑提供了 **多种内置交换机** 类型。如果你有更复杂的路由需求，可以将这些交换机组合起来使用，你甚至可以实现自己的交换机类型，并且当做 `RabbitMQ` 的 **插件** 来使用；
3. **消息集群**：在相同局域网中的多个 `RabbitMQ` 服务器可以 **聚合** 在一起，作为一个独立的逻辑代理来使用；
4. **队列高可用**：队列可以在集群中的机器上 **进行镜像**，以确保在硬件问题下还保证 **消息安全**；
5. **支持多种协议**：支持 **多种消息队列协议**；
6. **支持多种语言**：用 `Erlang` 语言编写，支持只要是你能想到的 **所有编程语言**；
7. **管理界面**： `RabbitMQ` 有一个易用的 **用户界面**，使得用户可以 **监控** 和 **管理** 消息 `Broker` 的许多方面；
8. **跟踪机制**：如果 **消息异常**，`RabbitMQ` 提供消息跟踪机制，使用者可以找出发生了什么；
9. **插件机制**：提供了许多 **插件**，来从多方面进行扩展，也可以编写自己的插件。
**(b) 部署环境**
`RabbitMQ` 可以运行在 `Erlang` 语言所支持的平台之上，包括 `Solaris`，`BSD`，`Linux`，`MacOSX`，`TRU64`，`Windows` 等。使用 `RabbitMQ` 需要：
- `ErLang` 语言包
- `RabbitMQ` 安装包
**(c) 优点**
1. 由于 `Erlang` 语言的特性，消息队列性能较好，支持 **高并发**；
2. 健壮、稳定、易用、**跨平台**、支持 **多种语言**、文档齐全；
3. 有消息 **确认机制** 和 **持久化机制**，可靠性高；
4. 高度可定制的 **路由**；
5. **管理界面** 较丰富，在互联网公司也有较大规模的应用，社区活跃度高。
**(d) 缺点**
1. 尽管结合 `Erlang` 语言本身的并发优势，性能较好，但是不利于做 **二次开发和维护**；
2. 实现了 **代理架构**，意味着消息在发送到客户端之前可以在 **中央节点** 上排队。此特性使得 `RabbitMQ` 易于使用和部署，但是使得其 **运行速度较慢**，因为中央节点 **增加了延迟**，**消息封装后** 也比较大；
3. 需要学习 **比较复杂** 的 **接口和协议**，学习和维护成本较高。
::: info RocketMQ
:::
`RocketMQ` 出自 **阿里** 的开源产品，用 `Java` 语言实现，在设计时参考了 `Kafka`，并做出了自己的一些改进，**消息可靠性上** 比 `Kafka` 更好。`RocketMQ` 在阿里内部 被广泛应用在 **订单**，**交易**，**充值**，**流计算**，**消息推送**，**日志流式处理**，`binglog` **分发** 等场景。
**(a) 主要特性**
1. 基于 **队列模型**：具有 **高性能**、**高可靠**、**高实时**、**分布式** 等特点；
2. `Producer`、`Consumer`、**队列** 都支持 **分布式**；
3. `Producer` 向一些队列轮流发送消息，**队列集合** 称为 `Topic`。`Consumer` 如果做 **广播消费**，则一个 `Consumer` 实例消费这个 `Topic` 对应的 **所有队列**；如果做 **集群消费**，则 **多个** `Consumer` 实例 **平均消费** 这个 `Topic` 对应的队列集合；
4. 能够保证 **严格的消息顺序**；
5. 提供丰富的 **消息拉取模式**；
6. 高效的订阅者 **水平扩展**能力；
7. **实时** 的 **消息订阅机制**；
8. 亿级 **消息堆积** 能力；
9. 较少的外部依赖。
**(b) 部署环境**
`RocketMQ` 可以运行在 `Java` 语言所支持的平台之上。使用 `RocketMQ` 需要：
- `Java JDK`
- 安装 `git`、`Maven`
- `RocketMQ` 安装包
**(c) 优点**
1. **单机** 支持 `1` 万以上 **持久化队列**；
2. `RocketMQ` 的所有消息都是 **持久化的**，先写入系统 `PAGECACHE`，然后 **刷盘**，可以保证 **内存** 与 **磁盘** 都有一份数据，而 **访问** 时，直接 **从内存读取**。
3. 模型简单，接口易用（`JMS` 的接口很多场合并不太实用）；
4. **性能非常好**，可以允许 **大量堆积消息** 在 `Broker` 中；
5. 支持 **多种消费模式**，包括 **集群消费**、**广播消费**等；
6. 各个环节 **分布式扩展设计**，支持 **主从** 和 **高可用**；
7. 开发度较活跃，版本更新很快。
**(d) 缺点**
1. 支持的 **客户端语言** 不多，目前是 `Java` 及 `C++`，其中 `C++` 还不成熟；
2. `RocketMQ` 社区关注度及成熟度也不及前两者；
3. 没有 `Web` 管理界面，提供了一个 `CLI` （命令行界面） 管理工具带来 **查询**、**管理** 和 **诊断各种问题**；
4. 没有在 `MQ` 核心里实现 `JMS` 等接口；
::: info Kafka
:::
`Apache Kafka` 是一个 **分布式消息发布订阅** 系统。它最初由 `LinkedIn` 公司基于独特的设计实现为一个 **分布式的日志提交系统** (`a distributed commit log`)，之后成为 `Apache` 项目的一部分。`Kafka` **性能高效**、**可扩展良好** 并且 **可持久化**。它的 **分区特性**，**可复制** 和 **可容错** 都是其不错的特性。
**(a) 主要特性**
1. **快速持久化**：可以在 `O(1)` 的系统开销下进行 **消息持久化**；
2. **高吞吐**：在一台普通的服务器上既可以达到 `10W/s` 的 **吞吐速率**；
3. **完全的分布式系统**：`Broker`、`Producer` 和 `Consumer` 都原生自动支持 **分布式**，自动实现 **负载均衡**；
4. 支持 **同步** 和 **异步** 复制两种 **高可用机制**；
5. 支持 **数据批量发送** 和 **拉取**；
6. **零拷贝技术 (zero-copy)**：减少 `IO` 操作步骤，提高 **系统吞吐量**；
7. **数据迁移**、**扩容** 对用户透明；
8. **无需停机** 即可扩展机器；
9. **其他特性**：丰富的 **消息拉取模型**、高效 **订阅者水平扩展**、实时的 **消息订阅**、亿级的 **消息堆积能力**、定期删除机制；
**(b) 部署环境**
使用 `Kafka` 需要：
- `Java JDK`
- `Kafka` 安装包
**(c) 优点**
1. **客户端语言丰富**：支持 `Java`、`.Net`、`PHP`、`Ruby`、`Python`、`Go` 等多种语言；
2. **高性能**：单机写入 `TPS` 约在 `100` 万条/秒，消息大小 `10` 个字节；
3. 提供 **完全分布式架构**，并有 `replica` 机制，拥有较高的 **可用性** 和 **可靠性**，理论上支持 **消息无限堆积**；
4. 支持批量操作；
5. **消费者** 采用 `Pull` 方式获取消息。**消息有序**，**通过控制** 能够保证所有消息被消费且仅被消费 **一次**；
6. 有优秀的第三方 `Kafka Web` 管理界面 `Kafka-Manager`；
7. 在 **日志领域** 比较成熟，被多家公司和多个开源项目使用。
**(d) 缺点**
1. `Kafka` 单机超过 `64` 个 **队列/分区** 时，`Load` 时会发生明显的飙高现象。**队列** 越多，**负载** 越高，发送消息 **响应时间变长**；
2. 使用 **短轮询方式**，**实时性** 取决于 **轮询间隔时间**；
3. 消费失败 **不支持重试**；
4. 支持 **消息顺序**，但是 **一台代理宕机** 后，就会产生 **消息乱序**；
5. 社区更新较慢。
::: info 技术选型
:::
| 特性 | ActiveMQ | RabbitMQ | RocketMQ | Kafka |
| ------------------------ | ------------------------------------- | -------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------- |
| 单机吞吐量 | 万级，比 RocketMQ、Kafka 低一个数量级 | 同 ActiveMQ | 10 万级，支撑高吞吐 | 10 万级，高吞吐，一般配合大数据类的系统来进行实时数据计算、日志采集等场景 |
| topic 数量对吞吐量的影响 | | | topic 可以达到几百/几千的级别，吞吐量会有较小幅度的下降，这是 RocketMQ 的一大优势，在同等机器下，可以支撑大量的 topic | topic 从几十到几百个时候，吞吐量会大幅度下降，在同等机器下，Kafka 尽量保证 topic 数量不要过多，如果要支撑大规模的 topic，需要增加更多的机器资源 |
| 时效性 | ms 级 | 微秒级，这是 RabbitMQ 的一大特点，延迟最低 | ms 级 | 延迟在 ms 级以内 |
| 可用性 | 高，基于主从架构实现高可用 | 同 ActiveMQ | 非常高，分布式架构 | 非常高，分布式，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用 |
| 消息可靠性 | 有较低的概率丢失数据 | 基本不丢 | 经过参数优化配置，可以做到 0 丢失 | 同 RocketMQ |
| 功能支持 | MQ 领域的功能极其完备 | 基于 erlang 开发，并发能力很强，性能极好，延时很低 | MQ 功能较为完善，还是分布式的，扩展性好 | 功能较为简单，主要支持简单的 MQ 功能，在大数据领域的实时计算以及日志采集被大规模使用 |
综上，各种对比之后，有如下建议：
- 一般的业务系统要引入 MQ，最早大家都用 ActiveMQ，但是现在确实大家用的不多了，没经过大规模吞吐量场景的验证，社区也不是很活跃，所以大家还是算了吧，我个人不推荐用这个了；
- 后来大家开始用 RabbitMQ，但是确实 erlang 语言阻止了大量的 Java 工程师去深入研究和掌控它，对公司而言，几乎处于不可控的状态，但是确实人家是开源的，比较稳定的支持，活跃度也高；
- 不过现在确实越来越多的公司会去用 RocketMQ，确实很不错，毕竟是阿里出品，但社区可能有突然黄掉的风险（目前 RocketMQ 已捐给 [Apache](https://github.com/apache/rocketmq)，但 GitHub 上的活跃度其实不算高）对自己公司技术实力有绝对自信的，推荐用 RocketMQ，否则回去老老实实用 RabbitMQ 吧，人家有活跃的开源社区，绝对不会黄。
- 所以**中小型公司**，技术实力较为一般，技术挑战不是特别高，用 RabbitMQ 是不错的选择；**大型公司**，基础架构研发实力较强，用 RocketMQ 是很好的选择。
- 如果是**大数据领域**的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范。
### 【困难】什么是 JMS？
提到 MQ，就顺便提一下 JMS 。
**JMS（JAVA Message Service，java 消息服务）API 是一个消息服务的标准/规范，允许应用程序组件基于 JavaEE 平台创建、发送、接收和读取消息**。它使分布式通信耦合度更低，消息服务更加可靠以及异步性。
在 EJB 架构中，有消息 bean 可以无缝的与 JMS 消息服务集成。在 J2EE 架构模式中，有消息服务者模式，用于实现消息与应用直接的解耦。
::: info JMS 消息模型
:::
在 JMS 标准中，有两种消息模型：
- P2P(Point to Point)
- Pub/Sub(Publish/Subscribe)
**P2P 模式**
P2P 模式包含三个角色：MQ（Queue），发送者 (Sender)，接收者 (Receiver)。每个消息都被发送到一个特定的队列，接收者从队列中获取消息。队列保留着消息，直到他们被消费或超时。
P2P 的特点
- 每个消息只有一个消费者（Consumer）（即一旦被消费，消息就不再在 MQ 中）
- 发送者和接收者之间在时间上没有依赖性，也就是说当发送者发送了消息之后，不管接收者有没有正在运行，它不会影响到消息被发送到队列
- 接收者在成功接收消息之后需向队列应答成功
如果希望发送的每个消息都会被成功处理的话，那么需要 P2P 模式。
**Pub/sub 模式**
包含三个角色主题（Topic），发布者（Publisher），订阅者（Subscriber） 。多个发布者将消息发送到 Topic, 系统将这些消息传递给多个订阅者。
Pub/Sub 的特点
- 每个消息可以有多个消费者
- 发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息。
- 为了消费消息，订阅者必须保持运行的状态。
为了缓和这样严格的时间相关性，JMS 允许订阅者创建一个可持久化的订阅。这样，即使订阅者没有被激活（运行），它也能接收到发布者的消息。
如果希望发送的消息可以不被做任何处理、或者只被一个消息者处理、或者可以被多个消费者处理的话，那么可以采用 Pub/Sub 模型。
::: info JMS 消息消费
:::
在 JMS 中，消息的产生和消费都是异步的。对于消费来说，JMS 的消息者可以通过两种方式来消费消息。
- **同步** - 订阅者或接收者通过 `receive` 方法来接收消息，`receive` 方法在接收到消息之前（或超时之前）将一直阻塞；
- **异步** - 订阅者或接收者可以注册为一个消息监听器。当消息到达之后，系统自动调用监听器的 `onMessage` 方法。
`JNDI` - Java 命名和目录接口，是一种标准的 Java 命名系统接口。可以在网络上查找和访问服务。通过指定一个资源名称，该名称对应于数据库或命名服务中的一个记录，同时返回资源连接建立所必须的信息。
JNDI 在 JMS 中起到查找和访问发送目标或消息来源的作用。
### 【中等】什么是 AMQP？
AMQP（Advanced Message Queuing Protocol）是一种应用层协议，**用于在消息队列系统中定义消息的格式、传输方式和处理机制**。
AMQP 是一个**面向消息的**、**异步传输**的协议，具有高可靠性、可拓展性、跨平台的特性，适合在分布式系统中传输重要数据。它是 RabbitMQ、ActiveMQ 等消息中间件的底层协议。
**核心组件**
| **组件** | **作用** |
| -------------- | ----------------------------------------------- |
| **Connection** | 客户端与消息代理（如 RabbitMQ）的物理连接 |
| **Channel** | 逻辑通信链路（多路复用连接，减少开销） |
| **Exchange** | 消息路由枢纽（支持多种路由策略） |
| **Queue** | 存储消息的容器，消费者从中拉取数据 |
| **Binding** | 定义 Exchange 与 Queue 的映射关系（含路由规则） |
**路由模型**
- **Direct**：精确匹配路由键（如 `order.create`）
- **Fanout**：广播到所有绑定队列（无视路由键）
- **Topic**：通配符匹配（如 `order.*`）
- **Headers**：基于消息头键值匹配（非路由键）
**可靠性保障**
- **消息确认**：消费者手动确认，失败则重入队列
- **持久化**：队列/消息持久化防丢失
- **事务**：支持原子性提交/回滚（批量操作）
**协议对比**
| **协议** | **优势场景** | **局限性** |
| -------- | ---------------------------- | --------------------------- |
| **AMQP** | 企业级应用（强事务、高可靠） | 略重（不适合 IoT 轻量场景） |
| **MQTT** | 物联网（低功耗、低带宽） | 功能简单（无复杂路由） |
| **JMS** | Java 生态集成 | 仅限 Java，跨平台性弱 |
## Kafka
### 【困难】Kafka 为什么性能高？
Kafka 的数据存储在磁盘上，为什么还能这么快？
说 Kafka 很快时，他们通常指的是 Kafka 高效移动大量数据的能力。Kafka 为了提高传输效率，做了很多精妙的设计。
::: info 顺序 I/O（追加写入）
:::
磁盘读写有两种方式：顺序读写或者随机读写。在顺序读写的情况下，磁盘的顺序读写速度和内存接近。因为磁盘是机械结构，每次读写都会寻址写入，其中寻址是一个“机械动作”。Kafka 利用了一种分段式的、只追加 (Append-Only) 的日志，基本上把自身的读写操作限制为**顺序 I/O**，也就使得它在各种存储介质上能有很快的速度。
::: info 零拷贝
:::
Kafka 数据传输是一个从网络到磁盘，再由磁盘到网络的过程。在网络和磁盘之间传输数据时，消除多余的复制是提高效率的关键。**Kafka 利用零拷贝技术来消除传输过程中的多余复制**。
如果不采用零拷贝，Kafka 将数据同步给消费者的大致流程是：
1. 从磁盘加载数据到 os buffer
2. 拷贝数据到 app buffer
3. 再拷贝数据到 socket buffer
4. 接下来，将数据拷贝到网卡 buffer
5. 最后，通过网络传输，将数据发送到消费者
采用零拷贝技术，Kafka 使用 `sendfile()` 系统方法，将数据从 os buffer 直接复制到网卡 buffer。这个过程中，唯一一次复制数据是从 os buffer 到网卡 buffer。这个复制过程是通过 DMA（Direct Memory Access，直接内存访问） 完成的。使用 DMA 时，CPU 不参与，这使得它非常高效。
::: info 其他性能设计
:::
- **页缓存** - Kafka 的数据并不是实时的写入磁盘，它充分利用了现代操作系统分页存储来利用内存提高 I/O 效率。具体来说，就是把磁盘中的数据缓存到内存中，把对磁盘的访问变为对内存的访问。Kafka 接收来自 socket buffer 的网络数据，应用进程不需要中间处理、直接进行持久化时。可以使用 mmap 内存文件映射。
- **压缩** - Kafka 内置了几种压缩算法，并允许定制化压缩算法。通过压缩算法，可以有效减少传输数据的大小，从而提升传输效率。
- **批处理** - Kafka 的 Clients 和 Brokers 会把多条读写的日志记录合并成一个批次，然后才通过网络发送出去。日志记录的批处理通过使用更大的包以及提高带宽效率来摊薄网络往返的开销。
- **分区** - Kafka 将 Topic 分区，每个分区对应一个名为的 Log 的磁盘目录，而 Log 又根据大小，可以分为多个 Log Segment 文件。这种分而治之的策略，使得 Kafka 可以**并发**读，以支撑非常高的吞吐量。此外，Kafka 支持负载均衡机制，将数据分区近似均匀地分配给消费者群组的各个消费者。
### 【困难】ZooKeeper 在 Kafka 中的作用是什么？
ZooKeeper 在 Kafka 中扮演着**核心的协调者角色**，主要负责集群的元数据管理、Broker 协调和状态维护。这些管理机制可以确保集群一致性、高可用性和协调能力。
**Broker 管理**
- **节点注册与存活监控**：每个 Kafka Broker 启动时会在 ZooKeeper 中注册临时节点（`/brokers/ids`）。ZooKeeper 通过心跳机制监控 Broker 存活状态，若 Broker 宕机，临时节点消失，集群会触发重新选举或分区重新分配。
- **Leader 选举**：Kafka 分区的 Leader 副本选举由 ZooKeeper 协调完成（旧版本依赖 ZooKeeper，新版本已逐步迁移至 Kafka 自身协议）。
**Topic 与分区元数据**
- **存储拓扑信息**：Topic 的分区数量、副本分布（`/brokers/topics/[topic]`）等元数据由 ZooKeeper 维护，供所有 Broker 和客户端查询。
- **分区状态同步**：分区 Leader 变更、ISR（In-Sync Replicas）列表更新等操作通过 ZooKeeper 通知其他 Broker。
**控制器（Controller）选举**
Kafka 集群中的某个 Broker 会被选举为 Controller（通过 ZooKeeper 的临时节点竞争），负责分区 Leader 选举、副本分配等关键任务。Controller 故障时，ZooKeeper 会触发重新选举。
**客户端服务发现**
生产者和消费者通过 ZooKeeper（旧版本）或直接通过 Broker（新版本）获取集群的 Broker 列表和 Topic 元数据，以确定数据的读写位置。
**ACL 与配额管理**（可选）
ZooKeeper 存储访问控制列表（ACL）和客户端配额配置，用于权限控制和流量限制。
**KRaft 将替代 ZooKeeper**
**Kafka 2.8+ 版本**开始支持** KRaft 模式**（基于 Raft 协议），逐步弃用 ZooKeeper，将元数据管理和选举逻辑内置于 Kafka 自身，以简化架构并提升性能。但在大多数生产环境中，ZooKeeper 仍广泛使用。
### 【中等】Kafka 为什么要弃用 Zookeeper？
Kafka 弃用 ZooKeeper 主要是为了**简化架构、提升性能、降低运维复杂度**。
**减少外部依赖**
- **架构简化**：ZooKeeper 是独立的外部系统，需额外部署和维护。移除后，Kafka 成为完全自包含的系统，降低部署和运维成本。
- **避免单点风险**：ZooKeeper 本身需要集群化，若出现故障会影响 Kafka 的元数据管理，内嵌治理逻辑可减少此类风险。
**提升扩展性与性能**
- **元数据效率**：ZooKeeper 的写操作（如 Leader 选举）是串行的，可能成为瓶颈。Kafka 内置的 **KRaft 协议**（基于 Raft）支持并行日志写入，显著提升元数据处理速度（如分区扩容、Leader 切换）。
- **降低延迟**：省去与 ZooKeeper 的网络通信，元数据操作（如 Broker 注册、Topic 变更）延迟更低。
**统一元数据管理**
- **一致性模型统一**：ZooKeeper 使用 ZAB 协议，而 Kafka 使用自身的日志复制机制，两者不一致可能导致协调问题。KRaft 模式通过单一协议（Raft）管理所有元数据，逻辑更清晰。
- **简化客户端访问**：旧版客户端需同时连接 Kafka 和 ZooKeeper，新版只需直连 Kafka Broker。
**支持更大规模集群**
**ZooKeeper 的局限性**：ZooKeeper 对节点数量（通常≤7）和 Watcher 数量有限制，影响 Kafka 集群的扩展性。KRaft 模式通过分片和流式元数据传递，支持超大规模集群（如数十万分区）。
**补充说明**
- Kafka 2.8+ 开始实验性支持 KRaft 模式，3.0+ 逐步稳定，但仍兼容 ZooKeeper 模式。
- 完全移除 ZooKeeper 需确保 KRaft 在生产环境中的成熟度（如故障恢复、监控工具链完善）。
### 【中等】Kafka 中如何实现时间轮？
时间轮是用于高效管理和调度大量定时任务的**环形数据结构**，通过时间分片（槽）管理任务，优化调度效率。
**数据结构**
- **环形数组**：每个槽代表一个时间片（如 1 秒），存储双向链表管理的任务。
- **指针移动**：以固定时间步长推进，触发当前槽的任务执行。
**工作原理**
- **任务插入**：根据延迟时间计算槽位（如 `延迟 % 槽数`），插入链表尾部（`O(1)` 复杂度）。
- **任务触发**：指针每移动一格，执行对应槽中所有任务。
**处理长延迟任务**
- **方案 1：轮次（Netty）**：计算轮数（如 `（延迟-1)/槽数`），轮数归零时触发。
- **方案 2：多层次时间轮（Kafka）**
- **层级递进**：高层槽覆盖更大时间范围（如秒→分→时）。
- **降级机制**：任务随时间推移从高层移至底层，保证精度。
**优点**
- **高效性**：插入/删除任务 O(1) 复杂度。
- **低内存开销**：固定槽数，内存占用稳定。
**应用场景**
- 高并发定时任务（如 Netty 的超时检测）。
- 网络服务器（连接/请求超时管理）。
- 分布式系统（节点间任务协调）。
**实际应用**
- **Netty**：`HashedWheelTimer`（单层+轮次）。
- **Kafka**：多层次时间轮+降级。
- **Caffeine Cache**：本地缓存的任务调度。
### 【中等】Kafka 的索引设计有什么亮点？
Kafka 的索引设计通过**稀疏索引 + 二分查找 + 分段存储**，在**查询效率、存储成本、扩展性**之间取得平衡，适合高吞吐、低延迟的消息系统需求。
Kafka 的索引设计（主要涉及**偏移量索引（.index）和时间戳索引（.timeindex）**）具有以下核心优势：
**高效查询（O(1) ~ O(logN) 复杂度）**
- **稀疏索引**：不存储每条消息的索引，而是按一定间隔（如每 4KB 数据）建立索引项，大幅减少索引文件大小。
- **二分查找**：通过索引快速定位消息所在的**物理位置（磁盘文件+偏移量）**，减少全量扫描。
**低存储开销**
- **紧凑结构**：索引文件仅存储**偏移量 + 物理位置**（固定 8 字节/项），占用空间极小。
- **分段存储**：每个日志段（Segment）独立维护索引，避免单一大文件索引的性能瓶颈。
**快速故障恢复**
- **内存映射（MMAP）**：索引文件通过内存映射加速读取，重启时无需全量加载。
- **懒加载**：仅加载活跃分片的索引，减少启动时间。
**支持时间范围查询**：**时间戳索引（.timeindex）**允许按时间戳快速定位消息，适用于日志回溯、监控等场景。
**索引自动更新**：日志压缩（Compaction）或删除（Retention）时，索引同步清理，避免无效查询。
### 【困难】Kafka 处理请求的全流程？
Kafka 采用 **多线程池 + 事件驱动** 模型，核心线程组分工如下：
**网络通信层（1 个 Acceptor + N 个 Processor）**
- **Acceptor 线程**（1 个）：监听 `ServerSocket`，接收客户端连接，轮询分发给 `Processor` 线程。
- **Processor 线程**（默认 3 个，可配置）：每个 `Processor` 维护一个 `Selector`（NIO），负责：
- **读请求**：解析请求数据，放入**共享请求队列**（`RequestChannel`）。
- **写响应**：从 `ResponseQueue` 获取结果，通过 `Socket` 返回客户端。
**请求处理层（KafkaRequestHandlerPool）**
**IO 线程池**（默认 8 个，可配置）从 `RequestChannel` 拉取请求，根据类型调用对应 `API 层` 处理（如 `handleProduceRequest`）。
关键操作：
- **生产请求**：写入 Leader 副本的 `LogSegment`（内存→PageCache→磁盘）。
- **消费请求**：从 `PageCache` 或磁盘读取数据（零拷贝优化）。
**后台线程**
- **Log Cleaner**：日志压缩（Compaction）和删除（Retention）。
- **Replica Manager**：副本同步（ISR）、Leader 选举。
- **Delayed Operation**：处理延迟操作（如 `Produce` 的 ACK 等待）。
**核心设计优势**
- **解耦网络 I/O 与业务处理**：`Processor` 仅负责通信，`Handler` 专注逻辑。
- **无锁队列**：`RequestChannel` 使用 `ConcurrentLinkedQueue`，减少竞争。
- **动态扩展**：可调整 `Processor` 和 `Handler` 线程数适配负载。
### 【困难】Kafka 中如何实现事务机制？
::: info Kafka 事务实现
:::
Kafka 事务是为了实现消息的 **Exactly-Once 语义**，确保消息在生产、传输、消费过程中**仅处理一次**。
**关键组件**
- **事务协调器**：管理事务生命周期，状态持久化到`__transaction_state`主题。
- **幂等生产者**：通过`Producer ID + Sequence Number`避免重复消息。
- **事务性消费**：支持`read_committed`隔离级别，过滤未提交消息。
**工作流程**
- **启事务**：生产者向协调器注册事务
- **写消息**：带事务 ID 的消息写入分区（标记为未提交）
- **定结局**：协调器最终提交（写 Commit 标记）或回滚
- **控消费**：消费者仅读取已提交消息
**设计特点**
- 轻量级事务（非完整 2PC）
- 与高吞吐架构兼容
- 故障自动恢复（通过事务日志）
::: info Kafka 事务 和 RocketMQ 事务的区别
:::
- **Kafka 事务**主要解决 **Exactly-Once 语义**（生产、消费的精确一次处理），适用于 **流处理场景**（如 Flink、Kafka Streams 的状态一致性）。
- **RocketMQ 事务**主要解决 **分布式事务**（如订单+库存的跨系统事务），适用于 **业务事务场景**（如电商、金融的最终一致性）。
**实现机制差异**
| **特性** | **Kafka** | **RocketMQ** |
| ------------------ | -------------------------------- | ------------------------------------- |
| **事务协调者** | 内置 `TransactionCoordinator` | 依赖外部 **事务消息存储**（如 MySQL） |
| **事务状态存储** | 内部 Topic `__transaction_state` | 独立存储（如数据库、ZK） |
| **事务消息可见性** | 支持 `read_committed` 隔离 | 需消费者主动回查（半消息机制） |
| **事务提交方式** | 两阶段提交（2PC） | 本地事务+定时任务（事务回查） |
**应用场景**
- **Kafka**：
- 流处理任务的 **端到端精确一次**（如 Flink+Kafka）。
- 跨分区的原子写入（如多个 Topic 的关联操作）。
- **RocketMQ**：
- 跨系统事务（如 **订单创建+扣库存**）。
- 需要 **业务回查** 的最终一致性场景。
### 【困难】Kafka 控制器如何处理事件？
控制器是 Kafka 集群的**管理中枢**，基于 ZooKeeper/KRaft 选举，其本质是**一个特殊的 Broker**，额外承担集群管理职责。
**控制器选举**
- 每个 Broker 启动时尝试在 ZooKeeper 创建`/controller`临时节点
- 成功创建的 Broker 成为控制器（Leader）
- 其他 Broker 监听该节点，实现故障转移
**事件处理流程**
- **事件捕获**：通过 ZooKeeper Watch 机制监听关键路径：
- `/brokers/ids`（Broker 上下线）
- `/admin/delete_topics`（主题删除）
- `/isr_change_notification`（ISR 变更）
- **事件入队**
- 控制器将事件放入**异步事件队列**（ControllerEventManager）
- 保证事件顺序处理（单线程模型）
- **事件处理**
- **分区再均衡**：触发 Leader 选举，更新 ISR 列表
- **Broker 故障**：
- 将该 Broker 上的 Leader 分区迁移到其他 Broker
- 更新元数据并同步到所有 Broker
- **主题变更**：更新分区分配方案并持久化到 ZooKeeper
- **元数据同步**
- 通过 **UpdateMetadataRequest** 将最新元数据广播给所有 Broker
- Broker 更新本地缓存（MetadataCache）
**关键设计**
- **单线程模型**：避免并发问题（所有事件串行处理）
- **批处理优化**：合并相似事件（如多个分区 Leader 选举）
- **ZooKeeper 解耦**：新版逐渐用 KRaft 替代 ZK 依赖
**容错机制**
- 控制器故障时通过 ZK 重新选举
- 事件处理失败会重试或触发新一轮选举
- 通过 **epoch 机制**防止脑裂（控制器切换时递增 epoch）
**流程图解**
```
ZooKeeper 事件触发
↓
控制器事件队列（ControllerEventManager）
↓
单线程事件处理器
├─ 分区 Leader 选举
├─ ISR 列表更新
├─ 副本重分配
↓
UpdateMetadataRequest 广播
↓
集群元数据最终一致
```
**特点**：高可靠性但存在单点瓶颈（新版 KRaft 改进为分布式控制器）
## RocketMQ
### 【中等】RocketMQ 的事务消息有什么缺点？你还了解过别的事务消息实现吗？
### 【中等】为什么 RocketMQ 不使用 Zookeeper 作为注册中心？
RocketMQ 自研 **NameServer** 是为了：
- 规避 ZooKeeper 的 **性能瓶颈**
- 适应 **大规模分布式队列场景**
- 实现 **简单高效的服务发现**
> 注：阿里内部早期版本曾用 ZK，后因性能问题重构
**设计目标差异**
- **ZooKeeper**：强一致性（CP），适合小规模元数据管理（如 Kafka 的 Controller 选举）
- **RocketMQ**：高可用性（AP），需要支持 **高频读写**（如 Broker 注册、心跳检测）
**性能瓶颈**
| **场景** | **ZooKeeper** | **RocketMQ 选择** |
| -------------------- | ------------------------- | --------------------------------- |
| **注册中心读写频率** | 低频（秒级） | 高频（毫秒级 Broker 心跳） |
| **节点规模** | 适合中小集群（<100 节点） | 支持大规模集群（数千节点） |
| **吞吐量** | 写性能受限（需全局有序） | 自研 NameServer（无强一致性要求） |
**架构简化**
- **NameServer 轻量化**：
- 无选举机制（所有节点平等）
- 最终一致性（通过定时心跳维护状态）
- 无持久化存储（重启后依赖 Broker 重新注册）
- **避免 ZooKeeper 的依赖**：
- 减少运维复杂度（无需单独维护 ZK 集群）
- 降低网络开销（ZK 的 Watcher 机制在大规模下压力大）
**容灾能力**
- **ZooKeeper**
- 集群半数以上节点存活才可用
- 脑裂问题需人工干预
- **NameServer**
- 单节点故障不影响整体服务（无状态）
- 任意存活节点均可提供服务
**业务场景适配**
RocketMQ 的核心需求
- 快速发现 Broker（路由信息）
- 容忍短暂数据不一致（消息队列场景可接受）
- 低延迟响应（Producer/Consumer 频繁拉取路由表）
### 【中等】RocketMQ 的延迟消息是怎么实现的？
RocketMQ 的延迟消息采用 **多级时间轮 + 定时任务扫描** 实现，非实时投递而是延迟触发。
**实现步骤**
1. **消息标记**
- Producer 发送消息时指定 `delayTimeLevel`（如 `3` 表示延迟 10 秒）
- 支持 18 个固定延迟级别（1s/5s/10s/30s/1m...2h）
2. **延迟存储**
- Broker 将延迟消息存入 **专用 Topic**（`SCHEDULE_TOPIC_XXXX`）
- 按延迟级别分队列（如 `delayLevel=3` 的消息存入 `SCHEDULE_TOPIC_XXXX` 的 Queue3）
3. **定时扫描**
- **时间轮算法** 管理延迟队列
- 每秒扫描对应队列，将到期消息 **重新投递** 到目标 Topic
4. **消息投递**
- 到期后，Broker 将消息从延迟 Topic 转移到原始 Topic
- Consumer 正常消费
**关键设计**
| **组件** | **作用** |
| ----------------- | ------------------------------------------ |
| **ScheduleTopic** | 存储所有延迟消息（内部 Topic，对用户透明） |
| **TimerWheel** | 高效触发延迟任务（O(1) 时间复杂度） |
| **定时线程** | 每秒扫描时间轮，将到期消息移出延迟队列 |
**特点**
- **固定延迟级别**：不支持任意时间精度（如 23 秒）
- **投递误差**：±1 秒（依赖扫描间隔）
- **高吞吐**：时间轮算法避免遍历所有消息
**示例代码**
```java
Message msg = new Message("TestTopic", "Hello Delay".getBytes());
// 设置延迟级别 3（对应 10 秒）
msg.setDelayTimeLevel(3);
producer.send(msg);
```
**对比 Kafka**
| **特性** | **RocketMQ** | **Kafka** |
| -------------- | ------------------ | --------------------------------- |
| **延迟精度** | 固定级别（18 种） | 需外部实现（如 Streams 延迟处理） |
| **实现复杂度** | 内置支持 | 需自定义逻辑 |
| **适用场景** | 订单超时、定时通知 | 流处理中的窗口操作 |
**注**：RocketMQ 5.0+ 支持 **定时消息**（精确到毫秒），底层改用时间戳+哈希分片。
### 【中等】RocketMQ 和 Kafka 在架构和功能上有什么区别？
::: info RocketMQ 和 Kafka 架构对比
:::
| **维度** | **RocketMQ** | **Kafka** |
| --------------- | ------------------------------------------------------- | ---------------------------------------------------------------------- |
| **注册中心** | 自研轻量级 **NameServer**（AP 模型） | 依赖 **ZooKeeper**（CP 模型），Kafka 新版本计划用 KRaft 取代 ZooKeeper |
| **Broker 角色** | 主从架构（Master-Slave） | 无主从区分（所有节点平等） |
| **存储模型** | 单个 CommitLog 文件（含所有 Topic 数据） + 消费队列索引 | **主题-分区-段**的层级结构 |
| **消费模式** | 支持 Push/Pull 两种模式 | 仅 Pull 模式 |
::: info RocketMQ 和 Kafka 功能对比
:::
设计差异：
- **RocketMQ**：
- **业务友好**：强事务、延迟消息、过滤等开箱即用
- **轻量可控**：NameServer 简化集群管理
- **Kafka**：
- **流式优先**：高吞吐、分区弹性伸缩
- **生态整合**：与 Flink/Spark 等深度兼容
主要功能差异如下：
| **功能** | **RocketMQ** | **Kafka** |
| ------------ | -------------------------------------- | -------------------------------------- |
| **消息类型** | 普通消息、顺序消息、事务消息、延迟消息 | 主要支持普通消息（需扩展实现其他特性） |
| **事务支持** | 二阶段提交+事务回查（业务级事务） | 精确一次语义（内部消息事务） |
| **延迟消息** | 内置 18 个固定延迟级别 | 需外部实现（如流处理或自定义调度） |
| **消息回溯** | 支持按时间/偏移量回溯 | 仅支持按偏移量回溯 |
| **消息过滤** | 支持给 Topic 打标签，进一步分类 | 仅支持通过 Topic 给消息分类 |
| **死信队列** | 消息消费重试失败多次后，进入死信队列 | Kafka 原生不支持 |
::: info RocketMQ 和 Kafka 性能与扩展对比
:::
| **维度** | **RocketMQ** | **Kafka** |
| ------------ | -------------------- | --------------------------------- |
| **吞吐量** | 单机约 10 万 TPS | 单机可达百万 TPS（更高吞吐） |
| **延迟** | 毫秒级（优化后更低） | 毫秒级（依赖 PageCache） |
| **水平扩展** | 需手动调整队列数 | 分区自动均衡（更易扩展） |
| **数据保留** | 基于时间/大小策略 | 基于时间/大小策略（支持日志压缩） |
::: info RocketMQ 和 Kafka 应用场景与选型
:::
| **场景** | **RocketMQ** | **Kafka** |
| ----------------- | ---------------------- | -------------------------------- |
| **电商/金融业务** | 订单、支付等业务级事务 | 日志收集、流处理（如 Flink） |
| **延迟/定时任务** | 内置支持（如订单超时） | 需外部系统配合 |
| **大规模日志流** | 适合中小规模 | 超大规模日志场景（如大数据管道） |
| **多协议支持** | 兼容 JMS、MQTT 等 | 仅原生协议 |
**总结选择建议**：
- 选 **RocketMQ**：需要 **业务级事务、延迟消息、多协议支持** 的场景
- 选 **Kafka**：需要 **超高吞吐、流处理集成、大规模日志** 的场景
### 【困难】RocketMQ 如何实现事务消息？
RocketMQ 事务消息采用 **"半消息（Half Message） + 事务状态回查"** 的二阶段方案，确保分布式事务的最终一致性。
**关键流程**
1. **第一阶段：发送半消息**
- Producer 发送**半消息**（对 Consumer 不可见）
- Broker 持久化消息并返回确认响应
- Producer 执行**本地事务**（如数据库操作）
2. **第二阶段：事务状态确认**
- **成功**：Producer 提交事务确认 → Broker 将消息改为可消费状态
- **失败**：Producer 回滚 → Broker 丢弃消息
- **超时未决**：Broker 主动回查 Producer 的事务状态（默认每分钟 1 次）
3. **消息投递**
- 仅提交后的消息会被 Consumer 消费
- 支持消息重试和死信队列机制
**核心组件**
- **事务监听器**（TransactionListener）：实现本地事务执行和回查逻辑
- **事务日志存储**：Broker 存储半消息和事务状态（基于文件或 DB）
- **定时回查线程**：处理超时未确认的事务
**特点**
- **最终一致性**：依赖定期回查机制
- **业务耦合**：需实现本地事务和回查接口
- **高可用**：Broker 集群保证消息不丢失
**应用场景**
- 订单支付（扣库存+生成订单）
- 跨系统数据同步
```java
// 示例：事务生产者
TransactionMQProducer producer = new TransactionMQProducer();
producer.setTransactionListener(new TransactionListener() {
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
// 执行本地事务
return LocalTransactionState.COMMIT_MESSAGE;
}
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
// 事务状态回查
return LocalTransactionState.UNKNOW;
}
});
```
## RabbitMQ
### 【简单】RabbitMQ 的 routing key 和 binding key 的最大长度是多少字节？
**长度限制**
- **最大 255 字节**（超限会抛出异常）。
- 适用于 **Routing Key**（生产者指定）和 **Binding Key**（队列绑定交换机时指定）。
**匹配规则（不同交换机类型）**
| **交换机类型** | **匹配方式** | **示例** |
| -------------- | -------------------------------------------- | ------------------------------------ |
| **Direct** | 完全匹配 | `routing_key == binding_key` |
| **Topic** | 通配符匹配（`*` 匹配一个词，`#` 匹配多个词） | `*.order.#` 匹配 `user.order.create` |
| **Headers** | 不依赖 Routing Key，基于消息头键值对匹配 | `x-match: all/any` |
**最佳实践**
- **保持简短**：避免接近 255 字节，提升性能。
- **命名规范**：如 `{服务}.{模块}.{事件}`（例：`user.order.paid`）。
- **Topic 通配符**：合理使用 `*` 和 `#`，避免过度复杂。
> ⚠️ **注意**：Headers 交换机忽略 Routing Key，仅依赖消息头（Headers）匹配。
### 【中等】RabbitMQ 中无法路由的消息会去到哪里？
在 RabbitMQ 中，**无法路由的消息**（即无法被投递到任何队列的消息）的处理方式取决于消息的 **`mandatory` 和 `immediate` 属性**（RabbitMQ 3.0+ 已弃用 `immediate`），具体规则如下：
**默认情况（未设置 `mandatory`）**
- **消息被直接丢弃**（即 "静默丢失"）。
- **生产者无感知**：Broker 不会返回任何通知。
**设置了 `mandatory=true`**
- 若消息无法路由到任何队列，Broker 会通过 **`basic.return`** 方法将消息返回给生产者。
- **生产者需监听返回消息**：
```java
channel.basicPublish("exchange", "routingKey",
new AMQP.BasicProperties.Builder().mandatory(true).build(),
message.getBytes());
// 添加 ReturnListener 监听返回消息
channel.addReturnListener((replyCode, replyText, exchange, routingKey, properties, body) -> {
System.out.println("消息未被路由：" + new String(body));
});
```
- **适用场景**：需严格确保消息路由成功的业务（如关键订单通知）。
**备用交换机（Alternate Exchange）**
- **预先声明一个备用交换机**，绑定一个队列（如 `unrouted_queue`）接收无法路由的消息。
- **配置方式**：
```java
Map args = new HashMap<>();
args.put("alternate-exchange", "my_ae"); // 指定备用交换机
channel.exchangeDeclare("main_exchange", "direct", false, false, args);
// 声明备用交换机和队列
channel.exchangeDeclare("my_ae", "fanout");
channel.queueDeclare("unrouted_queue", false, false, false, null);
channel.queueBind("unrouted_queue", "my_ae", "");
```
- **逻辑**：若消息无法通过 `main_exchange` 路由，则自动转发到 `my_ae`，最终进入 `unrouted_queue`。
**关键区别**
| 处理方式 | 条件 | 结果 | 适用场景 |
| -------------------- | ------------------------- | ---------------------------- | -------------------------- |
| **直接丢弃** | 默认情况 | 消息丢失，无通知 | 允许消息丢失的非关键业务 |
| **返回生产者** | `mandatory=true` | 通过 `basic.return` 回退消息 | 需严格监控路由失败的场景 |
| **转发到备用交换机** | 配置了 Alternate Exchange | 消息存入备用队列 | 需审计或补偿无法路由的消息 |
**最佳实践**
- **关键消息**：始终设置 `mandatory=true` 并监听 `basic.return`。
- **日志与监控**：使用备用交换机收集无法路由的消息，便于排查问题。
- **避免消息丢失**：确保交换机和队列的绑定关系正确，或使用 **死信队列（DLX）** 处理异常消息。
> 📌 **注意**：RabbitMQ 3.0+ 已移除 `immediate` 参数，旧版本中设置 `immediate=true` 会导致无法路由的消息被丢弃（除非同时设置 `mandatory`）。
### 【中等】RabbitMQ 中消息什么时候会进入死信交换机？
通过合理配置 DLX，可以实现消息的优雅降级和故障隔离。
::: info RabbitMQ 中消息进入死信交换机的触发情况
:::
在 RabbitMQ 中，消息进入 **死信交换机（Dead Letter Exchange, DLX）** 通常由以下 **5 种情况**触发：
**（1）消息被消费者拒绝**：消费者显式拒绝消息且不重新入队。
```java
channel.basicReject(deliveryTag, false); // 或 basicNack 且 requeue=false
```
- **典型场景**：消息处理失败且无需重试（如业务校验不通过）。
**（2）消息过期（TTL 超时）**：
- 消息设置了 **TTL（Time-To-Live）**，且未在过期前被消费。
- 队列设置了 `x-message-ttl`，消息在队列中停留超时。
```java
// 设置消息 TTL
AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
.expiration("60000") // 60 秒过期
.build();
channel.basicPublish("", "normal_queue", props, message.getBytes());
```
**（3）队列达到最大长度**：队列设置了 `x-max-length` 或 `x-max-length-bytes`，且新消息到达时队列已满。
```java
// 声明队列时设置最大长度
Map args = new HashMap<>();
args.put("x-max-length", 1000); // 最多 1000 条消息
channel.queueDeclare("normal_queue", false, false, false, args);
```
**（4）队列被删除**：消息所在的队列被删除（`queueDelete`），且消息未被消费。
**（5）主节点崩溃**：**镜像队列（Mirrored Queue）** 中主节点崩溃，且消息未同步到从节点。
::: info 关键配置步骤
:::
1. **声明死信交换机（DLX）和死信队列**：
```java
// 声明死信交换机（类型通常为 direct/fanout）
channel.exchangeDeclare("dlx_exchange", "direct");
// 声明死信队列
channel.queueDeclare("dlx_queue", false, false, false, null);
channel.queueBind("dlx_queue", "dlx_exchange", "dlx_routing_key");
```
````
2. **为普通队列绑定死信交换机**：
```java
Map args = new HashMap<>();
args.put("x-dead-letter-exchange", "dlx_exchange"); // 指定 DLX
args.put("x-dead-letter-routing-key", "dlx_routing_key"); // 可选
channel.queueDeclare("normal_queue", false, false, false, args);
````
**注意事项**
- 死信消息的 **原始属性**（如 headers）会被保留，但 `exchange` 和 `routingKey` 会被替换为 DLX 的配置。
- 若未指定 `x-dead-letter-routing-key`，则使用消息原来的 routing key。
**典型应用场景**
- **延迟队列**：通过 TTL+DLX 实现消息延迟投递。
- **失败处理**：将处理失败的消息自动路由到死信队列，供人工或异步处理。
- **流量控制**：队列满时转移旧消息，避免阻塞新消息。
### 【中等】RabbitMQ 如何实现事务机制？
RabbitMQ 的事务机制（Transaction）通过 **信道（Channel）** 提供了一种保证消息可靠投递的机制，但其设计简单且对性能影响较大。
RabbitMQ 的事务通过同步机制确保消息投递的原子性，但性能代价高。**在绝大多数生产环境中，推荐使用 Publisher Confirms 替代事务**，以兼顾可靠性和吞吐量。
**事务的核心操作**
- **开启事务**：
```java
channel.txSelect(); // 开启事务模式
```
- **提交事务**：
```java
channel.txCommit(); // 提交事务，消息真正投递到队列
```
- **回滚事务**：
```java
channel.txRollback(); // 回滚事务，丢弃未提交的消息
```
**事务的工作流程**
1. 生产者发送消息到 RabbitMQ（消息暂存于信道缓冲区，未写入队列）。
2. 执行 `txCommit()`：消息持久化到队列；若失败或调用 `txRollback()`，消息丢弃。
3. **同步阻塞**：事务提交/回滚需等待 Broker 确认，性能较低。
**事务的局限性**
- **性能差**：每次提交需等待 Broker 确认，吞吐量显著下降（通常降低 100~200 倍）。
- **无分布式事务**：仅保证生产者到 Broker 的可靠性，不涉及消费者或下游系统。
- **不推荐高频使用**：适合低频关键业务，高并发场景建议用 **确认机制（Publisher Confirms）**。
**事务 vs. 确认机制（Publisher Confirms）**
| 特性 | 事务（Transaction） | 确认机制（Publisher Confirms） |
| ------------ | ---------------------- | ------------------------------ |
| **可靠性** | 强一致（同步阻塞） | 最终一致（异步） |
| **性能** | 极低（同步等待） | 高（异步回调） |
| **适用场景** | 低频关键消息（如支付） | 高频业务（如日志、订单） |
| **复杂度** | 简单 | 需处理确认/未确认逻辑 |
**代码示例**
```java
try {
channel.txSelect(); // 开启事务
channel.basicPublish("", "queue1", null, msg1.getBytes());
channel.basicPublish("", "queue2", null, msg2.getBytes());
channel.txCommit(); // 提交事务
} catch (Exception e) {
channel.txRollback(); // 回滚事务
// 处理异常
}
```
**使用建议**
- **优先选择 Confirm 模式**：
```java
channel.confirmSelect(); // 开启确认模式
channel.addConfirmListener(...); // 异步回调
```
- **事务适用场景**：
- 严格保证单批次消息的原子性（如同时投递订单和库存消息）。
- 兼容旧版 RabbitMQ（Confirm 模式需 v3.3+）。
### 【中等】RabbitMQ 中有哪些核心概念？
RabbitMQ 通过 **生产者→交换机→队列→消费者** 的链路传递消息，核心设计围绕 **解耦** 和 **灵活路由**。理解这些角色和概念是掌握其工作原理的基础。
**核心角色**
| **角色** | **作用** |
| ---------------------- | --------------------------------------------------------------------- |
| **生产者（Producer）** | 发送消息到交换机的客户端 |
| **消费者（Consumer）** | 从队列订阅并处理消息的客户端 |
| **消息代理（Broker）** | RabbitMQ 服务实例，负责接收、路由和存储消息（即 RabbitMQ 服务器本身） |
**核心组件**
| **组件** | **功能** |
| ---------------------- | -------------------------------------------------------- |
| **连接（Connection）** | 生产者/消费者与 Broker 之间的 TCP 长连接（复用减少开销） |
| **信道（Channel）** | 连接中的虚拟通道（轻量级，避免频繁创建 TCP 连接） |
| **交换机（Exchange）** | 接收生产者消息，按规则路由到队列（类型决定路由逻辑） |
| **队列（Queue）** | 存储消息的容器，消费者从中拉取数据 |
| **绑定（Binding）** | 定义交换机与队列的映射关系（含路由键规则） |
**辅助概念**
| **概念** | **说明** |
| ------------------------- | -------------------------------------------------------------- |
| **路由键（Routing Key）** | 生产者发送时指定的关键字，用于交换机匹配队列（类似“邮件地址”） |
| **虚拟主机（VHost）** | 逻辑隔离单元（类似命名空间），不同 VHost 的队列/交换机互不可见 |
| **死信队列（DLX）** | 处理过期/被拒绝消息的特殊队列（用于延迟队列或异常处理） |
| **消息确认（ACK）** | 消费者手动确认消息处理完成，确保可靠性 |
**关键协议与机制**
- **AMQP 协议**：RabbitMQ 的核心通信协议，定义消息格式与交互规则
- **持久化（Durable）**：队列/消息可持久化到磁盘，防止服务重启丢失
- **事务（Transaction）**：确保批量操作的原子性（但性能较差，通常用 ACK 替代）
### 【中等】RabbitMQ 有哪些工作模式？
RabbitMQ 有以下几种主要的工作模式：
- 简单模式（Simple）
- 工作队列模式（Work Queue）
- 发布/订阅模式（Publish/Subscribe）
- 路由模式（Routing）
- 主题模式（Topic）
- RPC 模式（远程调用）
以下，对几种工作模式逐一进行说明：
**简单模式（Simple）**
- **角色**：1 生产者 → 1 队列 → 1 消费者
- **特点**：单向通信，无路由逻辑，即点对点模式
- **场景**：单任务处理（如日志记录）
**工作队列模式（Work Queue）**
- **角色**：1 生产者 → 1 队列 → **多个消费者竞争消费**
- **特点**：
- 消息**轮询分发**（默认）或**公平分发**（需设置`prefetch=1`）
- 消费者并行处理
- **场景**：任务分发（如订单处理）
**发布/订阅模式（Publish/Subscribe）**
- **角色**：1 生产者 → **Fanout 交换机** → 绑定多个队列 → 多个消费者
- **特点**：
- 消息**广播**到所有队列
- 消费者各自独立接收全量消息
- **场景**：事件通知（如系统公告）
**路由模式（Routing）**
- **角色**：1 生产者 → **Direct 交换机** → 根据`routing_key`路由到特定队列
- **特点**：
- **精确匹配**路由键
- 支持多队列绑定相同路由键
- **场景**：条件过滤（如错误日志分级处理）
**主题模式（Topic）**
- **角色**：1 生产者 → **Topic 交换机** → 基于通配符（`*`/`#`）匹配路由键
- **特点**：
- **模糊匹配**（如`order.*`匹配`order.create`）
- 灵活性高
- **场景**：复杂路由（如多维度消息分类）
**RPC 模式（远程调用）**
- **角色**：客户端 → 请求队列 → 服务端 → 响应队列 → 客户端
- **特点**：
- 通过`reply_to`和`correlation_id`关联请求/响应
- 同步阻塞式通信
- **场景**：服务间调用（需即时响应）
**模式对比**
| **模式** | **交换机类型** | **路由规则** | **典型应用** |
| --------- | -------------- | --------------------- | ------------ |
| 简单模式 | 无 | 无 | 单任务处理 |
| 工作队列 | 无 | 轮询/公平分发 | 并行任务 |
| 发布/订阅 | Fanout | 广播 | 多系统通知 |
| 路由模式 | Direct | 精确匹配`routing_key` | 条件过滤 |
| 主题模式 | Topic | 通配符匹配 | 复杂路由 |
| RPC 模式 | 无 | 请求-响应关联 | 同步服务调用 |
**选择建议**
- **广播需求** → Fanout
- **条件过滤** → Direct/Topic
- **任务并行** → Work Queue
- **服务调用** → RPC
### 【中等】RabbitMQ 有哪些集群模式？
RabbitMQ 有以下集群模式：
- 普通集群
- 镜像队列集群（高可用模式）
- 联邦集群
- 分片集群
所有集群模式均依赖 **Erlang Cookie** 实现节点间认证，需确保一致。
::: info 普通集群
:::
- **核心特点**
- 元数据（队列、交换机等）**全节点同步**
- **消息实体仅存于创建队列的节点**（其他节点通过指针访问）
- **优点**
- 节省存储（消息不冗余）
- 横向扩展方便
- **缺点**
- **单点故障风险**：若某节点宕机，其上的队列消息不可用
- 跨节点访问消息需网络传输
::: info 镜像队列集群（高可用模式）
:::
- **核心特点**
- 队列**跨节点镜像复制**（消息实体全节点冗余）
- 通过策略（Policy）定义镜像规则（如 `ha-mode=all` 表示全节点复制）
- **优点**
- **高可用**：任一节点宕机，其他节点可继续服务
- 自动故障转移（消费者无感知）
- **缺点**
- **存储开销大**（消息全量复制）
- 写入性能略低（需同步所有副本）
::: info 联邦集群（Federation）
:::
- **核心特点**
- **跨机房/地域**部署，消息按需异步转发
- 基于插件（`rabbitmq_federation`）实现
- **适用场景**
- 异地容灾
- 多区域消息同步
::: info 分片集群（Sharding）
:::
- **核心特点**
- 通过插件（`rabbitmq_sharding`）将队列**水平拆分**到不同节点
- 生产者自动路由到对应分片
- **适用场景**
- 超大规模队列（减轻单节点压力）
::: info 方案对比
:::
| **模式** | **数据冗余** | **高可用** | **跨地域** | **适用场景** |
| -------- | ------------ | ---------- | ---------- | ------------------------ |
| 普通集群 | 无 | ❌ | ❌ | 开发测试、低重要性数据 |
| 镜像队列 | 全量复制 | ✅ | ❌ | 生产环境（如订单、支付） |
| 联邦集群 | 按需同步 | ✅ | ✅ | 异地多活 |
| 分片集群 | 无 | ❌ | ❌ | 超大规模队列 |
**选择建议**
- **生产环境**：优先使用 **镜像队列集群**（需权衡性能与冗余）
- **异地容灾**：结合 **联邦集群** + 镜像队列
- **海量数据**：考虑 **分片集群**（但需业务适配）
### 【中等】RabbitMQ 如何实现延迟队列？
::: info 原生方案：TTL+死信队列（DLX）
:::
- **核心原理**：通过消息 TTL（存活时间）和死信交换机（DLX）实现延迟投递。
- **实现步骤**
1. **创建延迟队列**：设置 `x-message-ttl`（消息过期时间）和 `x-dead-letter-exchange`（死信交换机）
2. **消息投递**：发送到延迟队列，等待 TTL 到期
3. **自动转发**：过期后由 DLX 将消息路由到目标队列
4. **消费者**：从目标队列获取延迟消息
- **特点**
- 固定延迟时间（每条消息 TTL 需单独设置）
- 无需插件，但灵活性较差
::: info 插件方案：rabbitmq_delayed_message_exchange
:::
- **核心原理**：官方插件提供 `x-delayed-message` 交换机类型，支持动态延迟时间。
- **实现步骤**
1. **启用插件**：安装 `rabbitmq_delayed_message_exchange`
2. **声明交换机**：类型设为 `x-delayed-message`，并指定路由规则（如 direct/topic）
3. **发送消息**：通过 `headers` 设置 `x-delay` 参数（毫秒级延迟）
4. **自动投递**：插件内部调度，到期后投递到目标队列
- **特点**
- 支持动态延迟时间（每条消息可独立设置）
- 高精度（毫秒级）
- 需额外安装插件
::: info 方案对比
:::
| **维度** | **TTL+DLX** | **插件方案** |
| ------------ | -------------------------------------------------------- | ---------------------------------- |
| **灵活性** | 固定延迟（队列级别） | 动态延迟（消息级别） |
| **精度** | 秒级 | 毫秒级 |
| **复杂度** | 无需插件，需配置 DLX | 需安装插件 |
| **适用场景** | 简单延迟需求（如统一 30 秒延迟） | 复杂延迟需求（如不同订单超时时间） |
| **缺点** | 队列中消息若阻塞，会延迟后续消息投递（需确保 FIFO 消费） | 大量延迟消息可能占用较高内存 |
**示例代码（插件方案）**
```python
# 声明延迟交换机
channel.exchange_declare(
exchange='delayed_exchange',
exchange_type='x-delayed-message', # 关键参数
arguments={'x-delayed-type': 'direct'}
)
# 发送延迟消息（延迟 5 秒）
channel.basic_publish(
exchange='delayed_exchange',
routing_key='order_queue',
body=message,
properties=pika.BasicProperties(
headers={'x-delay': 5000} # 延迟毫秒数
)
)
```
## 参考资料
- [大型网站架构系列：分布式 MQ（一）](https://www.cnblogs.com/itfly8/p/5155983.html)
- [大型网站架构系列：MQ（二）](https://www.cnblogs.com/itfly8/p/5156155.html)
- [分布式开放 MQ(RocketMQ) 的原理与实践](https://www.jianshu.com/p/453c6e7ff81c)
- [阿里 RocketMQ 优势对比](https://juejin.im/entry/5a0abfb5f265da43062a4a91)
- [advanced-java 之 MQ](https://github.com/doocs/advanced-java/blob/master/docs/high-concurrency/mq-interview.md)
- [浅谈消息队列及常见的消息中间件](https://juejin.im/post/6844903635046924296)