我正在嘗試學(xué)習(xí) RxJava2 庫的基礎(chǔ)知識(shí)，現(xiàn)在我陷入了以下時(shí)刻：我已經(jīng)生成了myFlowablevia Flowable.generate(...)，現(xiàn)在我需要等待所有任務(wù)完成執(zhí)行，然后才能繼續(xù)下一步。這是展示問題的代碼：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(); }我運(yùn)行所有任務(wù)并將結(jié)果添加到集合中。如果我在 myFlowable 塊之后立即檢查集合大小，那么如果我在small之后檢查它，情況將會(huì)有所不同Thread.sleep()。有什么方法可以檢查所有任務(wù)是否已完成執(zhí)行并且我們可以進(jìn)一步進(jìn)行？任何幫助或指導(dǎo)將不勝感激。
3 回答
拉風(fēng)的咖菲貓
TA貢獻(xiàn)1995條經(jīng)驗(yàn) 獲得超2個(gè)贊
使用Flowable::blockingSubscribe()- 將當(dāng)前 Flowable 運(yùn)行到終端事件，忽略任何值并重新拋出任何異常。
守著星空守著你
TA貢獻(xiàn)1799條經(jīng)驗(yàn) 獲得超8個(gè)贊
由于 RxJava 是異步的，observable 下面的 java 代碼將運(yùn)行，而 observable 將在不同的線程中運(yùn)行，這就是為什么如果您想在 Flowable 已完成發(fā)送數(shù)據(jù)時(shí)收到通知，您應(yīng)該在 RxJava 流中執(zhí)行此操作。為此，您有一個(gè)運(yùn)算符 .doOnComplete 這里有一個(gè)示例如何檢測(cè)流何時(shí)完成
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貢獻(xiàn)1863條經(jīng)驗(yàn) 獲得超2個(gè)贊
您可以使用 AtomicBoolean，將其初始化為 false 并使用 將其設(shè)置為 true doFinally()。
doFinally()在 Observable 發(fā)出 onError 或 onCompleted 信號(hào)后調(diào)用，或者被下游處理。
然后讓主線程休眠，直到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();
}
添加回答
舉報(bào)
0/150
提交
取消