java

5.1 请求体读取与 JSON 解析

WebFlux 请求体深度处理：Mono/Flux 绑定方式与背压传递语义、Jackson 反序列化配置、表单与 multipart 文件上传、流式大文件 DataBuffer 读取与释放安全，以及请求体大小限制与自定义 HttpMessageReader。

Leeting Yan 5 分钟阅读 2351 字

@RequestBody 的响应式语义

WebFlux 的 @RequestBody 支持三种绑定方式：

@PostMapping("/users")
public Mono<User> createMono(@RequestBody Mono<User> userMono) { ... }   // A: Mono

@PostMapping("/users/batch")
public Flux<User> createFlux(@RequestBody Flux<User> userFlux) { ... }   // B: Flux

@PostMapping("/users/sync")
public Mono<User> createSync(@RequestBody User user) { ... }             // C: 直接对象
方式内部实现适用场景
A Mono<T>解码为一个对象后以 Mono 发出单对象请求
B Flux<T>流式反序列化，每个 JSON 元素独立NDJSON 批量上传
C TSpring 异步聚合完成后再调用方法简单场景，写起来更直接

方式 A 和 B 是响应式的核心：请求体字节流 → DataBuffer → JSON parser → User 对象，全程通过 request(n) 背压控制读取速率。

Mono 请求体的完整处理链

@PostMapping("/orders")
public Mono<Order> createOrder(@RequestBody Mono<Order> orderMono) {
    return orderMono
        .flatMap(this::validate)         // 校验
        .flatMap(this::enrich)           // 补充字段（如默认状态）
        .flatMap(this::saveToDb);        // 持久化（第 7 章 R2DBC）
}

private Mono<Order> validate(Order order) {
    if (order.amount() <= 0) {
        return Mono.error(new IllegalArgumentException("amount must > 0"));
    }
    return Mono.just(order);
}

private Mono<Order> enrich(Order order) {
    return Mono.just(new Order(
        order.id(),
        order.amount(),
        order.status() != null ? order.status() : "PENDING",
        Instant.now()
    ));
}

flatMap 把请求体反序列化、校验、补充、持久化串成一条响应式链。如果客户端发送极慢的请求体，每个 flatMap 都不会提前执行，背压一直传递到 TCP 层。

Flux 请求体：批量/流式上传

当客户端以 NDJSON（换行分隔 JSON）格式发送批量数据时，用 Flux<T> 逐条处理：

@PostMapping(value = "/orders/batch", consumes = MediaType.APPLICATION_NDJSON_VALUE)
public Flux<OrderResult> batchCreate(@RequestBody Flux<Order> orders) {
    return orders
        .flatMap(order ->
            validate(order)
                .flatMap(this::saveToDb)
                .map(saved -> new OrderResult(saved.id(), "CREATED"))
                .onErrorResume(err ->
                    Mono.just(new OrderResult(null, "ERROR: " + err.getMessage()))),
            4  // 并发度：同时处理 4 条
        );
}

flatMap(concurrency) 控制并行处理数量。如果后端数据库连接池只有 10 个连接，concurrency 不应超过 10，否则请求会排队而不是并行执行。

JSON 解析细节

Spring WebFlux 默认使用 Jackson 反序列化。两个需要注意的配置：

1. 字段命名策略

// application.yml
spring:
  jackson:
    property-naming-strategy: SNAKE_CASE  # 自动转换 snake_case

开启后，Java 的 orderId 字段自动匹配 JSON 的 order_id 键。

2. 忽略未知字段

// application.yml
spring:
  jackson:
    deserialization:
      fail-on-unknown-properties: false

Spring Boot 的 Jackson 自动配置默认已经关闭 FAIL_ON_UNKNOWN_PROPERTIES，所以多数应用不写这段也不会因为多余字段报错。显式写出来是为了让这个契约在配置里可见；如果你反而希望严格拒绝未知字段，把它设为 true，并准备好在客户端新增字段时接收 UnrecognizedPropertyException。

3. 自定义 Module

@Bean
public com.fasterxml.jackson.databind.Module customModule() {
    return new com.fasterxml.jackson.databind.module.SimpleModule()
        .addSerializer(Instant.class, new InstantSerializer())
        .addDeserializer(Instant.class, new InstantDeserializer());
}

自定义 Instant 的序列化格式（如转为 Unix 时间戳而非 ISO-8601）。

请求体过大保护

WebFlux 的 codec 内存上限默认是 256KB（spring.codec.max-in-memory-size），不是无限制。超过上限时解码器会抛 DataBufferLimitException（继承 IllegalStateException）；默认错误处理通常把它归为 500，需要在全局异常处理里显式映射成 413（见 5.3 的错误契约）：

@Configuration
public class WebConfig implements WebFluxConfigurer {
    @Override
    public void configureHttpMessageCodecs(ServerCodecConfigurer configurer) {
        configurer.defaultCodecs().maxInMemorySize(512 * 1024);  // 512KB 上限
    }
}

文件上传也受同一个配置约束：FilePart.transferTo() 不会把整个文件聚合进一个 DataBuffer，但默认的 multipart 解析器会把超过 maxInMemorySize 的 part 先写到临时文件，再复制到目标路径，额外占用磁盘和一次 I/O。大文件场景要同时确认 maxInMemorySize、临时目录和磁盘配额，不能只看 transferTo()。

大文件上传：流式处理

