如何在Spring WebFlux的WebFilter中修改请求体并传递至下游？

核心思路 #

WebFlux中ServerHttpRequest的请求体是不可变且只能被消费一次的，因此需要通过以下步骤实现修改：

1. 读取原始请求体并反序列化为MyRequest实例
2. 修改updateUser字段为请求头user的值
3. 将修改后的对象重新序列化为JSON，包装成新的DataBuffer
4. 构建包含新请求体的ServerHttpRequest，并传递给下游过滤器/处理器

完整实现代码 #

@Component
@Slf4j
public class IdentityFilter implements WebFilter {
    private static final String USER = "user";
    private final ObjectMapper objectMapper = new ObjectMapper();

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        ServerHttpRequest request = exchange.getRequest();

        // 仅处理POST请求且Content-Type为JSON的情况
        if (HttpMethod.POST.equals(request.getMethod()) && 
            request.getHeaders().getContentType() != null &&
            MediaType.APPLICATION_JSON.isCompatibleWith(request.getHeaders().getContentType())) {
            
            String userFullName = Optional.ofNullable(request.getHeaders().get(USER))
                    .orElse(Collections.singletonList(""))
                    .get(0);

            // 读取并修改请求体
            return DataBufferUtils.join(request.getBody())
                    .flatMap(dataBuffer -> {
                        // 保存原始数据缓冲区，避免重复消费
                        DataBufferUtils.retain(dataBuffer);
                        try {
                            // 反序列化为MyRequest实现类（替换为你项目中的实际实现类）
                            MyRequest myRequest = objectMapper.readValue(dataBuffer.asInputStream(), YourActualRequestClass.class);
                            myRequest.setUpdateUser(userFullName);

                            // 重新序列化为JSON
                            byte[] updatedBytes = objectMapper.writeValueAsBytes(myRequest);
                            DataBuffer updatedBuffer = exchange.getResponse().bufferFactory().wrap(updatedBytes);

                            // 构建新的请求
                            ServerHttpRequest newRequest = request.mutate()
                                    .header(HttpHeaders.CONTENT_LENGTH, String.valueOf(updatedBytes.length))
                                    .body(Flux.just(updatedBuffer))
                                    .build();

                            // 传递新的请求到下游
                            return chain.filter(exchange.mutate().request(newRequest).build());
                        } catch (IOException e) {
                            log.error("Failed to modify request body", e);
                            return Mono.error(new IllegalStateException("Request body modification failed", e));
                        } finally {
                            DataBufferUtils.release(dataBuffer);
                        }
                    });
        }

        // 非POST/非JSON请求直接放行
        return chain.filter(exchange);
    }
}

关键注意点 #

• 请求体资源管理：使用DataBufferUtils.join读取完整请求体，通过retain和release管理缓冲区资源，避免内存泄漏。
• 请求类型校验：仅处理JSON格式的POST请求，避免处理非预期的请求体格式导致异常。
• Content-Length更新：修改请求体后需同步更新Content-Length头，否则下游可能因长度不匹配解析失败；若不想手动计算长度，也可移除该头，让框架自动使用chunked编码。
• 异常处理：捕获序列化/反序列化异常，避免请求链中断后无友好错误反馈。

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