使用 Webflux 和 Mongodb 保存多条记录

Question

我正在开发一个使用 Spring web Flux 和 mongo DB 的项目，而且我对响应式编程和 WebFlux 非常陌生。

我有使用一项服务保存到 3 个集合中的场景。对于每个集合，我使用一个序列生成 id，然后保存它们。我有 FieldMaster，上面有 List ，每个 Field Info 都有 List 。我需要保存 FieldMaster、FileInfo 和 FieldOption。下面是我正在使用的代码。该代码仅在我在调试模式下运行时才有效，否则它会在下面的行中被阻止

整数字段_seq_id = Integer.parseInt(sequencesCollection.getNextSequence(FIELDINFO).block().getSeqValue());

这是完整的代码

public Mono createMasterData(Mono fieldmaster) {

    return fieldmaster.flatMap(fm -> {
        return sequencesCollection.getNextSequence(FIELDMASTER).flatMap(seqVal -> {
            LOGGER.info("Generated Sequence value :" + seqVal.getSeqValue());
            fm.setId(Integer.parseInt(seqVal.getSeqValue()));
            List<FieldInfo> fieldInfo = fm.getFieldInfo();
            fieldInfo.forEach(field -> {
                // saving Field Goes Here
                Integer field_seq_id = Integer.parseInt(sequencesCollection.getNextSequence(FIELDINFO).block().getSeqValue()); // stops execution at this line
                LOGGER.info("Generated Sequence value  Field Sequence:" + field_seq_id);
                field.setId(field_seq_id);
                field.setMasterFieldRefId(fm.getId());
                mongoTemplate.save(field).block();
                LOGGER.info("Field Details Saved");
                List<FieldOption> fieldOption = field.getFieldOptions();
                fieldOption.forEach(option -> {
                    // saving Field Option Goes Here
                    Integer opt_seq_id = Integer.parseInt(sequencesCollection.getNextSequence(FIELDOPTION).block().getSeqValue());
                    LOGGER.info("Generated Sequence value Options Sequence:" + opt_seq_id);
                    option.setId(opt_seq_id);
                    option.setFieldRefId(field_seq_id);
                    mongoTemplate.save(option).log().block();
                    LOGGER.info("Field Option Details Saved");
                });
            });
            return mongoTemplate.save(fm).log();
        });
    });

}

Answer 1

在响应式编程中首先使用 .block 并不好，因为您将非阻塞代码转换为阻塞代码。如果你想从一个流中获取并保存在 3 个流中，你可以这样做。 出于性能目的，有许多不同的方法可以做到这一点，但这取决于数据量。 在这里，您有一个使用简单数据和使用 concat 运算符的示例，但甚至还有 zip 和 merge。这取决于您的需求。

  public void run(String... args) throws Exception {
    Flux<Integer> dbData = Flux.range(0, 10);

    dbData.flatMap(integer -> Flux.concat(saveAllInFirstCollection(integer), saveAllInSecondCollection(integer), saveAllInThirdCollection(integer))).subscribe();
}

Flux<Integer> saveAllInFirstCollection(Integer integer) {
    System.out.println(integer);
    //process and save in collection
    return Flux.just(integer);
}

Flux<Integer> saveAllInSecondCollection(Integer integer) {
    System.out.println(integer);
    //process and save in collection
    return Flux.just(integer);
}

Flux<Integer> saveAllInThirdCollection(Integer integer) {
    System.out.println(integer);
    //process and save in collection
    return Flux.just(integer);
}