Spring-Webflux通过MDC注入Jaeger TraceId至日志的并发安全性疑问

核心结论 #

直接在WebFilter里用MDC.put存traceId，在高并发场景下绝对不安全，会出现traceId串用、日志与请求不匹配的问题。你测试有效只是因为请求量小、线程未被复用的理想场景，一旦线程池开始复用线程，问题就会暴露。

为什么普通MDC.put不安全？ #

MDC底层是基于ThreadLocal实现的，它的生命周期和线程绑定：

• 传统Servlet模型里，每个请求独占一个线程，请求结束后线程销毁或归池时清理ThreadLocal，所以没问题。
• 但Webflux用的是异步非阻塞模型，依赖Netty的EventLoop线程池（或Reactor的线程池），这些线程会被多个请求复用。如果直接在WebFilter里调用MDC.put("traceId", xxx)，当线程处理完请求A后，traceId会残留在ThreadLocal里；当这个线程接着处理请求B的异步步骤时，日志就会错误地带上请求A的traceId，导致日志混乱。

正确的实现方式 #

要在Webflux中安全传递traceId到MDC，必须结合Reactor Context——它是Reactor框架提供的、跟随请求处理链路的上下文容器，不受线程切换影响。

步骤1：在WebFilter中将traceId存入Reactor Context #

@Component
public class TraceIdWebFilter implements WebFilter {
    private final Tracer jaegerTracer;

    public TraceIdWebFilter(Tracer jaegerTracer) {
        this.jaegerTracer = jaegerTracer;
    }

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        // 从Jaeger当前Span中获取traceId
        String traceId = jaegerTracer.currentSpan().context().traceId();
        // 将traceId绑定到Reactor Context，跟随请求链路传递
        return chain.filter(exchange)
                .contextWrite(context -> context.put("traceId", traceId));
    }
}

步骤2：同步Reactor Context与MDC #

通过Reactor的操作符钩子，在每个请求处理的关键阶段（订阅、事件触发、结束）将Context中的traceId同步到MDC，并在请求结束后清理，避免污染线程：

@Configuration
public class ReactorMdcConfiguration {

    @PostConstruct
    public void registerMdcHook() {
        Hooks.onEachOperator((scannable, operator) -> {
            if (operator instanceof MonoOperator) {
                return new MdcMonoWrapper<>((MonoOperator<?, ?>) operator);
            } else if (operator instanceof FluxOperator) {
                return new MdcFluxWrapper<>((FluxOperator<?, ?>) operator);
            }
            return operator;
        });
    }

    // 包装Mono操作符，同步MDC
    private static class MdcMonoWrapper<T> extends MonoOperator<T, T> {
        public MdcMonoWrapper(MonoOperator<T, T> source) {
            super(source);
        }

        @Override
        public void subscribe(CoreSubscriber<? super T> subscriber) {
            Context context = subscriber.currentContext();
            String traceId = context.getOrDefault("traceId", "");
            // 订阅时将traceId存入MDC
            MDC.put("traceId", traceId);
            
            source.subscribe(new CoreSubscriber<T>() {
                @Override
                public Context currentContext() {
                    return subscriber.currentContext();
                }

                @Override
                public void onSubscribe(Subscription s) {
                    subscriber.onSubscribe(s);
                }

                @Override
                public void onNext(T t) {
                    subscriber.onNext(t);
                }

                @Override
                public void onError(Throwable t) {
                    try {
                        subscriber.onError(t);
                    } finally {
                        // 异常时清理MDC
                        MDC.remove("traceId");
                    }
                }

                @Override
                public void onComplete() {
                    try {
                        subscriber.onComplete();
                    } finally {
                        // 完成时清理MDC
                        MDC.remove("traceId");
                    }
                }
            });
        }
    }

    // 包装Flux操作符，逻辑与Mono一致
    private static class MdcFluxWrapper<T> extends FluxOperator<T, T> {
        public MdcFluxWrapper(FluxOperator<T, T> source) {
            super(source);
        }

        @Override
        public void subscribe(CoreSubscriber<? super T> subscriber) {
            Context context = subscriber.currentContext();
            String traceId = context.getOrDefault("traceId", "");
            MDC.put("traceId", traceId);

            source.subscribe(new CoreSubscriber<T>() {
                @Override
                public Context currentContext() {
                    return subscriber.currentContext();
                }

                @Override
                public void onSubscribe(Subscription s) {
                    subscriber.onSubscribe(s);
                }

                @Override
                public void onNext(T t) {
                    subscriber.onNext(t);
                }

                @Override
                public void onError(Throwable t) {
                    try {
                        subscriber.onError(t);
                    } finally {
                        MDC.remove("traceId");
                    }
                }

                @Override
                public void onComplete() {
                    try {
                        subscriber.onComplete();
                    } finally {
                        MDC.remove("traceId");
                    }
                }
            });
        }
    }
}

简化版实现（无需自定义操作符） #

如果不想写复杂的操作符包装，也可以在WebFilter中通过doOnEach信号处理来同步MDC：

@Override
public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
    String traceId = jaegerTracer.currentSpan().context().traceId();
    return chain.filter(exchange)
            .contextWrite(ctx -> ctx.put("traceId", traceId))
            .doOnEach(signal -> {
                if (signal.hasContext()) {
                    String ctxTraceId = signal.getContext().getOrDefault("traceId", "");
                    if (signal.isOnSubscribe() || signal.isOnNext()) {
                        MDC.put("traceId", ctxTraceId);
                    } else if (signal.isOnError() || signal.isOnComplete()) {
                        MDC.remove("traceId");
                    }
                }
            });
}

关键注意点 #

• 必须在请求处理结束（不管成功还是失败）时清理MDC，否则线程复用后会残留上一个请求的traceId。
• 不要依赖ThreadLocal传递traceId，Reactor Context才是Webflux中上下文传递的标准方式。

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