Spring WebFlux网关FormData丢失及block()调用报错问题求助

1. 禁用阻塞操作block() #

WebFlux基于异步非阻塞模型，Reactor的NIO线程绝对禁止使用block()/blockFirst()/blockLast()这类阻塞方法，必须通过链式异步操作处理请求体：

// 错误写法（会触发阻塞报错）
// MultiValueMap<String, String> formData = exchange.getFormData().block();

// 正确异步处理方式
exchange.getFormData()
    .flatMap(formData -> {
        // 在此处处理FormData逻辑
        return Mono.just(formData);
    })
    // 需整合到过滤器的Mono<Void>返回值中，而非单独subscribe
    .then(chain.filter(exchange));

2. 缓存请求体解决FormData丢失问题 #

WebFlux的请求体是一次性消费的Flux<DataBuffer>流，若在过滤器中读取了请求体，必须重新包装成新请求放回ServerWebExchange，否则下游服务无法读取。针对包含二进制文件的multipart/form-data请求，需特殊处理：

步骤1：缓存原始请求体 #

先将请求体缓存到交换属性中，再重新生成可重复读取的流：

DataBufferUtils.join(exchange.getRequest().getBody())
    .flatMap(dataBuffer -> {
        // 缓存原始请求体
        exchange.getAttributes().put("cachedRequestBody", dataBuffer);
        // 生成可重复读取的请求体流
        Flux<DataBuffer> cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));
        // 包装新请求
        ServerHttpRequest newRequest = exchange.getRequest().mutate()
            .body(cachedFlux)
            .build();
        ServerWebExchange decoratedExchange = exchange.mutate().request(newRequest).build();
        
        return Mono.just(decoratedExchange);
    })

步骤2：处理Multipart类型请求 #

对于带二进制文件的multipart/form-data，需用getMultipartData()解析，而非getFormData()（后者仅处理普通表单字段），修改后需重新构造请求体：

.flatMap(decoratedExchange -> {
    if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(decoratedExchange.getRequest().getHeaders().getContentType())) {
        return decoratedExchange.getMultipartData()
            .flatMap(multipartData -> {
                // 修改Multipart数据，比如添加字段
                multipartData.add(routeData.getFieldName(), username);
                
                // 重新构建Multipart请求体
                MultipartBodyBuilder builder = new MultipartBodyBuilder();
                multipartData.forEach((key, parts) -> {
                    parts.forEach(part -> {
                        if (part instanceof FilePart) {
                            builder.part(key, ((FilePart) part).content());
                        } else {
                            builder.part(key, part);
                        }
                    });
                });
                
                // 生成新请求体并包装请求
                ServerHttpRequest originalRequest = decoratedExchange.getRequest();
                Flux<DataBuffer> newBody = BodyInserters.fromMultipartData(builder.build())
                    .insert(originalRequest, new BodyInserterContext());
                ServerHttpRequest updatedRequest = originalRequest.mutate().body(newBody).build();
                ServerWebExchange updatedExchange = decoratedExchange.mutate().request(updatedRequest).build();
                
                return chain.filter(updatedExchange);
            });
    } else {
        // 处理普通表单请求
        return decoratedExchange.getFormData()
            .flatMap(formData -> {
                formData.add(routeData.getFieldName(), username);
                return chain.filter(decoratedExchange);
            });
    }
});

3. 规范修改查询参数的方式 #

原addQueryParam方法可优化为通过ServerHttpRequestDecorator包装请求，避免直接修改交换属性：

private ServerHttpRequest addQueryParam(ServerHttpRequest request, String username) {
    URI updatedUri = UriComponentsBuilder.fromUri(request.getURI())
        .queryParam(routeData.getFieldName(), username)
        .build(true)
        .toUri();
    return request.mutate().uri(updatedUri).build();
}

在过滤器中调用时，直接包装请求即可：

ServerHttpRequest updatedRequest = addQueryParam(exchange.getRequest(), username);
ServerWebExchange updatedExchange = exchange.mutate().request(updatedRequest).build();
return chain.filter(updatedExchange);

4. 全局过滤器的正确返回值 #

GlobalFilter的filter方法必须返回Mono<Void>，所有异步操作需整合到该返回值中，确保流程同步：

@Override
public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
    String username = // 从上下文获取用户名的逻辑
    return DataBufferUtils.join(exchange.getRequest().getBody())
        .flatMap(dataBuffer -> {
            // 缓存并包装请求体
            exchange.getAttributes().put("cachedRequestBody", dataBuffer);
            Flux<DataBuffer> cachedFlux = Flux.defer(() -> Flux.just(dataBuffer.slice(0, dataBuffer.readableByteCount())));
            ServerHttpRequest newRequest = exchange.getRequest().mutate().body(cachedFlux).build();
            ServerWebExchange decoratedExchange = exchange.mutate().request(newRequest).build();
            
            // 根据请求类型处理
            if (MediaType.MULTIPART_FORM_DATA.isCompatibleWith(decoratedExchange.getRequest().getHeaders().getContentType())) {
                return decoratedExchange.getMultipartData()
                    .flatMap(multipartData -> {
                        ServerHttpRequest updatedRequest = addQueryParam(decoratedExchange.getRequest(), username);
                        return chain.filter(decoratedExchange.mutate().request(updatedRequest).build());
                    });
            } else {
                return decoratedExchange.getFormData()
                    .flatMap(formData -> {
                        ServerHttpRequest updatedRequest = addQueryParam(decoratedExchange.getRequest(), username);
                        return chain.filter(decoratedExchange.mutate().request(updatedRequest).build());
                    });
            }
        })
        .onErrorResume(e -> chain.filter(exchange));
}

内容的提问来源于stack exchange，提问作者Ali Canlı