我参考了博客文章《使用 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);
}
}
以下是一个基于最新方法的解决方案，截至 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代码：
如果你希望X-Request-ID被用于分布式跟踪并在其他服务中传递，你需要从reactive context(响应式上下文)中读取它（而不是从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...
});
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 的正确方法。
结合由 AuthenticationWebFilter 填充的 SecurityContext::class.java 键，你将能够将用户登录信息放入日志中。
Schedulers.onScheduleHook 放在哪里？ - Marco Lackovic@PostConstruct。这篇文章中的方法对我很有效。虽然不确定性能影响如何，但我认为应该没问题。 - Dmytrocontext，在这个例子中，我只是使用了标题 - 但自定义类也可以正常工作。
如果您在此处设置它，则具有处理程序等功能的任何记录器也将访问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);
}
}
}
.subscriberContext和.doAfterSuccessOrError，因为上下文写入只对在其上方的操作符可见。 - Simon BaslélogOnNext包装器。使用上下文传播doOnNext内部没有一个真正好的替代方法，你需要手动将Reactor的上下文中的跟踪字段复制到MDC中，创建自己的特定桥梁，将响应式和命令式世界连接起来，然后你的日志就能有意义了。但是事情变了，终于有了一个新的解决方案-上下文传播。让我们来看看它。management.tracing.baggage.correlation.fields: trace, session。/**
* 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中。主要思路是以以下方式使用Context进行MDC传播：
.contextWrite(context -> Context.of(MDC.getCopyOfContextMap()))来完成）.doFinally(signalType -> MDC.clear())来完成）主要问题是在派生线程中访问下游Context。您可以使用最方便的方法实现步骤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);
}
});
}
我用以下方式实现了这个：
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值可以被记录。
logOnEach()和logOnNext()之间的实际区别是什么 - 动态更改最新内容？ - Zon