如何在spring-webflux WebFilter中正确使用slf4j MDC

Question

我引用了博客文章Contextual Logging with Reactor Context和MDC，但我不知道如何在WebFilter中访问Reactor上下文。

@Component
public class RequestIdFilter implements WebFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        List<String> myHeader =  exchange.getRequest().getHeaders().get("X-My-Header");

        if (myHeader != null && !myHeader.isEmpty()) {
            MDC.put("myHeader", myHeader.get(0));
        }

        return chain.filter(exchange);
    }
}

Answer 1

从 Spring Boot 2.2 开始，Schedulers.onScheduleHook 使您能够处理 MDC：

Schedulers.onScheduleHook("mdc", runnable -> {
    Map<String, String> map = MDC.getCopyOfContextMap();
    return () -> {
        if (map != null) {
            MDC.setContextMap(map);
        }
        try {
            runnable.run();
        } finally {
            MDC.clear();
        }
    };
});

或者，可以使用 Hooks.onEachOperator 通过订阅者上下文传递 MDC 值。

http://ttddyy.github.io/mdc-with-webclient-in-webmvc/

这不是完整的 MDC 解决方案，例如，在我的情况下，我无法清理 R2DBC 线程中的 MDC 值。

更新：这篇文章确实解决了我的 MDC 问题：https://www.novatec-gmbh.de/en/blog/how-can-the-mdc-context-be-used-in-the-reactive-spring-applications/

它提供了基于订阅者上下文更新 MDC 的正确方法。

将其与SecurityContext::class.java填充的密钥结合起来AuthenticationWebFilter，您将能够将用户登录记录到您的日志中。

Answer 2

您可以执行类似于下面的操作，可以context使用任何您喜欢的类设置，对于本例，我仅使用标头-但是自定义类就可以了。如果您在此处设置，则使用处理程序等进行的任何日志记录也都可以访问context。
在logWithContext下面，设置MDC并清除之后。显然，可以用任何您喜欢的东西代替它。

public class RequestIdFilter  implements WebFilter {

    private Logger LOG = LoggerFactory.getLogger(RequestIdFilter.class);

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        HttpHeaders headers = exchange.getRequest().getHeaders();
        return chain.filter(exchange)
                .doAfterSuccessOrError((r, t) -> logWithContext(headers, httpHeaders -> LOG.info("Some message with MDC set")))
                .subscriberContext(Context.of(HttpHeaders.class, headers));
    }

    static void logWithContext(HttpHeaders headers, Consumer<HttpHeaders> logAction) {
        try {
            headers.forEach((name, values) -> MDC.put(name, values.get(0)));
            logAction.accept(headers);
        } finally {
            headers.keySet().forEach(MDC::remove);
        }

    }

}

Answer 3

这是一种基于最新方法的解决方案，截至2021 年 5 月，取自官方文档：

import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.function.Consumer;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.HttpHeaders;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.web.server.ServerWebExchange;
import org.springframework.web.server.WebFilter;
import org.springframework.web.server.WebFilterChain;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Signal;
import reactor.util.context.Context;

@Slf4j
@Configuration
public class RequestIdFilter implements WebFilter {

  @Override
  public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
    ServerHttpRequest request = exchange.getRequest();
    String requestId = getRequestId(request.getHeaders());
    return chain
        .filter(exchange)
        .doOnEach(logOnEach(r -> log.info("{} {}", request.getMethod(), request.getURI())))
        .contextWrite(Context.of("CONTEXT_KEY", requestId));
  }

  private String getRequestId(HttpHeaders headers) {
    List<String> requestIdHeaders = headers.get("X-Request-ID");
    return requestIdHeaders == null || requestIdHeaders.isEmpty()
        ? UUID.randomUUID().toString()
        : requestIdHeaders.get(0);
  }

  public static <T> Consumer<Signal<T>> logOnEach(Consumer<T> logStatement) {
    return signal -> {
      String contextValue = signal.getContextView().get("CONTEXT_KEY");
      try (MDC.MDCCloseable cMdc = MDC.putCloseable("MDC_KEY", contextValue)) {
        logStatement.accept(signal.get());
      }
    };
  }

  public static <T> Consumer<Signal<T>> logOnNext(Consumer<T> logStatement) {
    return signal -> {
      if (!signal.isOnNext()) return;
      String contextValue = signal.getContextView().get("CONTEXT_KEY");
      try (MDC.MDCCloseable cMdc = MDC.putCloseable("MDC_KEY", contextValue)) {
        logStatement.accept(signal.get());
      }
    };
  }
}

