vault backup: 2026-05-24 20:51:06
This commit is contained in:
@@ -0,0 +1,178 @@
|
||||
---
|
||||
tags:
|
||||
- MQ
|
||||
- 监控
|
||||
- Prometheus
|
||||
- Grafana
|
||||
create time: 2026-05-24 19:52
|
||||
---
|
||||
|
||||
# MQ 监控指标与告警
|
||||
|
||||
## 概述
|
||||
|
||||
消息队列作为系统间的"黑盒"中间件，一旦出问题往往影响全局。本文梳理 MQ 监控的核心指标体系，涵盖 Broker、Producer、Consumer 三个维度，并介绍基于 Prometheus + Grafana 的监控告警实践。
|
||||
|
||||
## 正文
|
||||
|
||||
### 为什么 MQ 需要专门的监控
|
||||
|
||||
消息队列不像 Web 服务那样有直观的 HTTP 状态码，它的健康状况藏在各种内部指标里。一个 Topic 的消费延迟可能已经到了几小时，但表面上看 Producer 和 Consumer 都在正常运行——只是 Consumer 跟不上了。等到下游业务发现数据不一致时，往往已经积重难返。
|
||||
|
||||
MQ 监控的核心目标就三个字：**看得见**。看得见消息有没有堆积，看得见 Broker 有没有过载，看得见消费者有没有掉队。
|
||||
|
||||
### Broker 核心指标
|
||||
|
||||
| 指标 | 含义 | 关注点 |
|
||||
|------|------|--------|
|
||||
| 消息入队速率 (MessagesIn/s) | 每秒写入的消息数 | 突增可能表示上游流量异常 |
|
||||
| 消息出队速率 (BytesOut/s) | 每秒读出的字节数 | 与入队速率对比判断消费是否跟得上 |
|
||||
| 磁盘使用率 | 日志段占用的磁盘空间 | 超过 80% 就该警惕 |
|
||||
| 网络 IO | 网卡吞吐量 | 高负载时容易成为瓶颈 |
|
||||
| 连接数 | 当前活跃的客户端连接 | 突增可能是连接泄漏 |
|
||||
| ISR 数量 | In-Sync Replicas 数量 | ISR 收缩意味着有 Follower 掉队 |
|
||||
|
||||
> [!question]
|
||||
> ISR（In-Sync Replicas）数量减少时，消息的可靠性会受到什么影响？如果你设置了 `acks=all`，ISR 缩减到 1 会发生什么？
|
||||
|
||||
### Producer 核心指标
|
||||
|
||||
Producer 侧最需要关注的是**发送成功率**和**发送延迟**。
|
||||
|
||||
- **发送成功率**：失败的发送请求占比。如果持续有失败，说明 Broker 端有问题（磁盘满、网络分区）。
|
||||
- **发送延迟 P99**：99 分位的发送耗时。正常情况下应该在毫秒级，如果飙升到秒级，说明 Broker 压力过大或者网络抖动。
|
||||
- **重试率**：发送失败后重试的比例。高重试率意味着消息可能乱序（Kafka 中同一 Partition 内的消息顺序靠 offset 保证，但重试可能导致后发的消息先到）。
|
||||
- **批大小（Batch Size）**：Producer 端的批量发送大小。批越大吞吐越高，但延迟也越大。
|
||||
|
||||
### Consumer 核心指标
|
||||
|
||||
Consumer 侧最核心的指标是 **Consumer Lag**——消费者当前的消费位置与最新消息之间的差距。
|
||||
|
||||
Lag 本质上衡量的是"消费者落后了多远"。如果 Lag 持续增长，说明消费者处理不过来，消息正在积压。
|
||||
|
||||
- **消费速率**：每秒处理的消息数，应与 Producer 的入队速率基本持平。
|
||||
- **Rebalance 次数**：Consumer Group 发生 Rebalance 的频率。频繁 Rebalance 会导致消费暂停，通常是 Consumer 不稳定（频繁重启、处理超时）引起的。
|
||||
- **处理耗时**：单条消息从业务处理的平均耗时。如果处理耗时接近 `max.poll.interval.ms`，就有触发 Rebalance 的风险。
|
||||
|
||||
### Prometheus + Grafana 监控搭建
|
||||
|
||||
监控架构分四层：
|
||||
|
||||
```mermaid
|
||||
graph LR
|
||||
MQ["MQ Cluster"] -->|"指标暴露"| Exporter["MQ Exporter"]
|
||||
Exporter -->|"pull /metrics"| Prometheus["Prometheus"]
|
||||
Prometheus -->|"查询"| Grafana["Grafana"]
|
||||
Prometheus -->|"告警规则"| AlertManager["AlertManager"]
|
||||
AlertManager -->|"通知"| Notify["DingTalk / PagerDuty"]
|
||||
```
|
||||
|
||||
**Exporter 配置**：Kafka 可以用 `kafka_exporter` 或 JMX Exporter 暴露指标。`kafka_exporter` 轻量级，适合快速接入；JMX Exporter 功能更全，但配置更复杂。
|
||||
|
||||
```yaml
|
||||
# Prometheus 配置示例
|
||||
scrape_configs:
|
||||
- job_name: "kafka"
|
||||
static_configs:
|
||||
- targets: ["kafka-exporter:9308"]
|
||||
scrape_interval: 15s
|
||||
```
|
||||
|
||||
**核心 Dashboard 设计**：一个实用的 MQ Dashboard 通常包含以下几个面板：
|
||||
|
||||
1. **总览面板**：集群消息入队/出队速率、总 Lag 数、Broker 存活数。
|
||||
2. **Broker 面板**：每个 Broker 的磁盘使用率、网络 IO、ISR 数量、请求队列深度。
|
||||
3. **Topic 面板**：每个 Topic 的消息速率、Partition 分布、Lag 趋势。
|
||||
4. **Consumer Group 面板**：每个 Group 的 Lag、消费速率、Rebalance 次数。
|
||||
|
||||
### 告警策略
|
||||
|
||||
好的告警策略不是"什么都报"，而是"报了就要行动"。三种常用的告警模型：
|
||||
|
||||
**基于阈值**：最直观，比如 `Consumer Lag > 10000` 触发告警。适合有明确 SLO 的场景。
|
||||
|
||||
**基于趋势**：Lag 的绝对值不重要，重要的是它是否在持续增长。比如"Lag 连续 10 分钟单调递增"就该告警，即使当前 Lag 只有 100。
|
||||
|
||||
**基于异常检测**：用统计方法检测指标是否偏离正常范围。比如消费速率突然降到 0，但 Producer 侧没有任何变化，这很可能是 Consumer 挂了。
|
||||
|
||||
> [!question]
|
||||
> 你能设计一个既不会频繁误报、又不会漏报关键问题的 Consumer Lag 告警规则吗？阈值设多少合适？
|
||||
|
||||
### Go 代码示例：自定义 Consumer Lag 监控
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"net/http"
|
||||
"time"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"github.com/prometheus/client_golang/prometheus/promhttp"
|
||||
)
|
||||
|
||||
var (
|
||||
// consumerLag 每个 Partition 的消费延迟
|
||||
consumerLag = prometheus.NewGaugeVec(
|
||||
prometheus.GaugeOpts{
|
||||
Name: "myapp_consumer_lag",
|
||||
Help: "Consumer lag per partition",
|
||||
},
|
||||
[]string{"topic", "partition", "group"},
|
||||
)
|
||||
)
|
||||
|
||||
func init() {
|
||||
prometheus.MustRegister(consumerLag)
|
||||
}
|
||||
|
||||
// collectLag 定期采集 Consumer Lag 并暴露给 Prometheus
|
||||
func collectLag(brokers []string, group, topic string) {
|
||||
config := sarama.NewConfig()
|
||||
client, err := sarama.NewClient(brokers, config)
|
||||
if err != nil {
|
||||
log.Fatalf("create client: %v", err)
|
||||
}
|
||||
defer client.Close()
|
||||
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
for range ticker.C {
|
||||
partitions, _ := client.Partitions(topic)
|
||||
for _, p := range partitions {
|
||||
// 获取最新 offset
|
||||
newest, _ := client.GetOffset(topic, p, sarama.OffsetNewest)
|
||||
// 获取消费者组当前 offset（实际需通过 Admin API）
|
||||
// 这里简化为从外部获取
|
||||
groupOffset := getGroupOffset(group, topic, p)
|
||||
lag := float64(newest - groupOffset)
|
||||
consumerLag.WithLabelValues(topic,
|
||||
string(rune(p+'0')), group).Set(lag)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func getGroupOffset(group, topic string, partition int32) int64 {
|
||||
// 实际实现中通过 Kafka Admin API 获取
|
||||
return 0
|
||||
}
|
||||
|
||||
func main() {
|
||||
go collectLag([]string{"localhost:9092"}, "my-group", "orders")
|
||||
|
||||
http.Handle("/metrics", promhttp.Handler())
|
||||
log.Println("metrics server on :2112")
|
||||
log.Fatal(http.ListenAndServe(":2112", nil))
|
||||
}
|
||||
```
|
||||
|
||||
这段代码通过 Sarama 客户端定期查询每个 Partition 的最新 offset 和消费者组当前 offset，计算差值作为 Lag，然后通过 Prometheus Gauge 暴露。Grafana 中可以直接用 `myapp_consumer_lag` 指标配置 Dashboard 和告警规则。
|
||||
|
||||
## 关联笔记
|
||||
|
||||
- [[30-MQ-核心概念与选型]]
|
||||
- [[33-MQ-Kafka-架构与核心机制]]
|
||||
- [[37-MQ-消费积压治理]]
|
||||
- [[35-MQ-与-CDC]]
|
||||
@@ -0,0 +1,192 @@
|
||||
---
|
||||
tags:
|
||||
- MQ
|
||||
- 消费积压
|
||||
- 性能优化
|
||||
- Kafka
|
||||
create time: 2026-05-24 19:52
|
||||
---
|
||||
|
||||
# MQ 消费积压治理
|
||||
|
||||
## 概述
|
||||
|
||||
消息积压是 MQ 使用中最常见的"事故"之一：Producer 疯狂写入，Consumer 跟不上节奏，Lag 越堆越高，下游业务开始告警。本文从积压的成因、紧急处理到根本解决方案，系统性地梳理治理思路。
|
||||
|
||||
## 正文
|
||||
|
||||
### 消息积压是怎么产生的
|
||||
|
||||
消息积压的本质很简单：**生产速度 > 消费速度**。但具体原因千差万别：
|
||||
|
||||
1. **Consumer 消费速度跟不上**：业务逻辑变重了（比如新增了一次远程调用），单条消息处理耗时从 10ms 涨到 500ms，消费速率直接腰斩。
|
||||
2. **Consumer 故障**：Consumer 实例挂了、OOM 了、被 K8s 驱逐了，但你不知道——因为消息队列不会主动告诉你"你的消费者已经死了"。
|
||||
3. **Rebalance 风暴**：Consumer Group 频繁 Rebalance，每次 Rebalance 期间所有 Consumer 都暂停消费。如果 Consumer 处理超时触发 Rebalance，然后 Rebalance 又导致更多超时，就会进入恶性循环。
|
||||
4. **流量突增**：大促、秒杀、爬虫攻击，Producer 端流量瞬间翻几倍，但 Consumer 还是原来那点资源。
|
||||
|
||||
### 积压的影响
|
||||
|
||||
积压不是"慢一点"那么简单，它会引发连锁反应：
|
||||
|
||||
- **消息延迟增大**：用户下单后要等几分钟才能收到确认通知，体验极差。
|
||||
- **触发过期删除**：Kafka 的日志保留策略（比如 7 天），如果积压的消息超过了保留时间，还没被消费就被删了，数据丢失。
|
||||
- **下游业务受影响**：如果下游依赖 CDC 事件更新缓存或索引，积压会导致缓存长时间不更新，数据不一致。
|
||||
|
||||
### 紧急处理方案
|
||||
|
||||
发现积压后，第一步不是优化代码，而是先止血。
|
||||
|
||||
**方案一：快速扩容 Consumer 实例**
|
||||
|
||||
这是最直接的办法。Consumer Lag 了 10 万条？加 Consumer 实例就行。但有个前提：**Partition 数量要够**。
|
||||
|
||||
Kafka 中，一个 Partition 最多只能被同一个 Consumer Group 中的一个 Consumer 消费。如果你的 Topic 只有 4 个 Partition，那最多只能有 4 个 Consumer 同时消费，加再多实例也没用。
|
||||
|
||||
> [!question]
|
||||
> 你的 Topic 有 8 个 Partition，当前跑了 3 个 Consumer 实例，Lag 在持续增长。这时候你决定加到 10 个 Consumer，会发生什么？多出来的 2 个在干嘛？
|
||||
|
||||
**方案二：临时 Topic 转发**
|
||||
|
||||
如果 Consumer 的消费逻辑很重（比如涉及数据库写入、远程调用），而且短时间内改不了，可以这样做：
|
||||
|
||||
1. 写一个"轻量消费者"，只做一件事：从积压的 Topic 读消息，批量转发到一个临时 Topic。
|
||||
2. 启动一批新的 Consumer 消费临时 Topic，这些 Consumer 的消费逻辑可以是更轻量的版本（比如只写入一个临时表，不做完整业务处理）。
|
||||
3. 积压消化完后，再慢慢回补业务逻辑。
|
||||
|
||||
这个方案的核心思想是：**先快速消费完，再慢慢处理**。
|
||||
|
||||
**方案三：降级消费**
|
||||
|
||||
如果积压的消息中有关键消息和非关键消息（比如订单消息是关键的，日志消息是非关键的），可以临时修改 Consumer 逻辑，跳过非关键消息，优先处理核心业务。
|
||||
|
||||
```go
|
||||
// 降级消费：跳过非关键消息
|
||||
func handleMsg(msg *kafka.Message) {
|
||||
priority := msg.Headers.Get("priority")
|
||||
if priority == "low" && isBacklogHigh() {
|
||||
// 积压严重时，低优先级消息直接丢弃
|
||||
log.Printf("skip low priority msg: offset=%d", msg.Offset)
|
||||
return
|
||||
}
|
||||
// 正常处理关键消息
|
||||
processBusinessLogic(msg)
|
||||
}
|
||||
```
|
||||
|
||||
> [!question]
|
||||
> 降级消费意味着丢弃部分消息。在什么业务场景下这是可以接受的？你能想到哪些场景绝对不能降级？
|
||||
|
||||
### 根本解决：优化消费逻辑
|
||||
|
||||
紧急处理只是止血，要根治得从消费逻辑入手。
|
||||
|
||||
**减少 IO 操作**：消费逻辑中的数据库写入、远程调用是最常见的瓶颈。检查一下是不是每条消息都触发一次 DB 写入？能不能改成批量写入？
|
||||
|
||||
**异步处理**：消费逻辑中如果有非关键步骤（比如发通知、写日志），可以异步化。消费者只做最关键的操作（比如写主库），其余的丢到 goroutine 或者另一个队列。
|
||||
|
||||
**批量消费**：攒一批消息一起处理，减少网络往返和事务开销。
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/segmentio/kafka-go"
|
||||
)
|
||||
|
||||
func main() {
|
||||
reader := kafka.NewReader(kafka.ReaderConfig{
|
||||
Brokers: []string{"localhost:9092"},
|
||||
Topic: "orders",
|
||||
GroupID: "order-processor",
|
||||
MinBytes: 1, // 最少 1 字节就返回
|
||||
MaxBytes: 10e6, // 最多 10MB
|
||||
CommitInterval: 0, // 手动提交
|
||||
})
|
||||
defer reader.Close()
|
||||
|
||||
batch := make([]kafka.Message, 0, 100)
|
||||
ticker := time.NewTicker(500 * time.Millisecond) // 每 500ms 刷一次
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
if len(batch) == 0 {
|
||||
continue
|
||||
}
|
||||
// 批量写入数据库（一个事务搞定）
|
||||
if err := batchInsert(batch); err != nil {
|
||||
log.Printf("batch insert error: %v", err)
|
||||
continue
|
||||
}
|
||||
// 手动提交 offset
|
||||
if err := reader.CommitMessages(context.Background(), batch[len(batch)-1]); err != nil {
|
||||
log.Printf("commit error: %v", err)
|
||||
}
|
||||
log.Printf("processed batch of %d messages", len(batch))
|
||||
batch = batch[:0] // 清空批次
|
||||
|
||||
default:
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
|
||||
msg, err := reader.ReadMessage(ctx)
|
||||
cancel()
|
||||
if err != nil {
|
||||
continue // 超时，继续攒消息
|
||||
}
|
||||
batch = append(batch, msg)
|
||||
if len(batch) >= 100 {
|
||||
// 批次满了，立即处理
|
||||
ticker.Reset(0) // 触发立即刷入
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func batchInsert(msgs []kafka.Message) error {
|
||||
// 批量插入数据库的逻辑
|
||||
return nil
|
||||
}
|
||||
```
|
||||
|
||||
这段代码展示了批量消费的核心思路：攒够 100 条或者等 500ms，然后一次性写入数据库。相比逐条消费逐条写入，批量处理可以将数据库写入次数降低一到两个数量级。
|
||||
|
||||
**增加 Partition**：如果优化消费逻辑后还是不够，那就增加 Partition 数量，从根本上提升消费并行度。但这需要新建 Topic、迁移数据、修改 Producer 配置，成本不低，应该提前做好容量规划。
|
||||
|
||||
### 预防措施
|
||||
|
||||
与其事后救火，不如事前防火：
|
||||
|
||||
- **容量规划**：根据历史流量峰值，预留 2-3 倍的消费能力。
|
||||
- **压力测试**：上线前用压测工具（如 Kafka 的 `kafka-producer-perf-test`）验证消费能力。
|
||||
- **告警阈值**：设置 Lag 告警，比如 Lag > 5000 触发 P2 告警，Lag > 50000 触发 P1 告警，Lag 连续 5 分钟单调递增触发紧急告警。
|
||||
|
||||
### 积压处理决策流程
|
||||
|
||||
```mermaid
|
||||
flowchart TD
|
||||
A["发现消费积压"] --> B{"积压原因?"}
|
||||
B -->|"Consumer 实例挂了"| C["重启/扩容 Consumer"]
|
||||
B -->|"消费逻辑太慢"| D{"能否快速优化?"}
|
||||
B -->|"流量突增"| E{"Partition 数量够?"}
|
||||
D -->|"能"| F["异步化/批量消费"]
|
||||
D -->|"不能"| G["临时 Topic 转发"]
|
||||
E -->|"够"| C
|
||||
E -->|"不够"| H["降级消费 + 后续扩容 Partition"]
|
||||
C --> I["观察 Lag 趋势"]
|
||||
F --> I
|
||||
G --> I
|
||||
H --> I
|
||||
I -->|"Lag 恢复"| J["复盘 + 优化预案"]
|
||||
I -->|"Lag 未恢复"| B
|
||||
```
|
||||
|
||||
## 关联笔记
|
||||
|
||||
- [[30-MQ-核心概念与选型]]
|
||||
- [[33-MQ-Kafka-架构与核心机制]]
|
||||
- [[36-MQ-监控指标与告警]]
|
||||
- [[35-MQ-与-CDC]]
|
||||
@@ -0,0 +1,188 @@
|
||||
---
|
||||
tags:
|
||||
- MQ
|
||||
- 链路追踪
|
||||
- 可观测性
|
||||
- 分布式系统
|
||||
create time: 2026-05-24 19:52
|
||||
---
|
||||
|
||||
# MQ 消息轨迹与链路追踪
|
||||
|
||||
## 概述
|
||||
|
||||
"消息发了，但对方说没收到。"这句话是分布式系统调试中最让人头疼的开场白。消息轨迹与链路追踪就是为了解决这个问题——让你清楚地知道一条消息从生产到消费的完整生命周期。
|
||||
|
||||
## 正文
|
||||
|
||||
### 为什么需要消息轨迹
|
||||
|
||||
在微服务架构中，一条消息从 Producer 发出，经过 Broker 存储，最终被 Consumer 消费处理。这个链路上任何一个环节出问题，排查起来都非常困难：
|
||||
|
||||
- 消息到底发出去了吗？Broker 收到了吗？
|
||||
- 消息被哪个 Consumer Group 消费了？
|
||||
- 消费成功了还是失败了？失败的原因是什么？
|
||||
- 一条业务消息从 A 服务到 B 服务到 C 服务，整条链路耗时多少？
|
||||
|
||||
传统的日志排查方式效率极低——你需要登录多台机器，grep 关键字，然后手动拼凑时间线。消息轨迹就是把这些信息结构化记录下来，让排查变成"查一下就知道"。
|
||||
|
||||
### 消息唯一标识设计
|
||||
|
||||
要追踪一条消息，首先得能唯一标识它。这里有三个关键标识：
|
||||
|
||||
**MessageID**：消息在 Broker 端的唯一标识。Kafka 用 `Topic + Partition + Offset` 三元组定位一条消息，RocketMQ 有独立的 MessageID。这是技术层面的唯一标识。
|
||||
|
||||
**业务 Key**：业务层面的唯一标识，比如订单号 `order_id`。一条消息可能因为重试产生多个 MessageID，但业务 Key 是不变的。排查问题时，通常从业务 Key 入手。
|
||||
|
||||
**TraceID**：链路追踪层面的标识，贯穿整个调用链。一个 TraceID 可以关联"HTTP 请求 → 消息发送 → 消息消费 → 数据库写入"这条完整链路。
|
||||
|
||||
三者的关系：MessageID 定位消息，业务 Key 定位业务实体，TraceID 串联链路。排查问题时，通常先用业务 Key 找到消息，再用 TraceID 看完整链路。
|
||||
|
||||
> [!question]
|
||||
> 如果一条消息因为消费失败被重试了 3 次，最终成功。从消息轨迹的角度看，这 3 次重试应该怎么记录？是 3 条独立轨迹还是一条轨迹的 3 个消费记录？
|
||||
|
||||
### 消息轨迹的三个阶段
|
||||
|
||||
一条消息的完整生命周期可以分为三个阶段，每个阶段都需要记录关键信息：
|
||||
|
||||
**生产轨迹**：Producer 发送消息时记录。包括发送时间、发送结果（成功/失败）、目标 Topic、消息 Key、消息大小、耗时。
|
||||
|
||||
**存储轨迹**：Broker 接收消息时记录。包括入库时间、分配的 Partition、分配的 Offset、Broker 节点地址。
|
||||
|
||||
**消费轨迹**：Consumer 消费消息时记录。包括消费时间、消费结果（成功/失败/重试）、消费耗时、Consumer 实例标识、处理异常信息。
|
||||
|
||||
```mermaid
|
||||
graph LR
|
||||
Producer["Producer"] -->|"1. 发送消息"| Broker["Broker"]
|
||||
Broker -->|"2. 存储消息"| Storage["Storage"]
|
||||
Broker -->|"3. 投递消息"| Consumer["Consumer"]
|
||||
Producer -.->|"生产轨迹"| TraceDB["Trace Storage"]
|
||||
Broker -.->|"存储轨迹"| TraceDB
|
||||
Consumer -.->|"消费轨迹"| TraceDB
|
||||
TraceDB -->|"查询"| UI["Trace UI"]
|
||||
```
|
||||
|
||||
### RocketMQ 消息轨迹原生支持
|
||||
|
||||
RocketMQ 内置了消息轨迹功能，这是它相比 Kafka 的一个差异化特性。启用方式非常简单：
|
||||
|
||||
```go
|
||||
// Producer 端启用消息轨迹
|
||||
producer, _ := rocketmq.NewProducer(
|
||||
producer.WithNameServer([]string{"127.0.0.1:9876"}),
|
||||
producer.WithTrace(&primitive.TraceConfig{
|
||||
TraceTopic: "RMQ_SYS_TRACE_TOPIC", // 轨迹数据写入的 Topic
|
||||
Access: primitive.Local,
|
||||
}),
|
||||
)
|
||||
```
|
||||
|
||||
RocketMQ 的轨迹数据结构大致如下：
|
||||
|
||||
```go
|
||||
type TraceTransferBean struct {
|
||||
TimeStamp int64 // 时间戳
|
||||
TraceType string // 生产/消费
|
||||
MsgId string // 消息 ID
|
||||
Topic string // Topic
|
||||
MsgType string // 消息类型
|
||||
GroupName string // 消费组
|
||||
Keys string // 业务 Key
|
||||
CostTime int64 // 处理耗时
|
||||
Success bool // 是否成功
|
||||
TraceBeans []TraceBean // 轨迹详情
|
||||
}
|
||||
```
|
||||
|
||||
轨迹数据本身也被当作一条消息，写入一个专门的 Trace Topic。这样做的好处是轨迹记录不会影响主业务流程，轨迹数据也可以像普通消息一样被消费和分析。
|
||||
|
||||
### 链路集成：与分布式追踪系统对接
|
||||
|
||||
消息轨迹解决的是 MQ 维度的问题，但在微服务架构中，一条业务链路往往跨越 HTTP 调用、消息队列、数据库操作等多个维度。这时候需要将 MQ 轨迹集成到分布式追踪系统中。
|
||||
|
||||
主流的分布式追踪系统（Jaeger、Zipkin、SkyWalking）都支持通过 TraceID 串联跨服务的调用链。集成的关键是：**在消息 Header 中传播 TraceID**。
|
||||
|
||||
Producer 发送消息时，将当前 Span 的 TraceID 写入消息 Header：
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
|
||||
"github.com/segmentio/kafka-go"
|
||||
"go.opentelemetry.io/otel"
|
||||
"go.opentelemetry.io/otel/propagation"
|
||||
)
|
||||
|
||||
func sendMessage(ctx context.Context, topic string, value []byte) {
|
||||
// 从当前 Context 提取 TraceID，注入到消息 Header
|
||||
propagator := otel.GetTextMapPropagator()
|
||||
headers := make(map[string]string)
|
||||
propagator.Inject(ctx, propagation.MapCarrier(headers))
|
||||
|
||||
// 将 TraceID 作为消息 Header 发送
|
||||
kafkaHeaders := make([]kafka.Header, 0, len(headers))
|
||||
for k, v := range headers {
|
||||
kafkaHeaders = append(kafkaHeaders, kafka.Header{Key: k, Value: []byte(v)})
|
||||
}
|
||||
|
||||
writer := &kafka.Writer{
|
||||
Addr: kafka.TCP("localhost:9092"),
|
||||
Topic: topic,
|
||||
}
|
||||
defer writer.Close()
|
||||
|
||||
err := writer.WriteMessages(ctx, kafka.Message{
|
||||
Value: value,
|
||||
Headers: kafkaHeaders,
|
||||
})
|
||||
if err != nil {
|
||||
log.Printf("send error: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func consumeMessage(ctx context.Context, msg kafka.Message) {
|
||||
// 从消息 Header 中提取 TraceID，恢复 Trace 上下文
|
||||
propagator := otel.GetTextMapPropagator()
|
||||
headers := make(map[string]string)
|
||||
for _, h := range msg.Headers {
|
||||
headers[h.Key] = string(h.Value)
|
||||
}
|
||||
ctx = propagator.Extract(ctx, propagation.MapCarrier(headers))
|
||||
|
||||
// 在这条 Trace 下记录消费 Span
|
||||
tracer := otel.Tracer("mq-consumer")
|
||||
_, span := tracer.Start(ctx, "consume-message")
|
||||
defer span.End()
|
||||
|
||||
// 业务处理...
|
||||
}
|
||||
```
|
||||
|
||||
这段代码展示了 TraceID 的传播机制：Producer 端通过 OpenTelemetry 的 Propagator 将 TraceID 注入消息 Header，Consumer 端从 Header 中提取 TraceID 并恢复 Trace 上下文。这样，一次 HTTP 请求触发的消息发送和消费，就能在 Jaeger 中看到完整的链路视图。
|
||||
|
||||
### 消息审计日志
|
||||
|
||||
除了技术排查，消息轨迹还有一个重要用途：**审计**。
|
||||
|
||||
在金融、电商等场景中，你需要知道"谁在什么时候发了什么消息"、"谁在什么时候消费了什么消息"。这些信息不仅是排查工具，也是合规要求。
|
||||
|
||||
审计日志通常记录以下信息：
|
||||
|
||||
- 操作人/操作服务的标识
|
||||
- 操作时间
|
||||
- 操作类型（生产/消费/删除）
|
||||
- Topic 和消息 Key
|
||||
- 消息摘要（不是完整内容，通常是前 256 字节 + Hash）
|
||||
|
||||
> [!question]
|
||||
> 消息轨迹会增加额外的存储和性能开销。如果你的系统每天产生 1 亿条消息，每条消息记录 3 条轨迹（生产/存储/消费），每天就有 3 亿条轨迹数据。生产环境应该如何平衡追踪粒度和性能？采样是一个好策略吗？
|
||||
|
||||
## 关联笔记
|
||||
|
||||
- [[30-MQ-核心概念与选型]]
|
||||
- [[33-MQ-Kafka-架构与核心机制]]
|
||||
- [[35-MQ-与-CDC]]
|
||||
- [[36-MQ-监控指标与告警]]
|
||||
@@ -0,0 +1,200 @@
|
||||
---
|
||||
tags:
|
||||
- MQ
|
||||
- Kubernetes
|
||||
- 容器化
|
||||
- 运维
|
||||
create time: 2026-05-24 19:52
|
||||
---
|
||||
|
||||
# MQ 容器化与 K8s 部署
|
||||
|
||||
## 概述
|
||||
|
||||
将消息队列部署到 Kubernetes 上，是有状态服务容器化的核心挑战之一。本文从持久存储、网络标识、有序部署三个维度分析难点，介绍 StatefulSet、Operator 模式、Helm Chart 等主流方案，并以 Strimzi Kafka Operator 为例展示生产级部署实践。
|
||||
|
||||
## 正文
|
||||
|
||||
### 为什么 MQ 上 K8s 是个难题
|
||||
|
||||
Kubernetes 天生为无状态服务设计——Pod 可以随时被杀掉重建、IP 随时变化、多个副本之间没有区别。但 MQ 作为有状态服务，恰恰对这三点都很敏感：
|
||||
|
||||
- **持久存储**：消息必须落盘，Pod 重建后数据不能丢
|
||||
- **网络标识**：Broker 之间需要通过固定地址互相发现（如 Kafka 的 `broker.id` 对应固定端点）
|
||||
- **有序部署**：集群启动/扩容时需要按顺序进行，不能所有 Pod 同时拉起
|
||||
|
||||
> [!question]
|
||||
> 想象一下 Kafka 的 3 个 Broker 同时启动并互相注册，会发生什么？为什么有序启动很重要？
|
||||
|
||||
### StatefulSet：有状态应用的基石
|
||||
|
||||
StatefulSet 是 K8s 专门为有状态服务设计的控制器，解决了 Deployment 的两个核心问题：
|
||||
|
||||
1. **稳定的网络标识**：每个 Pod 拥有固定名称（`kafka-0`、`kafka-1`、`kafka-2`），配合 Headless Service 可以通过 DNS 直接访问
|
||||
2. **稳定的持久存储**：每个 Pod 绑定独立的 PVC（PersistentVolumeClaim），Pod 重建后自动挂载原来的卷
|
||||
|
||||
```yaml
|
||||
# StatefulSet 核心配置片段
|
||||
apiVersion: apps/v1
|
||||
kind: StatefulSet
|
||||
metadata:
|
||||
name: kafka
|
||||
spec:
|
||||
serviceName: kafka-headless # 关联 Headless Service
|
||||
replicas: 3
|
||||
template:
|
||||
spec:
|
||||
containers:
|
||||
- name: kafka
|
||||
volumeMounts:
|
||||
- name: data
|
||||
mountPath: /var/lib/kafka/data
|
||||
volumeClaimTemplates: # 每个 Pod 自动创建独立 PVC
|
||||
- metadata:
|
||||
name: data
|
||||
spec:
|
||||
accessModes: ["ReadWriteOnce"]
|
||||
storageClassName: fast-ssd
|
||||
resources:
|
||||
requests:
|
||||
storage: 100Gi
|
||||
```
|
||||
|
||||
但 StatefulSet 只解决了基础设施层面的问题。**什么时候该扩容、扩容后 Partition 怎么重分配、配置变更怎么滚动生效**——这些业务运维知识它一概不管。于是 Operator 登场了。
|
||||
|
||||
### Operator 模式：把运维知识编码
|
||||
|
||||
Operator = Custom Resource Definition（CRD）+ Custom Controller。核心思想是：**将人类运维专家的知识编码为程序**，让 K8s 自动执行运维操作。
|
||||
|
||||
对于 MQ 来说，Operator 能做到：
|
||||
- 一键部署完整集群（含 ZooKeeper/KRaft）
|
||||
- 滚动升级 Broker 版本（保证零停机）
|
||||
- 自动处理 Partition 重分配
|
||||
- 证书自动签发和轮转
|
||||
|
||||
主流 MQ Operator：
|
||||
|
||||
| MQ | Operator | 维护方 |
|
||||
|----|----------|--------|
|
||||
| Kafka | Strimzi | CNCF |
|
||||
| RocketMQ | rocketmq-operator | Apache |
|
||||
| RabbitMQ | cluster-operator | VMware |
|
||||
| Pulsar | pulsar-helm-chart | Apache |
|
||||
|
||||
### Strimzi Kafka Operator 实战
|
||||
|
||||
Strimzi 是 CNCF 孵化项目，通过 CRD 将 Kafka 集群声明式管理。核心 CRD 有三个：
|
||||
|
||||
- **Kafka**：定义 Kafka 集群拓扑（Broker 数量、存储、配置）
|
||||
- **KafkaTopic**：声明式管理 Topic
|
||||
- **KafkaUser**：声明式管理用户和 ACL 权限
|
||||
|
||||
```yaml
|
||||
# Strimzi Kafka CR 示例：3 节点集群
|
||||
apiVersion: kafka.strimzi.io/v1beta2
|
||||
kind: Kafka
|
||||
metadata:
|
||||
name: my-cluster
|
||||
spec:
|
||||
kafka:
|
||||
version: 3.7.0
|
||||
replicas: 3
|
||||
listeners:
|
||||
- name: plain
|
||||
port: 9092
|
||||
type: internal
|
||||
tls: false
|
||||
- name: tls
|
||||
port: 9093
|
||||
type: internal
|
||||
tls: true
|
||||
config:
|
||||
offsets.topic.replication.factor: 3
|
||||
transaction.state.log.replication.factor: 3
|
||||
transaction.state.log.min.isr: 2
|
||||
default.replication.factor: 3
|
||||
min.insync.replicas: 2
|
||||
storage:
|
||||
type: jbod
|
||||
volumes:
|
||||
- id: 0
|
||||
type: persistent-claim
|
||||
size: 100Gi
|
||||
class: fast-ssd
|
||||
zookeeper:
|
||||
replicas: 3
|
||||
storage:
|
||||
type: persistent-claim
|
||||
size: 20Gi
|
||||
```
|
||||
|
||||
Kafka 集群的完整部署架构如下：
|
||||
|
||||
```mermaid
|
||||
graph TB
|
||||
subgraph K8s["Kubernetes Cluster"]
|
||||
subgraph NS["namespace: kafka"]
|
||||
Operator["Strimzi Operator"]
|
||||
CR["Kafka CR"]
|
||||
Operator -->|"watch & reconcile"| CR
|
||||
CR -->|"create"| STS["StatefulSet"]
|
||||
STS -->|"manage"| Pod0["kafka-0"]
|
||||
STS -->|"manage"| Pod1["kafka-1"]
|
||||
STS -->|"manage"| Pod2["kafka-2"]
|
||||
Pod0 ---|"PVC"| PV0["PV-0 100Gi SSD"]
|
||||
Pod1 ---|"PVC"| PV1["PV-1 100Gi SSD"]
|
||||
Pod2 ---|"PVC"| PV2["PV-2 100Gi SSD"]
|
||||
HS["Headless Service"] -->|"DNS: kafka-0.kafka-headless"| Pod0
|
||||
HS -->|"DNS: kafka-1.kafka-headless"| Pod1
|
||||
HS -->|"DNS: kafka-2.kafka-headless"| Pod2
|
||||
LB["LoadBalancer Service"] -->|"external access"| Pod0
|
||||
LB --> Pod1
|
||||
LB --> Pod2
|
||||
end
|
||||
end
|
||||
Client["Client App"] -->|"produce/consume"| LB
|
||||
```
|
||||
|
||||
### Helm Chart 部署
|
||||
|
||||
对于不想引入 Operator 的轻量场景，Helm Chart 是更简单的选择。Bitnami 提供了完善的 Kafka Helm Chart，核心优势是参数化配置和环境隔离：
|
||||
|
||||
```bash
|
||||
# 安装 Kafka 集群
|
||||
helm install my-kafka bitnami/kafka \
|
||||
--namespace kafka --create-namespace \
|
||||
--set replicaCount=3 \
|
||||
--set persistence.size=100Gi \
|
||||
--set persistence.storageClass=fast-ssd \
|
||||
--set resources.requests.memory=4Gi \
|
||||
--set resources.requests.cpu=2 \
|
||||
--set resources.limits.memory=6Gi \
|
||||
--set resources.limits.cpu=4
|
||||
```
|
||||
|
||||
Helm 适合快速搭建和开发/测试环境，但在生产环境中，Operator 的自动化运维能力（升级、扩容、故障恢复）是 Helm 无法替代的。
|
||||
|
||||
### 资源规划建议
|
||||
|
||||
MQ 在 K8s 上的资源规划需要特别注意：
|
||||
|
||||
- **CPU/Memory**：Kafka Broker 建议至少 2C4G，生产环境推荐 4C8G。请求值和限制值都要设置，避免被驱逐或抢占资源
|
||||
- **存储**：SSD 用于高吞吐场景（低延迟随机读写），HDD 用于大容量归档。通过 StorageClass 区分
|
||||
- **JVM 堆**：Kafka Broker 堆不宜过大（4-6GB 足够），留更多内存给页缓存。推荐 G1GC 或 ZGC
|
||||
- **磁盘调度**：容器内使用 `none`（noop）调度算法，避免与宿主机调度冲突
|
||||
|
||||
### 网络方案
|
||||
|
||||
| 方式 | 适用场景 | 特点 |
|
||||
|------|----------|------|
|
||||
| Headless Service | Broker 间内部通信 | Pod 直连，无负载均衡 |
|
||||
| ClusterIP | 集群内客户端访问 | 默认方式 |
|
||||
| NodePort | 开发测试外部访问 | 端口范围 30000-32767 |
|
||||
| LoadBalancer | 生产外部访问 | 云厂商 LB，费用较高 |
|
||||
| Ingress | HTTP 协议暴露 | MQ 一般不走 HTTP，适用 REST Proxy |
|
||||
|
||||
## 关联笔记
|
||||
|
||||
- [[40-MQ-性能调优]]
|
||||
- [[42-MQ-认证与授权]]
|
||||
- [[43-MQ-加密与审计]]
|
||||
@@ -0,0 +1,180 @@
|
||||
---
|
||||
tags:
|
||||
- MQ
|
||||
- 性能调优
|
||||
- Kafka
|
||||
- JVM
|
||||
create time: 2026-05-24 19:52
|
||||
---
|
||||
|
||||
# MQ 性能调优
|
||||
|
||||
## 概述
|
||||
|
||||
MQ 的性能调优是一个系统工程，涉及 Producer、Broker、Consumer、JVM、操作系统多个层面。本文以 Kafka 为主线，从性能指标出发，逐层拆解调优策略，帮助你建立完整的调优方法论。
|
||||
|
||||
## 正文
|
||||
|
||||
### 性能指标三角
|
||||
|
||||
调优之前先明确目标。MQ 的核心性能指标有三个：
|
||||
|
||||
- **吞吐量（Throughput）**：每秒处理的消息数（msgs/sec）或字节数（MB/sec）
|
||||
- **延迟（Latency）**：消息从发送到被确认/消费的时间，关注 P50、P99、P999
|
||||
- **可用性（Availability）**：系统可正常服务的时间比例
|
||||
|
||||
> [!question]
|
||||
> 吞吐量和延迟往往是矛盾的——批处理提升吞吐但增加延迟。你的业务场景更看重哪个？为什么？
|
||||
|
||||
### Producer 调优
|
||||
|
||||
Producer 是性能瓶颈的第一个关口。核心参数和调优思路：
|
||||
|
||||
**batch.size**（默认 16KB）：批量发送的消息大小上限。增大可以提升吞吐（减少网络往返），但增加内存占用和首次延迟。生产环境建议设为 64KB-1MB。
|
||||
|
||||
**linger.ms**（默认 0）：等待凑满 batch 的时间。默认 0 表示来一条发一条，设为 5-100ms 可以显著提升吞吐。和 batch.size 配合使用——达到任一条件即发送。
|
||||
|
||||
**compression.type**（默认 none）：压缩算法。`lz4` 性价比最高（压缩率和速度均衡），`zstd` 压缩率更好但 CPU 开销略高。压缩能显著减少网络带宽和磁盘占用。
|
||||
|
||||
**acks**：可靠性级别。
|
||||
- `acks=0`：不等确认，最快但可能丢消息
|
||||
- `acks=1`：Leader 确认，平衡选择
|
||||
- `acks=all`：ISR 全部确认，最安全但最慢
|
||||
|
||||
**buffer.memory**（默认 32MB）：Producer 端发送缓冲区大小。高吞吐场景下如果 Broker 响应慢，缓冲区可能打满导致阻塞，可适当增大到 64-128MB。
|
||||
|
||||
### Broker 调优
|
||||
|
||||
Broker 是整个集群的性能中枢。
|
||||
|
||||
**刷盘策略**：Kafka 依赖页缓存和顺序写，刷盘策略直接影响性能。
|
||||
- `flush.messages` 和 `flush.ms` 控制刷盘频率
|
||||
- 生产环境建议**关闭主动刷盘**（设为超大值），依赖操作系统的页缓存和后台刷盘，配合多副本保证数据安全
|
||||
|
||||
**页缓存**：Kafka 的性能秘诀在于大量利用 OS 页缓存。消息写入先进页缓存，异步刷盘。消费者如果跟得上生产者，直接从页缓存读取，接近内存速度。
|
||||
|
||||
**网络线程和 IO 线程**：
|
||||
- `num.network.threads`：处理网络请求的线程数，建议设为 CPU 核数
|
||||
- `num.io.threads`：处理磁盘 IO 的线程数，建议设为 CPU 核数的 2 倍
|
||||
- `num.replica.fetchers`：副本同步线程数，高吞吐场景可增大到 2-4
|
||||
|
||||
### Consumer 调优
|
||||
|
||||
Consumer 的调优重点在于批量拉取和并发处理。
|
||||
|
||||
**fetch.min.bytes**（默认 1）：一次 fetch 请求的最小数据量。增大可以减少网络往返次数，但增加延迟。建议设为 1KB-64KB。
|
||||
|
||||
**fetch.max.wait.ms**（默认 500ms）：等待凑满 fetch.min.bytes 的最大时间。配合 fetch.min.bytes 使用，在吞吐和延迟之间找平衡。
|
||||
|
||||
**max.poll.records**（默认 500）：单次 poll 返回的最大消息数。如果消费逻辑较重（如写数据库），适当减小避免处理超时。
|
||||
|
||||
**并发消费**：单个 Consumer 的吞吐有限，通常通过增加 Consumer 数量（不超过 Partition 数量）来提升并行度。
|
||||
|
||||
> [!question]
|
||||
> Consumer 数量超过 Partition 数量会发生什么？如何在不增加 Partition 的情况下提升消费能力？
|
||||
|
||||
### JVM 调优
|
||||
|
||||
Kafka Broker 运行在 JVM 上，JVM 调优至关重要。
|
||||
|
||||
**堆大小**：不宜过大！Kafka Broker 推荐 4-6GB。堆越大 GC 停顿越长，而且 Kafka 的性能很大程度依赖页缓存——堆占了太多内存，页缓存就被挤压了。
|
||||
|
||||
**GC 选择**：
|
||||
- **G1GC**：成熟稳定，适合 4-8GB 堆。关键参数 `-XX:MaxGCPauseMillis=20`
|
||||
- **ZGC**：超低停顿（<10ms），适合大堆场景，JDK 15+ 生产可用
|
||||
|
||||
**直接内存**：Kafka 的网络传输使用堆外内存（DirectBuffer），需要通过 `-XX:MaxDirectMemorySize` 预留足够空间。
|
||||
|
||||
### 操作系统调优
|
||||
|
||||
操作系统层面的优化往往被忽视，但影响显著：
|
||||
|
||||
- **文件描述符**：Kafka 每个 Partition 对应多个日志段文件，需要大量 fd。建议设置 `ulimit -n 100000` 以上
|
||||
- **TCP 参数**：增大缓冲区 `net.core.rmem_max=16777216`、`net.core.wmem_max=16777216`，启用 TCP 快速打开
|
||||
- **vm.swappiness**：设为 1（非 0），避免系统过度 swap 导致 Broker 性能急剧下降
|
||||
- **磁盘调度算法**：SSD 使用 `none`（noop）或 `mq-deadline`，避免 `cfq` 的高开销
|
||||
|
||||
### 基准测试工具
|
||||
|
||||
调优前后都要用数据说话。常用工具：
|
||||
|
||||
- **kafka-producer-perf-test**：Producer 吞吐和延迟测试
|
||||
- **kafka-consumer-perf-test**：Consumer 吞吐测试
|
||||
- **JMeter**：支持自定义场景的压测框架
|
||||
|
||||
### 性能瓶颈定位决策树
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
Start["性能不达标"] --> CheckWhere{"瓶颈在哪?"}
|
||||
CheckWhere -->|"Producer 端"| P["Producer 调优"]
|
||||
CheckWhere -->|"Broker 端"| B["Broker 调优"]
|
||||
CheckWhere -->|"Consumer 端"| C["Consumer 调优"]
|
||||
P --> P1["增大 batch.size"]
|
||||
P --> P2["增大 linger.ms"]
|
||||
P --> P3["开启压缩"]
|
||||
P --> P4["acks=1"]
|
||||
B --> B1["关闭主动刷盘"]
|
||||
B --> B2["增大网络/IO线程"]
|
||||
B --> B3["JVM 调优"]
|
||||
B --> B4["OS 参数调优"]
|
||||
C --> C1["增大 fetch.min.bytes"]
|
||||
C --> C2["增加并发消费者"]
|
||||
C --> C3["增大 max.poll.records"]
|
||||
```
|
||||
|
||||
### Go 性能测试示例
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
)
|
||||
|
||||
func main() {
|
||||
config := sarama.NewConfig()
|
||||
config.Producer.RequiredAcks = sarama.WaitForLocal
|
||||
config.Producer.Compression = sarama.CompressionLZ4 // 开启压缩
|
||||
config.Producer.Flush.Bytes = 64 * 1024 // batch.size 64KB
|
||||
config.Producer.Flush.Frequency = 10 * time.Millisecond // linger.ms 10
|
||||
|
||||
producer, _ := sarama.NewAsyncProducer([]string{"localhost:9092"}, config)
|
||||
defer producer.Close()
|
||||
|
||||
var count int64
|
||||
msg := &sarama.ProducerMessage{
|
||||
Topic: "benchmark",
|
||||
Value: sarama.StringEncoder("test message payload"),
|
||||
}
|
||||
|
||||
// 10 个 goroutine 并发发送
|
||||
for i := 0; i < 10; i++ {
|
||||
go func() {
|
||||
for {
|
||||
producer.Input() <- msg
|
||||
atomic.AddInt64(&count, 1)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// 每秒输出吞吐量
|
||||
ticker := time.NewTicker(time.Second)
|
||||
for range ticker.C {
|
||||
n := atomic.SwapInt64(&count, 0)
|
||||
fmt.Printf("Throughput: %d msgs/sec\n", n)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
这段代码展示了 Sarama 异步 Producer 的基准测试：开启 LZ4 压缩、64KB 批量、10ms 等待时间，10 个 goroutine 并发发送，每秒统计吞吐量。通过调整参数组合对比不同配置下的性能表现。
|
||||
|
||||
## 关联笔记
|
||||
|
||||
- [[39-MQ-容器化与-K8s-部署]]
|
||||
- [[41-MQ-测试策略]]
|
||||
- [[42-MQ-认证与授权]]
|
||||
@@ -0,0 +1,189 @@
|
||||
---
|
||||
tags:
|
||||
- MQ
|
||||
- 测试
|
||||
- 集成测试
|
||||
- 契约测试
|
||||
create time: 2026-05-24 19:52
|
||||
---
|
||||
|
||||
# MQ 测试策略
|
||||
|
||||
## 概述
|
||||
|
||||
MQ 的测试面临异步性、消息顺序、环境依赖等独特挑战。本文从单元测试到混沌工程，逐层介绍 MQ 测试策略，重点讲解 Testcontainers 集成测试、Pact 契约测试和影子流量等实践方法。
|
||||
|
||||
## 正文
|
||||
|
||||
### MQ 测试的挑战
|
||||
|
||||
和同步的 HTTP API 相比，MQ 测试难在哪里？
|
||||
|
||||
1. **异步性**：消息发送后不立即得到结果，如何验证消息被正确处理？
|
||||
2. **消息顺序**：测试时如何保证消息的顺序和幂等性？
|
||||
3. **环境依赖**：没有真实的 Broker，很多行为无法验证（如 Partition 路由、Consumer Group Rebalance）
|
||||
4. **分布式一致性**：跨多个服务的事件流转，如何端到端验证？
|
||||
|
||||
> [!question]
|
||||
> 传统 API 测试可以用 `assert response.status == 200`。但 MQ 中，消息发出去后"成功"意味着什么？是 Broker 收到？还是被消费者处理完？
|
||||
|
||||
### 测试金字塔在 MQ 场景的应用
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
subgraph Pyramid["MQ 测试金字塔"]
|
||||
E2E["E2E 测试 - 完整集群 + 多服务"]
|
||||
Integration["集成测试 - Testcontainers 真实 Broker"]
|
||||
Contract["契约测试 - Pact 消息契约"]
|
||||
Unit["单元测试 - Mock Producer/Consumer"]
|
||||
end
|
||||
Unit -->|"快速, 大量"| Contract
|
||||
Contract -->|"契约保证兼容"| Integration
|
||||
Integration -->|"真实环境验证"| E2E
|
||||
```
|
||||
|
||||
### 单元测试：Mock 接口，验证逻辑
|
||||
|
||||
单元测试的核心是将消息处理逻辑与 Broker 解耦。在 Go 中，通过接口抽象 Producer 和 Consumer：
|
||||
|
||||
```go
|
||||
// 定义接口，方便 Mock
|
||||
type MessageProducer interface {
|
||||
Send(ctx context.Context, topic string, key, value []byte) error
|
||||
}
|
||||
|
||||
type OrderHandler struct {
|
||||
producer MessageProducer
|
||||
}
|
||||
|
||||
// 处理订单创建事件的业务逻辑
|
||||
func (h *OrderHandler) HandleOrderCreated(ctx context.Context, event OrderEvent) error {
|
||||
if event.Amount <= 0 {
|
||||
return fmt.Errorf("invalid amount: %d", event.Amount)
|
||||
}
|
||||
// 发送支付请求事件
|
||||
payment := PaymentEvent{OrderID: event.ID, Amount: event.Amount}
|
||||
data, _ := json.Marshal(payment)
|
||||
return h.producer.Send(ctx, "payment-requests", []byte(event.ID), data)
|
||||
}
|
||||
|
||||
// 单元测试：Mock Producer 验证业务逻辑
|
||||
func TestOrderHandler_HandleOrderCreated(t *testing.T) {
|
||||
mock := &MockProducer{} // 实现 MessageProducer 接口
|
||||
handler := &OrderHandler{producer: mock}
|
||||
|
||||
event := OrderEvent{ID: "order-1", Amount: 100}
|
||||
err := handler.HandleOrderCreated(context.Background(), event)
|
||||
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, "payment-requests", mock.LastTopic())
|
||||
assert.Equal(t, "order-1", mock.LastKey())
|
||||
}
|
||||
```
|
||||
|
||||
单元测试快速且无外部依赖，但只能验证业务逻辑，无法保证消息格式与下游兼容。
|
||||
|
||||
### 集成测试：Testcontainers 启动真实 Broker
|
||||
|
||||
Mock 测试不到的问题——Partition 路由、序列化、Consumer Group 行为——需要真实 Broker。Testcontainers 在测试中启动真实的 Docker 容器：
|
||||
|
||||
```go
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/IBM/sarama"
|
||||
"github.com/testcontainers/testcontainers-go"
|
||||
"github.com/testcontainers/testcontainers-go/modules/kafka"
|
||||
)
|
||||
|
||||
func TestKafkaIntegration(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
|
||||
// 启动真实的 Kafka 容器
|
||||
kafkaContainer, err := kafka.Run(ctx,
|
||||
"confluentinc/confluent-local:7.5.0",
|
||||
)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer kafkaContainer.Terminate(ctx)
|
||||
|
||||
// 获取 Broker 地址
|
||||
brokers, _ := kafkaContainer.Brokers(ctx)
|
||||
|
||||
// 创建 Producer 发送消息
|
||||
config := sarama.NewConfig()
|
||||
config.Producer.Return.Successes = true
|
||||
producer, _ := sarama.NewSyncProducer(brokers, config)
|
||||
defer producer.Close()
|
||||
|
||||
msg := &sarama.ProducerMessage{
|
||||
Topic: "test-topic",
|
||||
Value: sarama.StringEncoder("hello kafka"),
|
||||
}
|
||||
partition, offset, err := producer.SendMessage(msg)
|
||||
if err != nil {
|
||||
t.Fatalf("send failed: %v", err)
|
||||
}
|
||||
|
||||
// 创建 Consumer 消费并验证
|
||||
config2 := sarama.NewConfig()
|
||||
config2.Consumer.Offsets.Initial = sarama.OffsetOldest
|
||||
consumer, _ := sarama.NewConsumer(brokers, config2)
|
||||
defer consumer.Close()
|
||||
|
||||
pc, _ := consumer.ConsumePartition("test-topic", partition, offset)
|
||||
select {
|
||||
case msg := <-pc.Messages():
|
||||
if string(msg.Value) != "hello kafka" {
|
||||
t.Errorf("unexpected message: %s", msg.Value)
|
||||
}
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("timeout waiting for message")
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
这段测试用 Testcontainers 自动拉起 Kafka 容器，发送一条消息后立即消费验证。整个过程在 CI 中无需预置环境，测试结束自动清理容器。
|
||||
|
||||
### 消息契约测试
|
||||
|
||||
在事件驱动架构中，Producer 和 Consumer 通过消息格式（Schema）解耦。但如果 Producer 改了字段名，Consumer 就会崩溃。**契约测试**解决这个问题。
|
||||
|
||||
Pact 框架的工作流程：
|
||||
1. **Consumer 端**定义期望的消息格式（契约）
|
||||
2. **Producer 端**验证自己产生的消息满足所有 Consumer 的契约
|
||||
3. 契约存储在 Pact Broker 中，CI 中自动验证
|
||||
|
||||
> [!question]
|
||||
> 消息契约测试和传统的 API 测试有什么本质区别？为什么在事件驱动架构中更重要？
|
||||
|
||||
关键区别在于：API 测试验证的是"请求-响应"的一对一关系，而消息契约验证的是"事件-消费者"的一对多关系。一个事件可能被 5 个服务消费，任何格式变更都必须向后兼容。
|
||||
|
||||
### 影子流量（Shadow Testing）
|
||||
|
||||
将生产流量复制到测试环境，验证新版本的消息处理逻辑是否正确，而不影响真实业务。
|
||||
|
||||
实现方式：
|
||||
- Producer 端使用拦截器，将消息副本发送到 Shadow Topic
|
||||
- Shadow Consumer 消费并处理，对比结果
|
||||
- 关键：Shadow 消费者的处理结果**不会写入生产数据库**
|
||||
|
||||
### 故障注入
|
||||
|
||||
Chaos Engineering 在 MQ 场景的应用：
|
||||
|
||||
- **Kill Broker**：随机杀掉一个 Broker，验证 Producer/Consumer 的自动故障转移
|
||||
- **网络分区**：模拟 Broker 之间网络隔离，验证脑裂防护
|
||||
- **消息延迟注入**：人为增加 Broker 响应延迟，验证超时和重试机制
|
||||
- **磁盘满**：模拟磁盘空间不足，验证告警和降级策略
|
||||
|
||||
## 关联笔记
|
||||
|
||||
- [[40-MQ-性能调优]]
|
||||
- [[42-MQ-认证与授权]]
|
||||
- [[39-MQ-容器化与-K8s-部署]]
|
||||
Reference in New Issue
Block a user