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

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) }