使用 Spring Webflux 实现 trace-id

Question

我想为每个请求生成唯一的 traceId 并将其传递给所有服务。在 Spring MVC 中，通过使用 MDC 上下文并将 traceId 放在标头中相当容易，但在反应式堆栈中，由于 ThreadLocal，它根本不起作用。

一般来说，我想用单个 traceId 记录我拥有的每项服务的每个请求和响应，它可以识别整个系统中的特定操作。

我尝试根据文章创建自定义过滤器：https://azizulhaq-ananto.medium.com/how-to-handle-logs-and-tracing-in-spring-webflux-and-microservices-a0b45adc4610 但它是似乎不起作用。 我目前的解决方案只有日志响应和 traceId 在发出请求后丢失，所以没有响应。 让我们试着想象一下有两个服务：

service1

和

service2

。下面我试着勾勒出它应该如何工作。

它应该如何运作

1. client

   ->

   service1

   - service1 应该生成 traceId 和日志请求

2. service1

   ->

   service2

   - service2 应该从请求中获取 traceId，然后记录请求

3. service1

   <-

   service2

   - 经过一些计算服务 2 应该记录响应并将响应返回给服务 1

4. client

   <-

   service1

   - 最后 service1 应该记录响应（仍然使用相同的 traceId）并将响应返回给客户端

它是如何运作的

1. client

   ->

   service1

   - 日志中没有任何内容

2. service1

   ->

   service2

   - 日志中没有内容

3. service1

   <-

   service2

   - service2 正确记录并返回对 service1 的响应

4. client

   <-

   service1

   - service1 正在记录响应（但没有 traceId）

这是我的方法

@Component
public class TraceIdFilter implements WebFilter {

    private static final Logger log = LoggerFactory.getLogger(TraceIdFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        Map<String, String> headers = exchange.getRequest().getHeaders().toSingleValueMap();
        return Mono.fromCallable(() ->  {
            final long startTime = System.currentTimeMillis();

            return new ServerWebExchangeDecorator(exchange) {
                @Override
                public ServerHttpRequest getRequest() {
                    return new RequestLoggingInterceptor(super.getRequest(), false);
                }

                @Override
                public ServerHttpResponse getResponse() {
                    return new ResponseLoggingInterceptor(super.getResponse(), startTime, false);
                }
            };
        }).contextWrite(context -> {
            var traceId = "";
            if (headers.containsKey("X-B3-TRACEID")) {
                traceId = headers.get("X-B3-TRACEID");
                MDC.put("X-B3-TraceId", traceId);
            } else if (!exchange.getRequest().getURI().getPath().contains("/actuator")) {
                traceId = UUID.randomUUID().toString();
                MDC.put("X-B3-TraceId", traceId);
            }

            Context contextTmp = context.put("X-B3-TraceId", traceId);
            exchange.getAttributes().put("X-B3-TraceId", traceId);


            return contextTmp;
        }).flatMap(chain::filter);

    }


}

Github：https://github.com/Faelivrinx/kotlin-spring-boot

有什么现成的解决方案吗？

Answer 1

在 Spring Webflux 中，您不再有 ThreadLocal，但每个链请求都有一个唯一的上下文。您可以将 traceId 附加到此上下文，如下所示：

@Component
public class TraceIdFilter implements WebFilter {

    private static final Logger log = LoggerFactory.getLogger(TraceIdFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {

        return chain.filter(exchange)
                .subscriberContext(
                        ctx -> {
                            .....
                            var traceId = UUID.randomUUID().toString();
                            return   ctx.put("X-B3-TraceId", traceId);
                            .....
                        }
                );

    }


}

现在您服务中的链将在上下文中具有此属性。您可以使用静态方法 Mono.subscriberContext() 从您的服务中检索它。比如你可以这样获取traceId

Mono.subscriberContext()
    .flaMap(ctx -> {
       .....
       var traceId = ctx.getOrDefault("traceId", null);
       .....
    )

Answer 2

Sleuth 3.0 为 WebFlux 提供自动检测，这意味着，如果您什么都不做，您将始终获得订阅您的 Mono 或 Flux 的线程的当前跨度。如果你想覆盖它，例如因为你正在批处理大量请求并希望每个事务都有一个唯一的跟踪，你所要做的就是操纵你的运营商链的上下文。

private Mono<Data> performRequest() {
    var span = tracer.spanBuilder().setNoParent().start(); // generate a completely new trace
                                                           // note: you can also just generate a new span if you want
    return Mono.defer(() -> callRealService())
        .contextWrite(Context.of(TraceContext.class, span.context());
}

当采用这种方法时，确保导入

org.springframework.cloud.sleuth.Tracer

而不是 brave 的那个，因为它们使用不同的类型，如果它们没有正确对齐，Reactor 会丢弃你的 Mono 并显示一条丑陋的错误消息（不幸的是，因为 Context ist只是一个普通的老

Map<Object, Object>

，你不会得到编译器错误）。