Как ждать завершения параллельных потоков RxPy

Question

Основываясь на этом превосходном ответе 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

Answer 1

За ThreadPoolScheduler, вы можете:

1. scheduler = ThreadPoolScheduler (pool_size)
2. параллельные звонки.
3. scheduler.executor.shutdown()

затем вы можете получить все результаты, как только все будет сделано.

Answer 2

Размещать полное решение здесь:

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
from rx.concurrency import ThreadPoolScheduler

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__":
    scheduler = ThreadPoolScheduler(4)

    Observable.range(1, 3) \
        .select_many(lambda i: Observable.start(lambda: intense_calculation(i), scheduler=scheduler)) \
        .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")
    scheduler.executor.shutdown()
    # time.sleep(2)

Answer 3

Использовать run() дождаться завершения параллельных потоков RxPy.

BlockingObservables был удален из RxPY v3.

      from threading import current_thread
import rx, random, multiprocessing, time
from rx import operators as ops

def intense_calculation(value):
   delay = random.randint(5, 20) * 0.2
   time.sleep(delay)
   print("From adding_delay: {0} Value : {1} {2}".format(current_thread(), value, delay))
   return (value[0], value[1]+ " processed")

thread_pool_scheduler = rx.scheduler.NewThreadScheduler()

my_dict={'A':'url1', 'B':'url2', 'C':'url3'}

new_dict = rx.from_iterable(my_dict.items()).pipe(
    ops.flat_map(lambda a: rx.of(a).pipe(
        ops.map(lambda a: intense_calculation(a)),
        ops.subscribe_on(thread_pool_scheduler)
    )),
    ops.to_dict(lambda x: x[0], lambda x: x[1])
).run()

print("From main: {0}".format(current_thread()))
print(str(new_dict))