Spring R2DBC中如何将Flux转换为另一类型Flux后存入数据仓库？

1. 测试方法执行提前终止：响应式流的subscribe()是异步非阻塞操作，调用后仅触发订阅逻辑，不会阻塞当前线程等待后续IO操作完成。测试主线程执行完subscribe()后会直接退出，JVM进程终止，自然不会等待后续的Post遍历、转换、保存逻辑执行，这也是你只能看到第一个Post被处理、没有数据落库的核心原因。
2. 工具类存在索引越界风险：extractTagsReactive方法中如果入参tags为空或者处理后生成的标签列表长度为0，调用tagList.remove(tagList.size()-1)会直接抛出索引越界异常，直接终止响应式流的执行。
3. 测试事务默认回滚：如果你在测试类上加了@Transactional注解，Spring测试默认执行完所有逻辑后会自动回滚事务，就算保存逻辑执行完成也不会实际落库。

1. 修复工具类异常风险 #

先对extractTagsReactive做边界处理，避免索引越界：

public static Flux<String> extractTagsReactive(String tags){
    if (org.springframework.util.StringUtils.isBlank(tags)) {
        return Flux.empty();
    }
    // 合并两次替换操作，减少字符串创建开销
    tags = tags.replaceAll("<","").replaceAll(">",",");
    Set<String> tagSet = org.springframework.util.StringUtils.commaDelimitedListToSet(tags);
    if (tagSet.isEmpty()) {
        return Flux.empty();
    }
    List<String> tagList = new LinkedList<>(tagSet);
    // 仅当列表长度大于1时才移除最后一个元素，兼容原有业务逻辑同时避免越界
    if (tagList.size() > 1) {
        tagList.remove(tagList.size() - 1);
    }
    return Flux.fromIterable(tagList);
}

2. 修复测试方法的等待逻辑 #

测试场景下不要使用subscribe()，改用blockLast()阻塞当前线程，等待整个响应式流所有元素处理完成：

@Test
// 如果需要实际落库，加@Rollback(false)关闭测试自动回滚
@Rollback(false)
public void creatPostTagsReactor(){
    Map<String, Tag> nameMapTag = tagRepository.findAll().collectList().block()
            .parallelStream()
            .collect(Collectors.toMap(t-> t.getTagName(), t-> t));
    Flux<PostTags> postTagsFlux =  postRepository.findAll()
            // 如果你需要严格按照Post顺序处理，可以把flatMap换成concatMap
            .flatMap(p-> convertToPostTags(p, nameMapTag), 10); // 第二个参数控制并发数，避免数据库连接被打满
    // 阻塞等待所有保存操作完成
    postTagsRepository.saveAll(postTagsFlux).blockLast();
}

3. 事务适配（生产环境用） #

如果是生产环境的Service逻辑，加@Transactional注解时要确保项目已经配置了响应式事务管理器ReactiveTransactionManager，传统的JDBC事务管理器不支持响应式链路的事务传播。

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