使用Web Flux和Mongo DB保存多个记录

Question

我正在做一个使用Spring web Flux和mongo DB的项目，我对反应编程和WebFlux非常陌生。

我有使用一个服务保存到3个集合的场景。对于每个集合，im使用序列生成id，然后保存它们。我有FieldMaster，上面有列表，每个字段信息都有列表。我需要保存FieldMaster、FileInfo和FieldOption。下面是我正在使用的代码。代码仅在调试模式下运行时才起作用，否则会在行以下阻塞

    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

首先，在反应编程中是不好用的。块，因为你把非阻塞代码变成了阻塞。如果您想从一个流中获取并保存在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);
}