如何解决Spring WebClient响应式链路追踪中BaggageField值覆盖问题？

场景与问题描述 #

在Spring WebFlux环境中，使用Spring Cloud Sleuth的Baggage字段传递accountId时，出现异步IO流程上下文串扰问题：

• 调用eventService.send("account1")后启动IO流程；在第一个请求IO结束前，调用eventService.send("account2")修改了Baggage值
• 第一个请求返回错误时，错误日志中的accountId被错误标记为account2，而非实际的account1

相关代码 #

EventService的send方法

private Mono<ApiResponse> send(String accountId) {
    accountIdBaggageField.updateValue(accountId);
    logger.info("making the call");
    Mono<ApiResponse> res = apiClient.dispatchEvent(accountId);
    return res.doOnError(e -> {
        logger.error("Could not dispatch batch for events");
    });
}

Sleuth配置类

@Configuration
public class SleuthConfiguration {

    @Bean
    public BaggageField accountIdBaggageField() {
        return BaggageField.create(LoggingContextVariables.MDC_ACCOUNT_ID);
    }

    @Bean
    public BaggagePropagationCustomizer baggagePropagationCustomizer(BaggageField accountIdBaggageField) {
        return factoryBuilder -> {
            factoryBuilder.add(remote(accountIdBaggageField));
        };
    }

    @Bean
    public CorrelationScopeCustomizer correlationScopeCustomizer(BaggageField accountIdBaggageField) {
        return builder -> {
            builder.add(createCorrelationScopeConfig(accountIdBaggageField));
        };
    }

    private CorrelationScopeConfig createCorrelationScopeConfig(BaggageField field) {
        return CorrelationScopeConfig.SingleCorrelationField.newBuilder(field)
                .flushOnUpdate()
                .build();
    }
}

ApiClient的dispatchEvent方法

public Mono<ApiResponse> dispatchEvent(String accountId) {
        return webClient
                .post()
                .uri(properties.getEndpoints().getDispatchEvent(), Map.of("accountId", accountId))
                .retrieve()
                .onStatus(HttpStatus::isError, this::constructException)
                .bodyToMono(ApiResponse.class)
                .onErrorMap(WebClientRequestException.class, e -> new CGWException("Error during dispatching event to Connector Gateway", e));
    }

调用方式

eventService.send("account1");
eventService.send("account2");

异常日志

2023-01-09 11:50:56.791  INFO [account1] [Thread-1] c.t.e.s.impl.EventServiceImpl    making the call

2023-01-09 11:50:56.812  INFO [account2] [Thread-1] c.t.e.s.impl.EventServiceImpl    making the call

2023-01-09 11:50:58.241  INFO [account2] [reactor-http-nio-4] c.t.e.s.impl.EventServiceImpl    Could not dispatch batch for events

2023-01-09 11:50:58.281  INFO [account2] [reactor-http-nio-6] c.t.e.s.impl.EventServiceImpl    Could not dispatch batch for events

解决方案 #

核心是将Baggage上下文绑定到Reactor的异步序列中，确保每个请求的上下文独立隔离，避免外部线程修改导致串扰。

方案1：使用BaggageField的wrap方法包裹Mono链 #

private Mono<ApiResponse> send(String accountId) {
    // 用wrap方法将整个异步流程包裹，绑定当前accountId到上下文
    return accountIdBaggageField.wrap(
        Mono.defer(() -> {
            logger.info("making the call");
            return apiClient.dispatchEvent(accountId);
        })
    ).doOnError(e -> {
        logger.error("Could not dispatch batch for events");
    });
}

方案2：结合Reactor Context传递上下文 #

private Mono<ApiResponse> send(String accountId) {
    return Mono.defer(() -> {
        accountIdBaggageField.updateValue(accountId);
        logger.info("making the call");
        return apiClient.dispatchEvent(accountId);
    })
    // 将accountId写入Reactor Context，确保异步流程上下文传递
    .contextWrite(ctx -> ctx.put(accountIdBaggageField.name(), accountId))
    .doOnError(e -> {
        logger.error("Could not dispatch batch for events");
    });
}

原因说明 #

• 原代码中accountIdBaggageField.updateValue(accountId)在调用线程（Thread-1）执行，属于Reactor序列外部操作，后续异步IO线程会读取全局的Baggage值，当第二个请求修改Baggage后，第一个请求的错误处理会读取到最新值，导致串扰。
• 使用wrap方法或contextWrite可以将Baggage上下文绑定到当前Mono的执行链中，确保每个异步请求的上下文独立，不会被其他请求覆盖。
• 原Sleuth配置中的flushOnUpdate()是正确的，关键是要保证Baggage的修改操作在Reactor序列内部执行。

内容的提问来源于stack exchange，提问作者Süleyman Gezsat