java - 使用 RxJava 并行创建对象

Question

我使用 RxJava(聚合服务)编写了一个 Spring Boot 微服务来实现以下简化用例。总体情况是，当讲师上传类(class)内容文档时，应生成并保存一组问题。

• 用户将文档上传到系统。
• 系统调用文档服务将文档转换为文本。
• 然后它调用另一个问题生成服务来生成给定上述文本内容的一组问题。
• 最后将这些问题发布到一个基本的 CRUD 微服务中进行保存。

当用户上传文档时，会从中创建很多问题(可能有数百个左右)。这里的问题是我一次一个地发布问题，以便 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

默认情况下，flatMap 中的 observalbes 在您订阅它的同一调度器上运行。为了并行运行您的 createQuestion 可观察对象，您必须在计算调度程序上订阅它们。

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

检查 this article以获得完整的解释。