基于这个出色的SO答案,我可以让多个任务在RxPy中并行工作,我的问题是如何等待所有任务完成?我知道我可以使用线程,.join()
但是Rx Schedulers似乎没有这种选择。.to_blocking()
两者都不起作用,MainThread在所有通知都已触发并调用了完整的处理程序之前完成。这是一个例子:
from __future__ import print_function
import os, sys
import time
import random
from rx import Observable
from rx.core import Scheduler
from threading import current_thread
def printthread(val):
print("{}, thread: {}".format(val, current_thread().name))
def intense_calculation(value):
printthread("calc {}".format(value))
time.sleep(random.randint(5, 20) * .1)
return value
if __name__ == "__main__":
Observable.range(1, 3) \
.select_many(lambda i: Observable.start(lambda: intense_calculation(i), scheduler=Scheduler.timeout)) \
.observe_on(Scheduler.event_loop) \
.subscribe(
on_next=lambda x: printthread("on_next: {}".format(x)),
on_completed=lambda: printthread("on_completed"),
on_error=lambda err: printthread("on_error: {}".format(err)))
printthread("\nAll done")
# time.sleep(2)
calc 1, thread: Thread-1
calc 2, thread: Thread-2
calc 3, thread: Thread-3
on_next: 2, thread: Thread-4
on_next: 3, thread: Thread-4
on_next: 1, thread: Thread-4
on_completed, thread: Thread-4
All done, thread: MainThread
calc 1, thread: Thread-1
calc 2, thread: Thread-2
calc 3, thread: Thread-3
All done, thread: MainThread
calc 1, thread: Thread-1
calc 2, thread: Thread-2
calc 3, thread: Thread-3
All done, thread: MainThread
on_next: 2, thread: Thread-4
on_next: 3, thread: Thread-4
on_next: 1, thread: Thread-4
on_completed, thread: Thread-4
对于ThreadPoolScheduler
,您可以:
scheduler.executor.shutdown()
然后,完成所有操作后即可获得所有结果。
本文收集自互联网,转载请注明来源。
如有侵权,请联系 [email protected] 删除。
我来说两句