取消订阅的反应堆方式答案

【问题标题】：Reactor way to cancel Subscriptions取消订阅的反应堆方式
【发布时间】：2016-09-01 19:47:22
【问题描述】：

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

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

【问题讨论】：

标签： java reactive-programming project-reactor

【解决方案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()

【讨论】：

哦，我不知道那个运算符 :O 但对我来说最重要的是 Cancellation 对象正在发送我可以正确处理的 OnCancel 信号。但是好的，我找到了解决问题的方法 - 我通过向 Flux 中抛出异常来取消流，所以这还不错
我建议您更喜欢使用Cancellation 对象。请注意，它将在 3.1 中变为 Disposable（此时您必须调用 dispose() 而不是 cancel()）。或者寻找可以自然匹配您想要做的操作的运算符，并在需要时为您取消......根据您的用例，将异常抛出到 Flux 中听起来并不是一个很好的解决方案。
@SimonBaslé 我运行您的代码，如果take(2) 在log() 之前，则不会打印cancel() 信号。为什么？你说take 运算符取消了通量，而不是源。
恰恰相反，这正是我所说的：take 取消上游，也就是源。日志低于，它是下游而不是上游，因此它看不到取消