反应堆取消订阅的方式

反应堆取消订阅的方式

在Reactor中，

Subscription

在您调用之前取消a是没有意义的

subscribe()

（因为正是这种方法创建了，

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)114:17:48.732 [main] INFO  reactor.Flux.Array.1 - | onNext(2)214:17:48.732 [main] INFO  reactor.Flux.Array.1 - | cancel()