使用RxJava并行创建对象

Question

我使用RxJava（聚合服务）编写了一个Spring Boot微服务来实现以下简化用例。大局是当讲师上传课程内容文档时，应该生成并保存一组问题。

• 用户将文档上载到系统

当用户上载文档时，会从中创建许多问题（可能有数百个左右）。这里的问题是，我正在按顺序发布一个问题，以便CRUD服务保存它们。由于IO密集型网络调用，这会大大降低操作速度，因此完成整个过程大约需要20秒。以下是当前代码，假设所有问题都已制定。

questions.flatMapIterable(list -> list).flatMap(q -> createQuestion(q)).toList();

private Observable<QuestionDTO> createQuestion(QuestionDTO question) {
    return Observable.<QuestionDTO> create(sub -> {
        QuestionDTO questionCreated = restTemplate.postForEntity(QUESTIONSERVICE_API,
                new org.springframework.http.HttpEntity<QuestionDTO>(question), QuestionDTO.class).getBody();
        sub.onNext(questionCreated);
        sub.onCompleted();
    }).doOnNext(s -> log.debug("Question was created successfully."))
            .doOnError(e -> log.error("An ERROR occurred while creating a question: " + e.getMessage()));
}

现在我的要求是将所有问题并行发布到CRUD服务，并在完成时合并结果。还要注意，CRUD服务一次只接受一个问题对象，并且不能更改。我知道我可以为此目的使用Observable.zip运算符，但我不知道如何在这种情况下应用它，因为实际问题的数量不是预定的。我如何更改第1行中的代码，以便我可以提高应用程序的性能。任何帮助都是非常感谢的。

Answer 1

默认情况下，平面图中的观测值在您订阅的同一调度程序上运行。为了并行运行您的create询问观测值，您必须在计算调度程序上订阅它们。

questions.flatMapIterable(list -> list)
        .flatMap(q -> createQuestion(q).subscribeOn(Schedulers.computation()))
        .toList();

查看这篇文章以获得完整的解释。