如何在Project Reactor中遍历HashMap并构建响应式管道？

原非响应式代码 #

Map<String, String> idToName = new HashMap<>();

idToName.put("1", "John");
idToName.put("2", "Harris");
idToName.put("3", "Sally");
idToName.put("4", "Mary");

idToName.forEach((id, name) -> {
    System.out.println("id: " + name + " name: " + name);
});

需求说明 #

将上述代码改为响应式实现，调用返回Mono<String>的process方法，最终返回List<Mono<String>>。process方法定义如下：

public Mono<String> process(String key, String val) {
    // do some processing
    return Mono.just(......);
}

响应式实现代码 #

import reactor.core.publisher.Mono;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class ReactorHashMapProcessing {

    public Mono<String> process(String key, String val) {
        // 示例处理逻辑：返回包含键值对的处理结果
        return Mono.just(String.format("Processed entry - Key: %s, Value: %s", key, val));
    }

    public static void main(String[] args) {
        ReactorHashMapProcessing processor = new ReactorHashMapProcessing();
        Map<String, String> idToName = new HashMap<>();

        idToName.put("1", "John");
        idToName.put("2", "Harris");
        idToName.put("3", "Sally");
        idToName.put("4", "Mary");

        // 构建响应式管道，生成List<Mono<String>>
        List<Mono<String>> monoList = idToName.entrySet()
                .stream()
                .map(entry -> processor.process(entry.getKey(), entry.getValue()))
                .collect(Collectors.toList());

        // 可选：订阅每个Mono以触发处理逻辑（Reactor是惰性执行的）
        monoList.forEach(mono -> mono.subscribe(System.out::println));
    }
}

关键说明 #

• 遍历HashMap：通过entrySet()获取所有键值对条目，结合Java Stream进行遍历
• 转换为Mono：使用map操作将每个条目映射为调用process方法后的Mono<String>实例
• 收集结果：通过Collectors.toList()将所有Mono<String>收集到列表中
• 惰性执行：Reactor中的Mono是惰性的，只有调用subscribe()方法后，process内的逻辑才会实际执行

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