Reactive Spring使用Flux请求体报400请求体缺失且Delete重复执行如何解决

问题根因 #

你遇到的两个问题本质都是Flux冷数据流的特性导致的：

• 你对同一个请求体Flux<Body>做了两次订阅：第一次是执行flatMap处理删除逻辑的时候，第二次是thenMany(body)读取数据做保存的时候。Spring WebFlux的请求体流只能被消费一次，第二次订阅的时候自然就会抛出请求体缺失的400错误。
• 删除执行n次是因为你把delete操作放在了flatMap里，流里每有一个元素就会触发一次delete调用，不符合你只执行一次的要求。

补充：你之前只写delete逻辑时运行正常，是因为当时只对请求体Flux做了一次订阅，没有第二次消费流的操作，所以不会触发请求体缺失的错误。

改造实现方案 #

核心思路是先把请求体流一次性消费为内存集合缓存，后续所有操作都基于缓存的集合执行，避免多次订阅原始请求流。

1. Service层逻辑 #

public Mono<Boolean> update(Flux<RequestBody> body, Long id) {
    // 仅订阅一次原始请求流，收集为List缓存
    return body.collectList()
            // 第一步：统一校验所有请求参数
            .flatMap(requestList -> {
                // 可自定义校验逻辑，比如判空、字段合法性校验
                if (requestList.isEmpty()) {
                    return Mono.error(new IllegalArgumentException("请求体不能为空"));
                }
                for (RequestBody req : requestList) {
                    if (req.getValue() == null || req.getName() == null) {
                        return Mono.error(new IllegalArgumentException("请求字段不合法"));
                    }
                }
                // 第二步：执行1次删除操作，取匹配字段值（可根据业务调整取值逻辑）
                String deleteMatchValue = requestList.get(0).getValue();
                return repo.deleteCriteria(id, deleteMatchValue)
                        // 第三步：转换DTO为Entity并批量保存
                        .thenMany(Flux.fromIterable(requestList))
                        .map(dto -> mapper.convertToEntity(dto))
                        .flatMap(entity -> repo.save(entity.getName(), entity.getData()))
                        // 第四步：统计保存后的匹配行数返回结果
                        .then(repo.findById(id).count())
                        .map(count -> count.intValue() > 0);
            });
}

2. Controller层优化 #

增加校验异常处理，避免非法参数直接抛出未捕获异常：

@ResponseStatus(HttpStatus.CREATED)
@PostMapping("/abc")
public Mono<ResponseEntity<Void>> update(@RequestParam Long id, @RequestBody Flux<Body> body) {
    return this.service.update(body, id)
            .log()
            .map(success -> success ?
                    new ResponseEntity<Void>(HttpStatus.CREATED) :
                    new ResponseEntity<Void>(HttpStatus.INTERNAL_SERVER_ERROR))
            // 校验不合法时返回标准400状态码
            .onErrorResume(IllegalArgumentException.class, e ->
                    Mono.just(new ResponseEntity<>(HttpStatus.BAD_REQUEST)));
}

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