鉴于您的 中有以下行application.properties：

logging.pattern.level=[%X{MDC_KEY}] %5p

那么每次调用端点时，您的服务器日志都会包含这样的日志：

2021-05-06 17:07:41.852 [60b38305-7005-4a05-bac7-ab2636e74d94]  INFO 20158 --- [or-http-epoll-6] my.package.RequestIdFilter    : GET http://localhost:12345/my-endpoint/444444/

每次您想在反应式上下文中手动记录某些内容时，您都必须将以下内容添加到您的反应式链中：

.doOnEach(logOnNext(r -> log.info("Something")))

如果您希望X-Request-ID将 传播到其他服务以进行分布式跟踪，则需要从反应式上下文（而不是从 MDC）中读取它并WebClient使用以下内容包装您的代码：

Mono.deferContextual(
    ctx -> {
      RequestHeadersSpec<?> request = webClient.get().uri(uri);
      request = request.header("X-Request-ID", ctx.get("CONTEXT_KEY"));
      // The rest of your request logic...
    });

Answer 4

2023：忘记logOnNext包装纸。使用上下文传播

\n

我早在 2018 年就开始使用 Reactor，到目前为止，还没有一个真正好的替代方法来替代内部的包装方法，doOnNext您可以手动将跟踪字段从 Reactor 的上下文复制到 MDC，从而制作自己的临时方法反应式世界和命令式世界之间的桥梁，瞧\xc3\xa0，你的日志现在可以有意义了。但事情发生了变化，最后，一个新的解决方案是 \xe2\x80\x93上下文传播。我们来看看吧。

\n

想象一下，您有 Spring 服务，并且有一组定义诊断上下文的字段，这些字段用于跟踪服务的活动。假设您将该集存储为以下属性：management.tracing.baggage.correlation.fields: trace, session。

\n

现在，要将这些字段自动填充到执行反应式链调用的线程的 MDC 中，从反应式上下文中获取键值，您只需将上下文传播库添加到您的项目中，然后执行以下操作服务范围配置：

\n

/**\n * 1. Will register ThreadLocalAccessors into ContextRegistry for fields listed in application.yml as property value\n * <b>management.tracing.baggage.correlation.fields</b>\n * 2. Enables Automatic Context Propagation for all reactive methods\n *\n * @see <a href="https://github.com/micrometer-metrics/context-propagation">context-propagation</a>\n */\n@Configuration\n@ConditionalOnClass({ContextRegistry.class, ContextSnapshotFactory.class})\n@ConditionalOnProperty(value = "management.tracing.baggage.correlation.fields", matchIfMissing = true)\npublic class MdcContextPropagationConfiguration {\n\n    public MdcContextPropagationConfiguration(@Value("${management.tracing.baggage.correlation.fields}")\n                                              List<String> fields) {\n        if (!isEmpty(fields)) {\n            fields.forEach(claim -> ContextRegistry.getInstance()\n                                                   .registerThreadLocalAccessor(claim,\n                                                                                () -> MDC.get(claim),\n                                                                                value -> MDC.put(claim, value),\n                                                                                () -> MDC.remove(claim)));\n            return;\n        }\n\n        Hooks.enableAutomaticContextPropagation();\n    }\n}\n

\n

这里的技巧是使用Hooks.enableAutomaticContextPropagation(). 一旦我们注册了一组用于传播的 ThreadLocalsAccessors 来映射跟踪字段，该钩子将确保在链的每次调用中将注册字段的键下的值从反应式上下文传递到 MDC 。MDC 中的这些字段随后可以在任何首选项中在 slf4j 附加程序中引用。

\n

就是这样。

\n