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

问题描述 投票：0回答：6

我在博客文章使用 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);
    }
}
java spring-webflux project-reactor
6个回答
16
投票

这是一个基于最新方法的解决方案，截至 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...
    });
10
投票

从 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-应用/

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

将其与由 

SecurityContext::class.java
填充的 
AuthenticationWebFilter
键组合，您将能够将用户登录放入您的日志中。

9
投票

您可以执行类似于下面的操作，您可以使用您喜欢的任何类设置 

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);
        }

    }

}
1
投票

我的解决方案基于 Reactor 3 参考指南方法，但使用 doOnSuccess 而不是 doOnEach。

主要思想是使用Context以下面的方式进行MDC传播：

  1. 使用上游流中的 MDC 状态填充下游上下文（将由派生线程使用）（可以通过 
    .contextWrite(context -> Context.of(MDC.getCopyOfContextMap()))
    完成）
  2. 在派生线程中访问下游 Context，并使用下游 Context 中的值填充派生线程中的 MDC（主要挑战）
  3. 清除下游Context中的MDC（可以通过
    .doFinally(signalType -> MDC.clear())
    完成）

主要问题是在派生线程中访问下游上下文。您可以使用最方便的方法来实施步骤 2）。但这是我的解决方案：

webclient.post()
   .bodyValue(someRequestData)
   .retrieve()
   .bodyToMono(String.class)
// By this action we wrap our response with a new Mono and also 
// in parallel fill MDC with values from a downstream Context because
// we have an access to it
   .flatMap(wrapWithFilledMDC())
   .doOnSuccess(response -> someActionWhichRequiresFilledMdc(response)))
// Fill a downstream context with the current MDC state
   .contextWrite(context -> Context.of(MDC.getCopyOfContextMap())) 
// Allows us to clear MDC from derived threads
   .doFinally(signalType -> MDC.clear())
   .block();

// Function which implements second step from the above main idea
public static <T> Function<T, Mono<T>> wrapWithFilledMDC() {
// Using deferContextual we have an access to downstream Context, so
// we can just fill MDC in derived threads with 
// values from the downstream Context
   return item -> Mono.deferContextual(contextView -> {
// Function for filling MDC with Context values 
// (you can apply your action)
      fillMdcWithContextView(contextView);
      return Mono.just(item);
   });
}

public static void fillMdcWithContextValues(ContextView contextView) {
   contextView.forEach(
      (key, value) -> {
          if (key instanceof String keyStr && value instanceof String valueStr) {
             MDC.put(keyStr, valueStr);
          }
   });
}

这种方法也可以应用于 doOnError 和 onErrorResume 方法，因为主要思想是相同的。

使用过的版本：

  • 弹簧启动：2.7.3
  • spring-webflux：5.3.22（来自 spring-boot）
  • reactor-core：3.4.22（来自 spring-webflux）
  • reactor-netty：1.0.22（来自 spring-webflux）
0
投票

2023：不再有
logOnNext
包装纸了。更好的解决方案在这里：

我早在 2018 年就开始使用 Reactor，到目前为止，还没有一个真正好的替代方案来替代 

doOnNext
内部的包装器方法，您可以手动将跟踪字段从 Reactor 的上下文复制到 MDC，在两者之间建立自己的临时桥梁反应式世界和命令式世界，瞧，您的日志现在可以有意义了。但事情发生了变化，最终，一个新的解决方案出现了——上下文传播。我们来看看吧。

假设您有 Spring 服务，并且有一组定义诊断上下文的字段，这些字段用于跟踪服务的活动。假设您将该集存储为以下属性：

management.tracing.baggage.correlation.fields: trace, session
。

现在，要将这些字段自动填充到执行反应式链调用的线程的 MDC 中，并将这些字段用于反应式上下文，您只需添加以下服务范围配置：

/**
 * 1. Will register ThreadLocalAccessors into ContextRegistry for fields listed in application.yml as property value
 * <b>management.tracing.baggage.correlation.fields</b>
 * 2. Enables Automatic Context Propagation for all reactive methods
 *
 * @see <a href="https://github.com/micrometer-metrics/context-propagation">context-propagation</a>
 */
@Configuration
@ConditionalOnClass({ContextRegistry.class, ContextSnapshotFactory.class})
@ConditionalOnProperty(value = "management.tracing.baggage.correlation.fields", matchIfMissing = true)
public class MdcContextPropagationConfiguration {

    public MdcContextPropagationConfiguration(@Value("${management.tracing.baggage.correlation.fields}")
                                              List<String> fields) {
        if (!isEmpty(fields)) {
            fields.forEach(claim -> ContextRegistry.getInstance()
                                                   .registerThreadLocalAccessor(claim,
                                                                                () -> MDC.get(claim),
                                                                                value -> MDC.put(claim, value),
                                                                                () -> MDC.remove(claim)));
            return;
        }

        Hooks.enableAutomaticContextPropagation();
    }
}

这里的技巧是使用

Hooks.enableAutomaticContextPropagation()
。一旦我们注册了一组用于传播的 ThreadLocalsAccessors 来映射跟踪字段，该挂钩将确保在每次调用链时将注册字段的键下的值从反应式上下文传递到 MDC 。

就是这样。

-3
投票
我通过以下方式实现了这一点：-

package com.nks.app.filter; import lombok.extern.slf4j.Slf4j; import org.slf4j.MDC; import org.springframework.stereotype.Component; import org.springframework.web.server.ServerWebExchange; import org.springframework.web.server.WebFilter; import org.springframework.web.server.WebFilterChain; import reactor.core.publisher.Mono; /** * @author nks */ @Component @Slf4j public class SessionIDFilter implements WebFilter { private static final String APP_SESSION_ID = "app-session-id"; /** * Process the Web request and (optionally) delegate to the next * {@code WebFilter} through the given {@link WebFilterChain}. * * @param serverWebExchange the current server exchange * @param webFilterChain provides a way to delegate to the next filter * @return {@code Mono<Void>} to indicate when request processing is complete */ @Override public Mono<Void> filter(ServerWebExchange serverWebExchange, WebFilterChain webFilterChain) { serverWebExchange.getResponse() .getHeaders().add(APP_SESSION_ID, serverWebExchange.getRequest().getHeaders().getFirst(APP_SESSION_ID)); MDC.put(APP_SESSION_ID, serverWebExchange.getRequest().getHeaders().getFirst(APP_SESSION_ID)); log.info("[{}] : Inside filter of SessionIDFilter, ADDED app-session-id in MDC Logs", MDC.get(APP_SESSION_ID)); return webFilterChain.filter(serverWebExchange); } } 
并且，可以记录与线程 
app-session-id
 关联的值。

    



      

      
    

  

  

    
最新问题


  





  

    © www.soinside.com 2019 - 2024. All rights reserved.