RxPY - Đồng thời sử dụng Bộ lập lịch

Một tính năng quan trọng của RxPy là đồng thời, tức là cho phép tác vụ thực thi song song. Để điều đó xảy ra, chúng tôi có hai toán tử subscribe_on () và Obser_on () sẽ hoạt động với một bộ lập lịch, sẽ quyết định việc thực thi tác vụ đã đăng ký.

Đây là một ví dụ làm việc cho thấy sự cần thiết của subscibe_on (), Obser_on () và bộ lập lịch.

Thí dụ

import random
import time
import rx
from rx import operators as ops
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a))
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a))
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
) 
input("Press any key to exit\n")

Trong ví dụ trên, tôi có 2 nhiệm vụ: Nhiệm vụ 1 và Nhiệm vụ 2. Việc thực hiện nhiệm vụ theo trình tự. Nhiệm vụ thứ hai chỉ bắt đầu khi nhiệm vụ đầu tiên được hoàn thành.

Đầu ra

E:\pyrx>python testrx.py
From Task 1: 1
From Task 1: 2
From Task 1: 3
From Task 1: 4
From Task 1: 5
Task 1 complete
From Task 2: 1
From Task 2: 2
From Task 2: 3
From Task 2: 4
Task 2 complete

RxPy hỗ trợ nhiều Scheduler và ở đây, chúng ta sẽ sử dụng ThreadPoolScheduler. ThreadPoolScheduler chủ yếu sẽ cố gắng quản lý với các luồng CPU có sẵn.

Trong ví dụ, chúng ta đã thấy trước đó, chúng ta sẽ sử dụng một mô-đun đa xử lý sẽ cung cấp cho chúng ta cpu_count. Số lượng sẽ được trao cho ThreadPoolScheduler sẽ quản lý để tác vụ hoạt động song song dựa trên các luồng có sẵn.

Đây là một ví dụ hoạt động -

import multiprocessing
import random
import time
from threading import current_thread
import rx
from rx.scheduler import ThreadPoolScheduler
from rx import operators as ops
# calculate cpu count, using which will create a ThreadPoolScheduler
thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count)
print("Cpu count is : {0}".format(thread_count))
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
)
input("Press any key to exit\n")

Trong ví dụ trên, tôi có 2 tác vụ và số cpu_count là 4. Vì, tác vụ là 2 và luồng có sẵn với chúng tôi là 4, cả hai tác vụ có thể bắt đầu song song.

Đầu ra

E:\pyrx>python testrx.py
Cpu count is : 4
Press any key to exit
From Task 1: 1
From Task 2: 1
From Task 1: 2
From Task 2: 2
From Task 2: 3
From Task 1: 3
From Task 2: 4
Task 2 complete
From Task 1: 4
From Task 1: 5
Task 1 complete

Nếu bạn thấy đầu ra, cả hai tác vụ đã bắt đầu song song.

Bây giờ, hãy xem xét một tình huống, trong đó nhiệm vụ nhiều hơn số CPU, tức là số CPU là 4 và nhiệm vụ là 5. Trong trường hợp này, chúng ta sẽ cần kiểm tra xem có luồng nào còn trống sau khi hoàn thành tác vụ không, vì vậy, nó có thể được giao cho nhiệm vụ mới có sẵn trong hàng đợi.

Với mục đích này, chúng ta có thể sử dụng toán tử Obser_on () sẽ quan sát bộ lập lịch nếu bất kỳ luồng nào còn trống. Đây là một ví dụ làm việc bằng cách sử dụng Obser_on ()

Thí dụ

import multiprocessing
import random
import time
from threading import current_thread
import rx
from rx.scheduler import ThreadPoolScheduler
from rx import operators as ops
# calculate cpu count, using which will create a ThreadPoolScheduler
thread_count = multiprocessing.cpu_count()
thread_pool_scheduler = ThreadPoolScheduler(thread_count)
print("Cpu count is : {0}".format(thread_count))
def adding_delay(value):
   time.sleep(random.randint(5, 20) * 0.1)
   return value
# Task 1
rx.of(1,2,3,4,5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 1: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 1 complete")
)
# Task 2
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 2: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 2 complete")
)
#Task 3
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 3: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 3 complete")
)
#Task 4
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.subscribe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 4: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 4 complete")
)
#Task 5
rx.range(1, 5).pipe(
   ops.map(lambda a: adding_delay(a)),
   ops.observe_on(thread_pool_scheduler)
).subscribe(
   lambda s: print("From Task 5: {0}".format(s)),
   lambda e: print(e),
   lambda: print("Task 5 complete")
)
input("Press any key to exit\n")

Đầu ra

E:\pyrx>python testrx.py
Cpu count is : 4
From Task 4: 1
From Task 4: 2
From Task 1: 1
From Task 2: 1
From Task 3: 1
From Task 1: 2
From Task 3: 2
From Task 4: 3
From Task 3: 3
From Task 2: 2
From Task 1: 3
From Task 4: 4
Task 4 complete
From Task 5: 1
From Task 5: 2
From Task 5: 3
From Task 3: 4
Task 3 complete
From Task 2: 3
Press any key to exit
From Task 5: 4
Task 5 complete
From Task 1: 4
From Task 2: 4
Task 2 complete
From Task 1: 5
Task 1 complete

Nếu bạn thấy kết quả đầu ra, thời điểm tác vụ 4 hoàn thành, luồng được trao cho tác vụ tiếp theo, tức là tác vụ 5 và cùng một bắt đầu thực hiện.