非HTTP请求场景下Trace Context初始化与上下文传递问题求助

问题 #

我们的服务从队列消费消息后启动处理流程（包含大量日志），并通过HTTP客户端调用多个下游服务。目前能为下游HTTP请求生成Trace Context（traceId、txnId），但无法在消息消费环节生成并传递Trace Context，期望在消费消息时立即生成traceId、txnId，用于本服务日志和下游调用追踪。

已尝试以下方案但均无效：

• 使用Hooks.enableAutomaticContextPropagation();并配置contextWrite（该特性需要Reactor core v3.5.3，当前环境无法升级）
• 调用contextWrite(Context.of(TraceContext.class, context));
• 初始化Observation observation = Observation.createNotStarted(...);
• 使用MDC.put('txnId','')

当前环境版本：

• Spring Boot v2.7.14
• Reactor core v3.4.31

附尝试代码：

long generatedLong = new Random().nextLong();

String txnId = StringUtils.lowerCase(RandomStringUtils.randomAlphabetic(16));

final BaggageField baggageField = 
BaggageField.create(TXN_ID_BAGGAGE);
baggageField.updateValue(txnId);

final BaggageFields baggageFields = BaggageFields.newFactory(List.of(baggageField), 1).create();

final TraceContext traceContext = TraceContext.newBuilder().traceId(generatedLong)
 .spanId(generatedLong)
 .parentId(generatedLong)
 .addExtra(baggageFields).build();

return Mono.defer(() -> {
   return validate(request).then(Mono.defer(() -> {
           return service.modify(request)
                  .then(successResponse())
                  .onErrorResume(e -> errorResponse(e));
            })).onErrorResume(e -> errorResponse(e));
        }).contextWrite(Context.of(TraceContext.class, traceContext));

可行方案 #

1. 手动绑定Sleuth Trace Context到Reactor Context #

Spring Boot 2.7.x对应Spring Cloud Sleuth 3.1.x，该版本针对Reactor 3.4.x提供了手动上下文绑定能力，需用Sleuth内置的上下文操作类替代直接写入TraceContext：

// 生成符合Sleuth规范的traceId/spanId（必须为16进制字符串）
String traceId = TraceId.fromBytes(RandomUtils.nextBytes(16)).toString();
String spanId = SpanId.fromBytes(RandomUtils.nextBytes(8)).toString();
String txnId = StringUtils.lowerCase(RandomStringUtils.randomAlphabetic(16));

// 创建Baggage并绑定到TraceContext
BaggageField txnIdField = BaggageField.create("txnId");
TraceContext traceContext = TraceContext.newBuilder()
        .traceId(traceId)
        .spanId(spanId)
        .build();
Span span = DefaultSpan.create(traceContext);
txnIdField.updateValue(span.context(), txnId);

// 绑定Sleuth上下文到Reactor Context，确保下游HTTP调用能读取到Trace信息
return Mono.defer(() -> {
    return validate(request)
            .then(service.modify(request)
                    .then(successResponse())
                    .onErrorResume(this::errorResponse))
            .onErrorResume(this::errorResponse);
})
.contextWrite(SleuthReactor.contextOperator().decorateContextWith(span, Span.current()));

2. 适配MDC日志传递 #

Reactor异步模型下直接调用MDC.put()无法自动传递上下文，需结合MDCContext手动绑定：

// 先填充MDC
MDC.put("traceId", traceId);
MDC.put("txnId", txnId);

// 将MDC上下文绑定到Reactor Context，保证后续异步流程的日志能输出traceId/txnId
return Mono.defer(() -> {
    return validate(request)
            .then(service.modify(request)
                    .then(successResponse())
                    .onErrorResume(this::errorResponse))
            .onErrorResume(this::errorResponse);
})
.contextWrite(MDCContext.of(MDC.getCopyOfContextMap()))
.doFinally(signalType -> MDC.clear()); // 清理MDC避免污染其他请求

3. 完整整合方案 #

将Sleuth上下文绑定与MDC传递结合，同时满足下游调用追踪和本服务日志需求：

String traceId = TraceId.fromBytes(RandomUtils.nextBytes(16)).toString();
String spanId = SpanId.fromBytes(RandomUtils.nextBytes(8)).toString();
String txnId = StringUtils.lowerCase(RandomStringUtils.randomAlphabetic(16));

// 初始化Span和Baggage
BaggageField txnIdField = BaggageField.create("txnId");
TraceContext traceContext = TraceContext.newBuilder()
        .traceId(traceId)
        .spanId(spanId)
        .build();
Span span = DefaultSpan.create(traceContext);
txnIdField.updateValue(span.context(), txnId);

// 填充MDC
MDC.put("traceId", traceId);
MDC.put("txnId", txnId);

return Mono.defer(() -> {
    return validate(request)
            .then(service.modify(request)
                    .then(successResponse())
                    .onErrorResume(this::errorResponse))
            .onErrorResume(this::errorResponse);
})
// 绑定Sleuth上下文到Reactor Context
.contextWrite(SleuthReactor.contextOperator().decorateContextWith(span, Span.current()))
// 绑定MDC上下文
.contextWrite(MDCContext.of(MDC.getCopyOfContextMap()))
// 清理资源
.doFinally(signalType -> {
    MDC.clear();
    Span.current().end(); // 手动结束Span，避免追踪数据泄漏
});

关键注意事项 #

• 禁止用long直接生成traceId/spanId：Sleuth要求traceId为32位16进制字符串、spanId为16位16进制字符串，格式不兼容会导致下游服务无法识别。
• 必须手动结束Span：自行创建的Span需调用Span.current().end()，防止追踪数据堆积。
• 确保HTTP客户端使用Sleuth适配实现：如RestTemplate或WebClient，这类客户端会自动从Reactor Context读取Trace Context并注入到下游请求头中。

内容的提问来源于stack exchange，提问作者nmanandhan