TraceID 跨越异步任务：不要让消息队列切断调用链

一、一个订单从创建到发货，TraceID 断在了 Kafka

分布式追踪在同步调用中很好用。
HTTP 请求自动携带 TraceID 头，网关注入，逐服务传递。
但一旦进入异步链路，追踪就断了。

订单服务把消息扔进 Kafka。
消费端启动新的 goroutine 处理。
TraceID 存在 HTTP 请求的 context 里。
消息消费者拿不到这个 context。
新生成的 Span 没有父 Span，调用链在 Kafka 处断裂。

排查问题时，只能靠时间戳和订单号人工关联。
一条日志链跨越了五个服务、两个消息队列。
人工拼接花了 40 分钟。
如果有完整的 TraceID，一次查询就能定位。

二、消息中间件的追踪上下文传播机制

分布式追踪的核心是上下文传播。
在同步调用中，W3C Trace Context 标准定义了 traceparent 头。
异步场景需要显式地将追踪信息注入消息体或消息头。

sequenceDiagram participant P as Producer (订单服务) participant K as Kafka participant C as Consumer (物流服务) participant T as Tracing Backend P->>P: 生成 TraceID + SpanID P->>T: 📝 Span: order.create P->>P: 将 traceparent 写入消息头 P->>K: produce(message + headers) Note over K: 消息暂存 C->>K: consume(message) C->>C: 从消息头提取 traceparent C->>C: 恢复 Context C->>T: 📝 Span: logistics.dispatch (父Span=order.create) C->>C: 处理物流逻辑

关键点：消息头是上下文传播的载体。
Kafka、RabbitMQ、Pulsar 都支持消息头。
只需在生产端注入，消费端提取即可。

三、Go 实现的 TraceID 传播

package main import ( "context" "crypto/rand" "encoding/hex" "fmt" "log" "sync" "time" "github.com/segmentio/kafka-go" ) // TraceContext 存储分布式追踪信息 type TraceContext struct { TraceID string `json:"trace_id"` SpanID string `json:"span_id"` ParentSpan string `json:"parent_span,omitempty"` } // NewTraceContext 生成新的追踪上下文 func NewTraceContext() *TraceContext { return &TraceContext{ TraceID: generateID(16), SpanID: generateID(8), } } // NewChildSpan 创建子 Span func (tc *TraceContext) NewChildSpan() *TraceContext { return &TraceContext{ TraceID: tc.TraceID, SpanID: generateID(8), ParentSpan: tc.SpanID, } } // ToHeaders 序列化为 Kafka 消息头 func (tc *TraceContext) ToHeaders() []kafka.Header { return []kafka.Header{ {Key: "trace-id", Value: []byte(tc.TraceID)}, {Key: "span-id", Value: []byte(tc.SpanID)}, {Key: "parent-span", Value: []byte(tc.ParentSpan)}, } } // FromHeaders 从 Kafka 消息头恢复上下文 func FromHeaders(headers []kafka.Header) *TraceContext { tc := &TraceContext{} for _, h := range headers { switch h.Key { case "trace-id": tc.TraceID = string(h.Value) case "span-id": tc.SpanID = string(h.Value) case "parent-span": tc.ParentSpan = string(h.Value) } } if tc.TraceID == "" { return nil } return tc } // InjectContext 将 TraceContext 注入 context func (tc *TraceContext) InjectContext(ctx context.Context) context.Context { return context.WithValue(ctx, "trace_context", tc) } // ExtractContext 从 context 提取 TraceContext func ExtractContext(ctx context.Context) *TraceContext { if tc, ok := ctx.Value("trace_context").(*TraceContext); ok { return tc } return nil } // ---- 生产者 ---- func produceOrder(ctx context.Context, writer *kafka.Writer, orderID string) error { // 从当前 context 获取追踪信息，或创建新的 tc := ExtractContext(ctx) if tc == nil { tc = NewTraceContext() } childSpan := tc.NewChildSpan() // 记录 Span：order.create log.Printf("[%s/%s] 创建订单 %s", childSpan.TraceID, childSpan.SpanID, orderID) msg := kafka.Message{ Topic: "orders", Key: []byte(orderID), Value: []byte(fmt.Sprintf(`{"order_id":"%s"}`, orderID)), Headers: childSpan.ToHeaders(), // ← 注入追踪头 } if err := writer.WriteMessages(ctx, msg); err != nil { log.Printf("[%s] 发送消息失败: %v", childSpan.TraceID, err) return fmt.Errorf("kafka write: %w", err) } return nil } // ---- 消费者 ---- func consumeOrders(ctx context.Context, reader *kafka.Reader) { for { msg, err := reader.ReadMessage(ctx) if err != nil { if ctx.Err() != nil { return // 正常退出 } log.Printf("消费消息失败: %v", err) continue } // 从消息头恢复追踪上下文 tc := FromHeaders(msg.Headers) if tc == nil { tc = NewTraceContext() // 兜底：生成新 TraceID log.Printf("警告: 消息无追踪信息，生成新 TraceID: %s", tc.TraceID) } // 注入 context，传递至下游 spanCtx := tc.InjectContext(context.Background()) processOrder(spanCtx, string(msg.Value)) } } func processOrder(ctx context.Context, payload string) { tc := ExtractContext(ctx) childSpan := tc.NewChildSpan() log.Printf("[%s/%s] 处理物流配送: %s", childSpan.TraceID, childSpan.SpanID, payload) // 这里继续异步调用时，同样注入追踪头 } func generateID(byteLen int) string { b := make([]byte, byteLen) rand.Read(b) return hex.EncodeToString(b) }

四、异步追踪的额外注意事项

消息队列引入了天然的延迟。
Span 的开始时间应该是消息生产时间，而非消费时间。
否则延迟看起来是消费者慢，实则是队列堆积。

批量消费时，每个消息应使用独立的 Span。
如果共享 Span，无法区分单条消息的处理耗时。

追踪数据爆炸问题需要关注。
高频消息场景（每秒万条），每条消息生成 Span 成本高。
建议对非核心链路做采样：每 100 条记录 1 条。

不适合全量追踪的场景：
日志类消息（不参与核心业务链路）；
心跳和监控数据；
延迟要求极低（< 1ms）的高频交易管道。

五、总结

TraceID 通过消息头跨越异步边界。
生产端将追踪上下文注入消息头，消费端提取并恢复。
Kafka、RabbitMQ 等主流中间件都支持消息头传递。
异步场景还需注意 Span 时间语义和采样策略。
完整的异步链路追踪，把"靠时间戳猜"变成了"一次查询定位"。