Spring Webflux响应式流中单次创建带状态对象的实现合理性咨询

你的代码里虽然通过.share()+.block()让功能跑起来了，但这种写法不规范，核心问题是用block()破坏了响应式编程的非阻塞原则，还存在性能和线程安全隐患。下面具体拆解问题并给出正确实现：

当前代码的核心问题 #

• 阻塞调用违背响应式设计：block()会直接阻塞当前线程等待结果，彻底浪费Reactor的非并发优势，拖垮应用的并发处理能力。
• 冗余的cache()和share()：你试图用这两个操作让所有byte[]共享同一个加密实例，但逻辑绕了弯路，每个byte[]都触发一次订阅+阻塞，完全没必要。
• 状态安全风险：带状态的加密对象在多次block()调用中被重复获取，若调度器发生变化，可能出现并发访问导致状态错乱。

规范的实现方式 #

正确思路是：拿到KeyVault后一次性创建加密对象，让整个body的Flux复用这个带状态实例，全程用响应式操作串联，完全避免阻塞调用。

修改后的代码如下：

@Override
public Mono<FileResultDto> addEntry(final Flux<byte[]> body,
                                    final String fileId) {
    return keyVaultRepository.findByFiletId(fileId)
            .switchIfEmpty(Mono.defer(() -> {
                final KeyVault keyVault = KeyVault.of(fileId);
                return keyVaultRepository.save(keyVault);
            }))
            // 每个请求仅创建一次带状态加密对象，绑定到后续整个body流
            .flatMap(keyVault -> {
                final EncryptionState encryptionState = encryption.createEncryption(keyVault.getKey(), ENCRYPT_MODE);
                // 复用同一个加密实例处理所有byte[]
                Flux<byte[]> encryptedBody = body.map(encryptionState::update);
                return persistenceService.addEntry(encryptedBody, fileId);
            });
}

写法说明 #

• 单实例保证：每个请求对应一个KeyVault，在flatMap里创建唯一的加密对象，后续整个body流的所有byte[]都会复用这个实例，满足“每个请求仅创建一次”的要求。
• 非阻塞流程：全程用响应式操作串联，没有任何阻塞调用，完全符合Reactor的设计原则，能充分发挥并发性能。
• 逻辑简洁：直接把加密对象和body流绑定，避免了原代码中嵌套Mono、多次订阅的冗余逻辑，可读性和可维护性更强。

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