如何取消正在进行的Spring Flux？

2019-07-27 01:28:40 阅读：266 来源： 互联网

标签：project-reactor spring-webflux spring reactive-programming

我正在使用spring flux向服务发送并行请求,这是非常简化的版本：

Flux.fromIterable(customers)
  .flatMap { customer ->
     client.call(customer)
  } ...

我想知道如何取消这种通量,就像在某种程度上获取对通量的引用并告诉它关闭一样.

解决方法:

您可能知道,对于反应对象,all operators are lazy.这意味着管道的执行会延迟到您订阅反应流的那一刻.

所以,在你的例子中,没有什么可以取消,因为那时没有发生任何事情.

但假设您的示例扩展到：

Disposable disp = Flux.fromIterable(customers)
  .flatMap { customer ->
     client.call(customer)
  }
  .subscribe();

然后,正如您所看到的,您的订阅会返回一个Disposable对象,您可以根据需要使用该对象取消整个对象,例如：

disp.dispose()

dispose的文件说：

Cancel or dispose the underlying task or resource.

有another section of the documentation表示以下内容：

These variants [of operators] return a reference to the subscription
that you can use to cancel the subscription when no more data is
needed. Upon cancellation, the source should stop producing values and
clean up any resources it created. This cancel and clean-up behavior
is represented in Reactor by the general-purpose Disposable interface.

因此,取消流的执行并非在反应对象方面没有复杂性,因为如果在处理过程中取消流,则需要确保让世界保持一致状态.例如,如果您正在构建某些内容,则可能需要丢弃资源,销毁任何部分聚合结果,关闭文件,通道,释放内存或您拥有的任何其他资源,可能会撤消更改或对其进行补偿.

您可能希望阅读有关此内容的cleanup文档,以便您也考虑在反应对象方面可以执行的操作.

Flux<String> bridge = Flux.create(sink -> {
    sink.onRequest(n -> channel.poll(n))
        .onCancel(() -> channel.cancel()) 
        .onDispose(() -> channel.close())  
    });

标签：project-reactor,spring-webflux,spring,reactive-programming
来源： https://codeday.me/bug/20190727/1549462.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

如何取消正在进行的Spring Flux？