189 lines
7.4 KiB
Markdown
189 lines
7.4 KiB
Markdown
---
|
||
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-监控指标与告警]]
|