我正在嘗試學習 RxJava2 庫的基礎知識，現在我陷入了以下時刻：我已經生成了myFlowablevia Flowable.generate(...)，現在我需要等待所有任務完成執行，然后才能繼續下一步。這是展示問題的代碼：myFlowable.parallel() .runOn(Schedulers.computation()) .map(val -> myCollection.add(val)) .sequential() .subscribe(val -> { System.out.println("Thread from subscribe: " + Thread.currentThread().getName()); System.out.println("Value from subscribe: " + val.toString()); }); System.out.println("Before sleep - Number of objects: " + myCollection.size()); try { Thread.sleep(1000); System.out.println("After sleep - Number of objects: " + myCollection.size()); } catch (InterruptedException e) { e.printStackTrace(); }我運行所有任務并將結果添加到集合中。如果我在 myFlowable 塊之后立即檢查集合大小，那么如果我在small之后檢查它，情況將會有所不同Thread.sleep()。有什么方法可以檢查所有任務是否已完成執行并且我們可以進一步進行？任何幫助或指導將不勝感激。
3 回答
守著星空守著你
TA貢獻1799條經驗 獲得超8個贊
由于 RxJava 是異步的，observable 下面的 java 代碼將運行，而 observable 將在不同的線程中運行，這就是為什么如果您想在 Flowable 已完成發送數據時收到通知，您應該在 RxJava 流中執行此操作。為此，您有一個運算符 .doOnComplete 這里有一個示例如何檢測流何時完成
Flowable.range(0, 100).parallel()
.runOn(Schedulers.computation())
.map(integer -> {
return integer;
})
.sequential()
.doOnComplete(() -> {
System.out.println("finished");
})
.subscribe(integer -> System.out.println(integer));
小唯快跑啊
TA貢獻1863條經驗 獲得超2個贊
您可以使用 AtomicBoolean，將其初始化為 false 并使用 將其設置為 true doFinally()。
doFinally()在 Observable 發出 onError 或 onCompleted 信號后調用，或者被下游處理。
然后讓主線程休眠，直到completedvalue 為 true。
使用你的例子：
AtomicBoolean completed = new AtomicBoolean(false);
myFlowable.parallel()
.runOn(Schedulers.computation())
.map(val -> myCollection.add(val))
.sequential()
.doFinally(() -> completed.set(true))
.subscribe(val -> {
...
});
...
try {
while(!completed.get()){
Thread.sleep(1000);
...
}
...
} catch (InterruptedException e) {
e.printStackTrace();
}
添加回答
舉報
0/150
提交
取消