java - 有没有一种方法可以成功执行嵌套通量操作而不实际阻塞您的代码？

Question

在使用 Spring Webflux 时，我尝试在 Realm 对象服务器中插入一些数据，该服务器通过 Rest API 与 Java 应用程序交互。所以基本上我有一组学生，他们有一组科目，我的目标是以非阻塞的方式坚持这些科目。因此，我使用通过休息端点公开的微服务，该服务为我提供了学生卷号的 Flux，对于该流量，我使用通过休息端点公开的另一个微服务，该微服务为我提供了科目的 Flux，对于每个科目，我想通过另一个休息端点将它们保留在 Realm 服务器中。我想让这一切都非常非阻塞，这就是为什么我希望我的代码看起来像这样。

void foo() {
studentService.getAllRollnumbers().flatMap(rollnumber -> {
    return subjectDirectory.getAllSubjects().map(subject -> {
        return dbService.addSubject(subject);
    })
});

}

但是由于某种原因这不起作用。但是一旦我对这些东西调用了 block ，它们就会就位，就像这样。

Flux<Done> foo() {
    List<Integer> rollNumbers = studentService.getAllRollnumbers().collectList().block();

    rollNumbers.forEach(rollNumber -> {
        List<Subject> subjects = subjectDirectory.getAllSubjects().collectList().block();

    subjects.forEach(subject -> {dbService.addSubject(subject).block();});
    });

    return Flux.just(new NotUsed());
}

<小时/>

getAllRollnumbers() returns a flux of integers.
getAllSubjects() returns a flux of subject.
and addSubject() returns a Mono of DBResponse pojo.

<小时/>

我能理解的是，执行此函数的线程在大部分内容被触发之前就已过期。请帮助我以异步非阻塞方式工作此代码。

Answer 1

您在第一个实例中根本没有订阅发布者，这就是它不执行的原因。你可以这样做:

studentService.getAllRollnumbers().flatMap(rollnumber -> {
    return subjectDirectory.getAllSubjects().map(subject -> {
        return dbService.addSubject(subject);
    })
}).subscribe();

但是，通常最好让框架来处理订阅，但如果没有看到其余代码，我无法建议。