Reactor 取消订阅的方式

Question

我试图搞清楚 Reactor 项目，我正在寻找取消订阅的方法。我知道在订阅例如 Flux 之后，我可以获得对 Cancellation 对象的引用，该对象可用于发送 onCancel 信号，但这只是在订阅之后，我需要在某种集合中保存该引用。

有没有更好的方式获取Cancellation对象？或者只是取消订阅。也许某个地方包含对所有活动订阅的引用 - 是的，那会很棒......

Answer 1

在 Reactor 中，在您调用 subscribe() 之前想要取消 Subscription 是没有意义的（因为正是这个方法创建了 Subscription 并传播它在链上发出信号以开始发送数据）。

所有订阅都没有一个集中的地方，这没有多大意义，因为您需要一种方法来找到要取消的特定订阅（请记住，您链中的每个运营商都可以使用还有一个中级订阅...)。

请注意，一些运营商也会代表您取消订阅！例如 take(int) 就是这种情况，一旦发出了足够多的项目，它将取消上游：

Flux.just(1, 2, 3, 4).log().take(2).subscribe(System.out::println);

将输出：

14:17:48.729 [main] INFO  reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
14:17:48.732 [main] INFO  reactor.Flux.Array.1 - | request(unbounded)
14:17:48.732 [main] INFO  reactor.Flux.Array.1 - | onNext(1)
1
14:17:48.732 [main] INFO  reactor.Flux.Array.1 - | onNext(2)
2
14:17:48.732 [main] INFO  reactor.Flux.Array.1 - | cancel()

Reactor 取消订阅的方式

Reactor way to cancel Subscriptions

java

reactive-programming

project-reactor