V2EX = way to explore V2EX 是一个关于分享和探索的地方
Sign Up Now For Existing Member Sign In
This topic created in 1584 days ago, the information mentioned may be changed or developed.
如下代码，本意是将 flux 流发送到子线程处理，再将处理结果汇聚到主线程。如何能够实现主线程处理报错时停止 Flux 的呢。
public static void main(String[] args) { String[] data = {"2", "2", "2", "0", "8", "9", "10", "11", "12", "13", "14", "15"}; Iterable<Integer> integers = Flux.fromArray(data) .flatMapSequential(s -> Mono.fromSupplier(() -> Integer.parseInt(s)).subscribeOn(Schedulers.boundedElastic()), 3) .doOnNext(s -> { System.out.println(Thread.currentThread().getName() + "---------->" + s); }) .toIterable(); for (Integer i : integers) { //如何实现这里报错时，停止 Flux System.out.println((10 / i) + "------>>>>" + Thread.currentThread().getName()); } }  | 1 yazinnnn Dec 28, 2021 fun main() { val data = arrayOf("2", "2", "2", "0", "8", "9", "10", "11", "12", "13", "14", "15") val integers = Flux.fromArray(data) .flatMapSequential({ s: String -> Mono.fromSupplier { s.toInt() }.subscribeOn(Schedulers.boundedElastic()) }, 3) val a = integers.doOnNext { s -> println(Thread.currentThread().name + "---------->" + s) }.subscribe() for (i in integers.toIterable()) { try { 10 / i } catch (t: Throwable) { a.dispose() } } CountDownLatch(1).await() } 不清楚你的具体需求，如果整个链路不使用 reactivestream 的话，似乎性能（吞吐）并没什么提高 toIterable()迭代时会阻塞当前线程，这样写跟直接用线程池处理比没啥优点 |
 | 2 Macolor21 Dec 28, 2021 CompletableFuture 最后 join 回主线程？ |
Select Language