如何在WebFlux中使用MDC+Logbook记录请求与响应体

针对你遇到的WebClient请求体显示FluxMap、响应体为空，且无法复用现有Feign日志逻辑的问题，以下是可行的实现方案：

核心思路 #

1. 正确读取响应式请求/响应体：响应式数据流是一次性消费的，必须读取后复制并重新生成数据流，避免后续处理失败
2. 复用现有日志逻辑：将WebClient的请求/响应转换成Feign日志服务能识别的参数格式
3. 维护MDC上下文：响应式线程切换会丢失MDC，需要手动传递并恢复上下文

1. 实现WebClient拦截器（ExchangeFilterFunction） #

这个拦截器负责读取请求/响应体，转换日志参数，并调用现有日志服务：

import org.slf4j.MDC;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import org.springframework.http.HttpHeaders;
import org.springframework.http.client.reactive.ClientHttpResponse;
import org.springframework.http.client.reactive.ClientHttpResponseDecorator;
import org.springframework.web.reactive.function.client.ClientRequest;
import org.springframework.web.reactive.function.client.ExchangeFilterFunction;
import org.springframework.web.reactive.function.client.ExchangeFunction;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.function.Consumer;

public class LogbookWebClientFilter implements ExchangeFilterFunction {

    private final Consumer<Map<String, String>> logConsumer;

    public LogbookWebClientFilter(Consumer<Map<String, String>> logConsumer) {
        this.logConsumer = logConsumer;
    }

    @Override
    public Mono<ClientHttpResponse> filter(ClientRequest request, ExchangeFunction next) {
        // 保存当前MDC上下文，避免线程切换丢失
        Map<String, String> mdcSnapshot = MDC.getCopyOfContextMap();

        // 读取并复制请求体
        return DataBufferUtils.join(request.body())
                .map(this::bufferToString)
                .defaultIfEmpty("")
                .flatMap(requestBody -> {
                    // 构建和Feign一致的请求日志参数
                    Map<String, String> logParams = buildRequestLogParams(request, requestBody);

                    // 重新构建请求体（因为已被消费），并转发请求
                    return next.exchange(ClientRequest.from(request)
                                    .body(Flux.just(request.body()))
                                    .build())
                            .flatMap(response -> handleResponse(response, logParams, mdcSnapshot));
                });
    }

    private Mono<ClientHttpResponse> handleResponse(ClientHttpResponse response, Map<String, String> logParams, Map<String, String> mdcSnapshot) {
        // 读取并复制响应体
        return DataBufferUtils.join(response.body())
                .map(this::bufferToString)
                .defaultIfEmpty("")
                .map(responseBody -> {
                    // 补充响应日志参数
                    logParams.put("RESPONSE_CODE", String.valueOf(response.statusCode().value()));
                    logParams.put("RESPONSE_HEADERS", formatHeaders(response.headers().asHttpHeaders()));
                    logParams.put("RESPONSE_BODY", responseBody);
                    logParams.put("TYPE", "OUTREQ"); // 与Feign的TYPE保持一致，区分外部请求

                    // 恢复MDC上下文并记录日志
                    if (mdcSnapshot != null) {
                        MDC.setContextMap(mdcSnapshot);
                    }
                    logConsumer.accept(logParams);
                    MDC.clear();

                    // 装饰响应，重新生成响应体数据流
                    return new ClientHttpResponseDecorator(response) {
                        @Override
                        public Flux<DataBuffer> body() {
                            return Flux.just(responseBody.getBytes(StandardCharsets.UTF_8))
                                    .map(bufferFactory::wrap);
                        }
                    };
                });
    }

    private Map<String, String> buildRequestLogParams(ClientRequest request, String requestBody) {
        String query = request.url().getQuery() != null ? request.url().getQuery() : "{}";
        return Map.of(
                "METHOD", request.method().name(),
                "REQUEST_PATH", request.url().getPath(),
                "REQUEST_QUERY", query,
                "REQUEST_HEADERS", formatHeaders(request.headers()),
                "REQUEST_BODY", requestBody
        );
    }

    private String bufferToString(DataBuffer buffer) {
        byte[] bytes = new byte[buffer.readableByteCount()];
        buffer.read(bytes);
        DataBufferUtils.release(buffer);
        return new String(bytes, StandardCharsets.UTF_8);
    }

    private String formatHeaders(HttpHeaders headers) {
        return headers.entrySet().stream()
                .map(entry -> entry.getKey() + ":\"" + String.join(",", entry.getValue()) + "\"")
                .reduce((a, b) -> a + ", " + b)
                .map(s -> "[" + s + "]")
                .orElse("[]");
    }
}

2. 注册拦截器到WebClient #

将拦截器绑定到WebClient，并复用你现有的Feign日志服务：

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.reactive.function.client.WebClient;

@Configuration
public class WebClientConfig {

    private final YourExistingLogService existingLogService; // 替换成你实际的Feign日志服务类

    public WebClientConfig(YourExistingLogService existingLogService) {
        this.existingLogService = existingLogService;
    }

    @Bean
    public WebClient webClient() {
        return WebClient.builder()
                .filter(new LogbookWebClientFilter(logParams -> {
                    // 调用现有日志服务，生成和Feign一致格式的日志
                    existingLogService.logExternalRequest(logParams);
                }))
                .build();
    }
}

3. 维护MDC上下文传递 #

在调用WebClient时，将MDC上下文放入Reactor Context，避免线程切换丢失：

// 示例WebClient调用代码
webClient.post()
        .uri("/ping")
        .bodyValue(yourRequestBody)
        .header("Session", MDC.get("Session"))
        .exchangeToMono(response -> response.bodyToMono(YourResponse.class))
        // 将MDC上下文存入Reactor Context
        .contextWrite(ctx -> ctx.put("MDC_CONTEXT", MDC.getCopyOfContextMap()))
        .doOnNext(response -> {
            // 处理响应时可恢复MDC（如果需要）
            Map<String, String> mdc = ctx.getOrDefault("MDC_CONTEXT", null);
            if (mdc != null) {
                MDC.setContextMap(mdc);
            }
            // 业务处理逻辑
            MDC.clear();
        });

关键注意点 #

• 不要使用WebFilter：WebFilter是拦截服务器端请求的，WebClient作为客户端需要用ExchangeFilterFunction
• 必须复制数据流：响应式DataBuffer是一次性消费的，读取后必须重新生成数据流，否则后续请求/响应处理会失败
• 对齐日志参数：确保logParams中的key（如REQUEST_BODY、RESPONSE_CODE）和Feign日志的key完全一致，这样就能复用现有格式

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