java - Spring react 器: What's the corresponding class to Optional?

Question

所以我有一个 Flux<Foo>我想映射每个 Foo至Baz 。问题是，getBaz(Foo foo)可能会抛出 IOException .

所以我想到了 Mono<Baz> getBazRx(Foo foo)方法将返回 Mono.just(baz)或Mono.empty()如有异常。

然后将得到 Flux<Mono<Baz>>哪种提醒Optional<T>容器。

这是 Spring Reactor 中的做法吗？如何正确食用？

Answer 1

在响应式(Reactive)流中，“可选”通常通过从流中删除不存在的元素来处理(例如空的 Mono ，或删除了元素的 Flux 。)，而不是拥有 Flux<Optional> 。 , Mono<Optional> ，或Flux<Mono>

调用同步时getBaz方法，您可以使用单个 .handle操作如下:

flux
    .handle((foo, sink) -> {
        try {
            // propagate Baz down the stream
            sink.next(getBaz(foo));
        } catch (IOException e) {
            // Since sink.next is not called here,
            // the problematic element will be dropped from the stream
            log.error(e);
        }
    })

调用异步时getBazRx方法(返回 Mono )，您可以使用 onErrorResume里面 flatMap/flatMapSequential/concatMap ，像这样:

flux
    .flatMap(foo -> getBazRx(foo)
        .onErrorResume(t -> {
            log.error(t);
            return Mono.empty();
        }))

(或者您可以将 .onErrorResume 移动到 .getBazRx 内，具体取决于您想要捕获并忽略异常的位置)

另外，既然您在问题中提到了它...如果您要创建 getBazRx包裹 getBaz ，如果 getBaz 你不应该永远做这样的事情有可能阻止:

Mono<Baz> getBazRx(Foo foo) {
    // BAD!!!
    try {
        return Mono.just(getBaz(foo));
    } catch (IOException e) {
        return Mono.error(e)  // or Mono.empty() if you want to ignore
    }
}

该实现实际上只是一个模拟异步方法的同步方法。它有两个问题:

1. 工作立即完成，而不是在订阅返回的Mono之后
2. 如果getBaz block ，您最终可能会阻塞事件循环

相反，您应该推迟工作直到订阅了 mono，并在 Scheduler 上运行任何阻塞操作。用于阻塞操作，如下所示:

Mono<Baz> getBazRx(Foo foo) {
    return Mono.fromSupplier(() -> {
            try {
                return getBaz(foo);
            } catch (IOException e) {
                throw Exceptions.propagate(e);  // or return null to ignore and complete empty
            }
        })
        .subscribeOn(Schedulers.elastic());  // run on a scheduler suitable for blocking work
}