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

我正在使用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