cs-note/hhs/MQ/10-监控与运维/38-MQ-消息轨迹与链路追踪.md

---
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-监控指标与告警]]