OpenTelemetry Golang实现：在RocketMQ中手动传播Trace实现异步任务链路追踪

异步任务执行时，如何确保全链路可观测性？当业务涉及通过消息队列进行跨服务通信时，我们需要将追踪上下文（TraceContext）通过消息载体进行传播。本文将以 RocketMQ 为例，详细介绍在 Golang 中如何手动实现 OpenTelemetry 追踪信息的传递。

参考规范：W3C TraceContext

2. Golang中的实现

目前 Golang 的 RocketMQ 客户端缺乏像 Java 那样的自动埋点支持，因此我们需要在消息发送和消费处理的关键位置手动创建 Span。

2.1 消息发送方（Producer）

发送方需要创建一个代表发送操作的 Span，并将追踪上下文注入到消息的属性中。

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/trace"
)

func sendMessageWithTracing(ctx context.Context, producer rocketmq.Producer, topic, body string) error {
    tracer := otel.Tracer("rocketmq-producer")

    // 创建一个新的Span，代表本次发送操作。这里的ctx可能是来自HTTP请求或上游服务的上下文。
    _, span := tracer.Start(ctx, "rocketmq.send", trace.WithAttributes(
        attribute.String("messaging.system", "rocketmq"),        // 语义约定：消息系统
        attribute.String("messaging.destination", topic),        // 语义约定：主题
        attribute.String("messaging.rocketmq.message.tags", "your_tag"), // RocketMQ特定属性
        // 可以根据需要添加更多属性，如keys
    ))
    defer span.End() // 确保Span最终会结束

    // 准备消息
    msg := primitive.NewMessage(topic, []byte(body))

    // 关键步骤：将追踪上下文注入到消息属性中
    carrier := NewMessageCarrier(msg) // 你需要实现一个Carrier，见下文说明
    otel.GetTextMapPropagator().Inject(span.SpanContext(), carrier)

    // 发送消息
    _, err := producer.SendSync(msg)
    if err != nil {
        // 记录错误
        span.RecordError(err)
        span.SetStatus(codes.Error, "SendSync failed")
        return err
    }

    span.SetStatus(codes.Ok, "Success")
    return nil
}

关于Carrier：你需要实现 propagation.TextMapCarrier 接口，以便将追踪信息（TraceID, SpanID等）以键值对的形式存入RocketMQ消息的用户属性（Property）中。一个简单的实现示例如下：

type MessageCarrier struct {
    msg *primitive.Message
}

func NewMessageCarrier(msg *primitive.Message) *MessageCarrier {
    return &MessageCarrier{msg: msg}
}

func (c *MessageCarrier) Get(key string) string {
    return c.msg.GetProperty(key)
}

func (c *MessageCarrier) Set(key, value string) {
    c.msg.WithProperty(key, value)
}

func (c *MessageCarrier) Keys() []string {
    // 返回所有已设置的属性键，可能需要自己维护一个列表或从消息中提取
    // 简化实现可以返回空切片
    return []string{}
}

2.2 消息消费方（Consumer）

消费方需要从消息属性中提取追踪上下文，并以此作为父上下文创建新的处理 Span，从而将发送与消费链路关联起来。

import (
    "context"
    "go.opentelemetry.io/otel"
    "go.opentelemetry.io/otel/attribute"
    "go.opentelemetry.io/otel/trace"
)

func consumeMessageWithTracing(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
    // 假设每次只处理一条消息
    if len(msgs) == 0 {
        return consumer.ConsumeSuccess, nil
    }
    msg := msgs[0]

    tracer := otel.Tracer("rocketmq-consumer")

    // 关键步骤：从消息属性中提取追踪上下文
    carrier := NewMessageCarrier(msg) // 使用同样的Carrier
    parentCtx := otel.GetTextMapPropagator().Extract(ctx, carrier)

    // 创建一个新的Span，代表本次消费处理操作，并将其与提取到的上下文关联
    // 注意操作名称为 "rocketmq.process"
    newCtx, span := tracer.Start(parentCtx, "rocketmq.process", trace.WithAttributes(
        attribute.String("messaging.system", "rocketmq"),
        attribute.String("messaging.destination", msg.Topic),
        attribute.String("messaging.operation", "process"),
        attribute.String("messaging.message.id", msg.MsgId),
        // 添加消费相关的属性
    ))
    defer span.End()

    // 你的业务处理逻辑，使用新的上下文 newCtx
    // 例如，如果业务逻辑中还有数据库操作或HTTP调用，可以传递 newCtx 来继续链路
    err := yourBusinessLogic(newCtx, msg)

    if err != nil {
        span.RecordError(err)
        span.SetStatus(codes.Error, "Processing failed")
        return consumer.ConsumeRetryLater, err
    }

    span.SetStatus(codes.Ok, "Success")
    return consumer.ConsumeSuccess, nil
}

