如何读取Flux<DataBuffer>内容？解析Multipart/FormData的Part为String遇困

1. bufferToStr方法的返回值会是null吗？ #

会。Flux.subscribe()是异步执行的，调用bufferToStr后方法会立刻返回res.get()，此时subscribe中的逻辑还未执行，AtomicReference<String> res的初始值为null，所以返回null是大概率会出现的情况。

2. 正确将Part解析为String的方法 #

方案1：符合Reactive设计的异步处理（推荐） #

将getFormData改为返回Mono<Map<String, String>>，用Reactive API安全拼接DataBuffer并转换为String：

private Mono<Map<String, String>> getFormData(String path, MultiValueMap<String, Part> partMultiValueMap) {
    if (partMultiValueMap == null) {
        return Mono.just(Collections.emptyMap());
    }
    Map<String, Part> multiPartMap = partMultiValueMap.toSingleValueMap();
    return Flux.fromIterable(multiPartMap.entrySet())
            .flatMap(entry -> {
                String key = entry.getKey();
                Part part = entry.getValue();
                if (part instanceof FormFieldPart) {
                    return Mono.just(new AbstractMap.SimpleEntry<>(key, ((FormFieldPart) part).value()));
                } else {
                    return dataBufferToString(part.content())
                            .map(str -> new AbstractMap.SimpleEntry<>(key, str));
                }
            })
            .collectMap(Map.Entry::getKey, Map.Entry::getValue);
}

private Mono<String> dataBufferToString(Flux<DataBuffer> content) {
    return content
            .reduce(DataBuffer::write)
            .map(buffer -> {
                byte[] bytes = new byte[buffer.readableByteCount()];
                buffer.read(bytes);
                DataBufferUtils.release(buffer);
                return new String(bytes, StandardCharsets.UTF_8);
            });
}

方案2：同步阻塞处理（仅适用于允许阻塞的场景） #

如果必须保持方法的同步返回类型，可使用block()等待Flux执行完成，但会破坏WebFlux的非阻塞特性，仅在特殊场景下使用：

private String bufferToStr(Flux<DataBuffer> content){
    return content
            .reduce(DataBuffer::write)
            .map(buffer -> {
                byte[] bytes = new byte[buffer.readableByteCount()];
                buffer.read(bytes);
                DataBufferUtils.release(buffer);
                return new String(bytes, StandardCharsets.UTF_8);
            })
            .block(); // 阻塞直到结果返回
}

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