@PostMapping(value = "/upload", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public Mono<String> upload(@RequestPart("file") FilePart filePart) {
    Path dest = Path.of("/tmp/uploads", filePart.filename());

    return filePart.transferTo(dest)
        .then(Mono.just("Saved to " + dest));
}

FilePart.transferTo() 避免把整个文件聚合进一个 DataBuffer，内存占用有上限，但默认的 multipart 解析器并不是网络流直通：小于 maxInMemorySize 的 part 放在内存，超过的 part 先写到临时文件，transferTo() 再把临时文件复制到目标路径。大文件会多一次磁盘写入和复制，临时目录和磁盘配额都要纳入容量评估。需要真正边收边写时，应启用 multipart 流式解析或自定义 reader。

注意 filePart.filename() 完全来自客户端。直接拼进目标路径会遇到 ../ 路径穿越（例如文件名 ../../etc/cron.d/x）；生产代码应只取 Paths.get(filename).getFileName()，更稳妥的是由服务端生成文件名，把原始文件名只存进数据库。

如果需要边上传边处理（如计算 MD5）：

@PostMapping(value = "/upload-hash", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
public Mono<String> uploadWithHash(@RequestPart("file") FilePart filePart) {
    java.security.MessageDigest md;
    try { md = java.security.MessageDigest.getInstance("MD5"); }
    catch (Exception e) { return Mono.error(e); }

    return filePart.content()
        .doOnNext(buffer -> {
            try {
                byte[] bytes = new byte[buffer.readableByteCount()];
                buffer.read(bytes);
                md.update(bytes);
            } finally {
                // 必须释放 DataBuffer，否则内存泄漏；放在 finally 里防止读异常时漏掉
                DataBufferUtils.release(buffer);
            }
        })
        .then(Mono.fromCallable(() -> {
            byte[] digest = md.digest();
            return bytesToHex(digest);
        }));
}

static String bytesToHex(byte[] bytes) {
    StringBuilder sb = new StringBuilder();
    for (byte b : bytes) sb.append(String.format("%02x", b));
    return sb.toString();
}

关键：DataBufferUtils.release(buffer) 在读取后必须调用。DataBuffer 这个 Java 对象本身仍由 JVM GC 管理，但它底层包装的池化 ByteBuf 采用引用计数；GC 不会替你把引用计数减到 0。不释放，池化内存就回不到池子，最终表现为直接内存泄漏。

自定义 HttpMessageReader

如果默认 Jackson 反序列化不满足需求（如需要流式 CSV 解析），可以自定义 reader：

public class CsvMessageReader implements HttpMessageReader<String[]> {

    private static final MediaType TEXT_CSV = MediaType.valueOf("text/csv");

    @Override
    public List<MediaType> getReadableMediaTypes() {
        return List.of(TEXT_CSV);
    }

    @Override
    public boolean canRead(ResolvableType elementType, MediaType mediaType) {
        return mediaType != null && mediaType.isCompatibleWith(TEXT_CSV);
    }

    @Override
    public Flux<String[]> read(ResolvableType elementType,
                               ReactiveHttpInputMessage message,
                               Map<String, Object> hints) {
        // message.getBody() 是 Flux<DataBuffer>，没有 asInputStream()
        InputStream in = DataBufferUtils.subscriberInputStream(message.getBody(), 8192);
        return Flux.using(
            () -> new BufferedReader(new InputStreamReader(in, StandardCharsets.UTF_8)),
            reader -> Flux.fromStream(reader.lines())
                .map(line -> line.split(",")),
            reader -> {
                try { reader.close(); } catch (IOException ignored) { }
            })
            // reader.lines() 是阻塞迭代，不能跑在 Netty EventLoop 上
            .subscribeOn(Schedulers.boundedElastic());
    }

    @Override
    public Mono<String[]> readMono(ResolvableType elementType,
                                   ReactiveHttpInputMessage message,
                                   Map<String, Object> hints) {
        return read(elementType, message, hints).next();
    }
}

自定义 reader 注册为 Spring Bean 后，WebFlux 遇到 Content-Type: text/csv 请求会自动使用它。subscriberInputStream 的底层仍是 DataBuffer 流，关闭 BufferedReader 会取消订阅并释放缓冲区。但 reader.lines() 是阻塞迭代，必须像上面那样 subscribeOn(Schedulers.boundedElastic())，否则会占住 Netty EventLoop；这也说明这个 reader 只适合中小 CSV，超大文件应改用真正的非阻塞解析器。

eventd v2 的请求体设计

eventd v2 的 API 层遵循以下规则：

场景请求体类型处理模式
单条创建（用户、订单）Mono<T>校验 → 补充 → R2DBC insert
批量导入Flux<T> + NDJSON流式处理，并发度 4
文件上传FileParttransferTo() 直接到磁盘
流式日志上报Flux<String> + SSE逐行解析，实时入库

小结

能力方式注意点
单对象请求体@RequestBody Mono<T>惰性反序列化，推荐
批量/流式@RequestBody Flux<T>NDJSON 格式，flatMap(concurrency)
直接对象@RequestBody TSpring 异步聚合后调用，不阻塞线程
文件上传@RequestPart FileParttransferTo() 流式写盘，不整体驻留内存
内存安全DataBufferUtils.release()手动消费 content()/receive() 给出的 DataBuffer 后要释放；框架已转成 String/byte[] 的 body 不用手动释放
大小限制maxInMemorySize防恶意大 JSON
自定义格式HttpMessageReaderBean 自动注册

下一节 5.2 响应式响应流 ：SSE、chunkedTransfer、Flux 响应、MediaType 协商与背压。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南