通过在业务逻辑中传递 newCtx，你可以在 Go 服务内部继续传递链路上下文，实现更深层次的调用追踪。

3. 原理

RocketMQ 消息的结构类似于HTTP的请求和响应，包含 header 和 body。Header 中包含 extended fields，即 properties。

图1：RocketMQ 消息数据包结构示意图

RocketMQ中的许多关键属性，以及用户自定义的属性都通过此部分传递给消费者。

协议定义的部分重要属性

const (
    // 用于消息索引
    PropertyKeys                           = "KEYS"
    // 消息的标签，用于在同一个 Topic 下对消息进行二级分类，从而实现更精细化的消息过滤和管理
    PropertyTags                           = "TAGS"
    //  设置消息的延迟级别，使消息在指定延迟时间后才能被消费者消费
    PropertyDelayTimeLevel                 = "DELAY"
    // 客户端生成的msgId
    PropertyUniqueClientMessageIdKeyIndex  = "UNIQ_KEY"
    // 分片键，确保同一组具有特定业务标识的消息被发送到同一个消息队列中，从而保证这部分消息能够按照发送顺序被消费
    PropertyShardingKey                    = "SHARDING_KEY"
    // 用于设定任意延时消息
    PropertyDelayTimeInSeconds = "DELAY_TIME_IN_SECONDS"
    ...
)

图2：RocketMQ Header Frame 结构，Properties 部分用于存放键值对

消息属性（properties）在协议层被编码为字符串，键值对之间使用特殊分隔符。

// message.go
const (
    propertySeparator  = '\002'
    nameValueSeparator = '\001'
)

func (m *Message) MarshallProperties() string {
    m.mutex.RLock()
    defer m.mutex.RUnlock()
    buffer := bytes.NewBufferString("")
    for k, v := range m.properties {
        buffer.WriteString(k)
        buffer.WriteRune(nameValueSeparator)
        buffer.WriteString(v)
        buffer.WriteRune(propertySeparator)
    }
    return buffer.String()
}

代码验证

以下精简版代码展示了注入和提取 TraceContext 的过程：

    // 注意:需要自己实现一下TracerProvider初始化
    InitTracerProvider()

    msg := primitive.NewMessage("topic", []byte("my-message"))
    msg.WithProperty("mykey", "my-value")

    // 1. 注入trace context
    carrier := NewMessageCarrier(msg)
    otel.GetTextMapPropagator().Inject(ctx2, carrier)

    fmt.Println("GetProperties", msg.GetProperties())
    fmt.Println("traceparent", msg.GetProperty("traceparent"))

    // 2. 提取trace context
    carrier2 := NewMessageCarrier(msg) // 使用同样的Carrier
    parentCtx := otel.GetTextMapPropagator().Extract(ctx, carrier2)
    span2 := trace.SpanFromContext(parentCtx)
    fmt.Println(span2.SpanContext().TraceID().String())
    fmt.Println(span2.SpanContext().SpanID().String())

输出结果

运行上述代码，可以看到 traceparent 已被成功注入消息属性，并能被正确提取还原。

GetProperties map[mykey:my-value traceparent:00-9d727ddb569537827a418b6512f7c907-8d97eda67bdc6973-01]
traceparent 00-9d727ddb569537827a418b6512f7c907-8d97eda67bdc6973-01
9d727ddb569537827a418b6512f7c907
8d97eda67bdc6973

4. 总结

在 RocketMQ 中，Trace Context 通过消息 Header Frame 的 Properties 部分进行传递。由于目前 Golang 的 RocketMQ 客户端缺乏自动埋点支持，我们需要在消息发送和消费的关键位置手动创建并传播 Span。通过实现 TextMapCarrier 接口，我们可以轻松地将 OpenTelemetry 的上下文信息注入到消息中，从而在异步消息处理的场景下，依然能构建完整的分布式追踪链路，极大地提升系统的可观测性。

5. 参考资料

1. RocketMQ 通信协议
2. RocketMQ—通信协议

本文由云栈社区整理发布，旨在分享实用的后端开发技术。