Spring Webflux场景下向HashMap插入数据出现竞态条件问题咨询

核心思路 #

响应式编程规范中禁止在异步回调中修改共享可变状态，你遇到的竞态问题本质是多个异步返回结果同时修改同一个HashMap导致的，不需要加锁也不需要额外的并发集合操作，直接用Reactor内置的聚合操作符即可从根源规避问题。

最优方案：使用Flux聚合操作符直接生成结果Map #

你不需要自己提前声明Map来逐个填充，将20个Webflux调用返回的Mono合并为Flux后，直接用collectMultimap或自定义收集器生成目标结构即可，整个聚合过程由Reactor保证线程安全，无锁无阻塞，完全符合Webflux的设计规范。
示例代码如下：

// 1. 先构造所有文档上传请求的Mono列表
List<Mono<FileUploadLodgeReturnDetailsDTO>> uploadTasks = documentList.stream()
        .map(this::uploadSingleDocument) // 你的单个文档上传Webflux调用方法
        .toList();

// 2. 合并所有请求结果，直接聚合为目标Map结构
Mono<Map<DocumentType, List<FileUploadDetailsDTO>>> resultMap = Flux.merge(uploadTasks)
        .collect(
                // 初始容器
                () -> new HashMap<DocumentType, List<FileUploadDetailsDTO>>(),
                // 元素入库逻辑，单个聚合操作在Reactor中是线程安全的
                (map, dto) -> map.computeIfAbsent(
                        DocumentType.make(dto.getDocumentType()),
                        k -> new ArrayList<>()
                ).add(dto)
        );

后续你需要使用这个Map时，直接把resultMap拼接在你的响应式调用链中即可，不需要手动维护任何共享可变变量。

原有方案问题说明 #

• 你之前用ConcurrentHashMap出现元素丢失，是因为containsKey、get、put是三个独立的操作，不具备原子性，就算用ConcurrentHashMap也无法避免竞态。如果非要在回调中修改Map，应该用ConcurrentHashMap.computeIfAbsent这个原子操作来插入元素，不会出现丢数据的问题，但依然属于共享可变状态的写法，不推荐在Webflux中使用。
• 加synchronized会阻塞Webflux的事件循环线程，直接丧失响应式架构的非阻塞性能优势，完全不可取。

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