Đồng tiền trong Python - Nhóm các quy trình
Nhóm quy trình có thể được tạo và sử dụng theo cách giống như chúng ta đã tạo và sử dụng nhóm quy trình. Nhóm quy trình có thể được định nghĩa là nhóm các quy trình khởi tạo trước và không hoạt động, sẵn sàng để được giao công việc. Tạo nhóm quy trình được ưu tiên hơn là tạo quy trình mới cho mọi tác vụ khi chúng ta cần thực hiện một số lượng lớn tác vụ.
Mô-đun Python - Concurrent.futures
Thư viện chuẩn Python có một mô-đun được gọi là concurrent.futures. Mô-đun này đã được thêm vào Python 3.2 để cung cấp cho các nhà phát triển một giao diện cấp cao để khởi chạy các tác vụ không đồng bộ. Nó là một lớp trừu tượng trên đầu các mô-đun phân luồng và đa xử lý của Python để cung cấp giao diện để chạy các tác vụ bằng cách sử dụng nhóm luồng hoặc quy trình.
Trong các phần tiếp theo, chúng ta sẽ xem xét các lớp con khác nhau của mô-đun concurrent.futures.
Lớp thừa hành
Executor là một lớp trừu tượng của concurrent.futuresMô-đun Python. Nó không thể được sử dụng trực tiếp và chúng tôi cần sử dụng một trong các lớp con bê tông sau:
- ThreadPoolExecutor
- ProcessPoolExecutor
ProcessPoolExecutor - Một lớp con cụ thể
Nó là một trong những lớp con cụ thể của lớp Executor. Nó sử dụng đa xử lý và chúng tôi nhận được một nhóm các quy trình để gửi nhiệm vụ. Nhóm này chỉ định nhiệm vụ cho các quy trình có sẵn và lên lịch để chúng chạy.
Làm cách nào để tạo ProcessPoolExecutor?
Với sự giúp đỡ của concurrent.futures mô-đun và lớp con cụ thể của nó Executor, chúng ta có thể dễ dàng tạo một nhóm quy trình. Đối với điều này, chúng ta cần xây dựng mộtProcessPoolExecutorvới số lượng quy trình chúng tôi muốn trong nhóm. Theo mặc định, số là 5. Tiếp theo là gửi một nhiệm vụ đến nhóm quy trình.
Thí dụ
Bây giờ chúng ta sẽ xem xét cùng một ví dụ mà chúng ta đã sử dụng khi tạo nhóm luồng, sự khác biệt duy nhất là bây giờ chúng ta sẽ sử dụng ProcessPoolExecutor thay vì ThreadPoolExecutor .
from concurrent.futures import ProcessPoolExecutor
from time import sleep
def task(message):
sleep(2)
return message
def main():
executor = ProcessPoolExecutor(5)
future = executor.submit(task, ("Completed"))
print(future.done())
sleep(2)
print(future.done())
print(future.result())
if __name__ == '__main__':
main()Đầu ra
False
False
CompletedTrong ví dụ trên, một Quy trìnhPoolExecutorđã được xây dựng với 5 chủ đề. Sau đó, một nhiệm vụ, sẽ đợi trong 2 giây trước khi đưa ra thông báo, sẽ được gửi cho người thực thi nhóm tiến trình. Như đã thấy từ đầu ra, nhiệm vụ không hoàn thành cho đến 2 giây, vì vậy cuộc gọi đầu tiên đếndone()sẽ trả về Sai. Sau 2 giây, nhiệm vụ được hoàn thành và chúng tôi nhận được kết quả của tương lai bằng cách gọiresult() phương pháp trên đó.
Instantiating ProcessPoolExecutor - Trình quản lý ngữ cảnh
Một cách khác để khởi tạo ProcessPoolExecutor là với sự trợ giúp của trình quản lý ngữ cảnh. Nó hoạt động tương tự như phương pháp được sử dụng trong ví dụ trên. Ưu điểm chính của việc sử dụng trình quản lý ngữ cảnh là nó trông tốt về mặt cú pháp. Việc khởi tạo có thể được thực hiện với sự trợ giúp của đoạn mã sau:
with ProcessPoolExecutor(max_workers = 5) as executorThí dụ
Để hiểu rõ hơn, chúng tôi đang lấy ví dụ tương tự như được sử dụng trong khi tạo nhóm luồng. Trong ví dụ này, chúng ta cần bắt đầu bằng cách nhậpconcurrent.futuresmô-đun. Sau đó, một hàm có tênload_url()được tạo sẽ tải url được yêu cầu. CácProcessPoolExecutorsau đó được tạo với 5 số luồng trong nhóm. Quá trìnhPoolExecutorđã được sử dụng làm trình quản lý ngữ cảnh. Chúng ta có thể nhận được kết quả của tương lai bằng cách gọiresult() phương pháp trên đó.
import concurrent.futures
from concurrent.futures import ProcessPoolExecutor
import urllib.request
URLS = ['http://www.foxnews.com/',
'http://www.cnn.com/',
'http://europe.wsj.com/',
'http://www.bbc.co.uk/',
'http://some-made-up-domain.com/']
def load_url(url, timeout):
with urllib.request.urlopen(url, timeout = timeout) as conn:
return conn.read()
def main():
with concurrent.futures.ProcessPoolExecutor(max_workers=5) as executor:
future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
for future in concurrent.futures.as_completed(future_to_url):
url = future_to_url[future]
try:
data = future.result()
except Exception as exc:
print('%r generated an exception: %s' % (url, exc))
else:
print('%r page is %d bytes' % (url, len(data)))
if __name__ == '__main__':
main()Đầu ra
Tập lệnh Python ở trên sẽ tạo ra kết quả sau:
'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229476 bytes
'http://www.cnn.com/' page is 165323 bytes
'http://www.bbc.co.uk/' page is 284981 bytes
'http://europe.wsj.com/' page is 967575 bytesSử dụng hàm Executor.map ()
Con trăn map()chức năng được sử dụng rộng rãi để thực hiện một số nhiệm vụ. Một trong những nhiệm vụ như vậy là áp dụng một hàm nhất định cho mọi phần tử trong các vòng lặp. Tương tự, chúng ta có thể ánh xạ tất cả các phần tử của một trình vòng lặp tới một hàm và gửi chúng dưới dạng các công việc độc lập choProcessPoolExecutor. Hãy xem xét ví dụ sau về tập lệnh Python để hiểu điều này.
Thí dụ
Chúng tôi sẽ xem xét cùng một ví dụ mà chúng tôi đã sử dụng trong khi tạo nhóm luồng bằng cách sử dụng Executor.map()chức năng. Trong ví dụ dưới đây, hàm bản đồ được sử dụng để áp dụngsquare() hàm cho mọi giá trị trong mảng giá trị.
from concurrent.futures import ProcessPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
return n * n
def main():
with ProcessPoolExecutor(max_workers = 3) as executor:
results = executor.map(square, values)
for result in results:
print(result)
if __name__ == '__main__':
main()Đầu ra
Tập lệnh Python ở trên sẽ tạo ra kết quả sau
4
9
16
25Khi nào sử dụng ProcessPoolExecutor và ThreadPoolExecutor?
Bây giờ chúng ta đã nghiên cứu về cả hai lớp Executor - ThreadPoolExecutor và ProcessPoolExecutor, chúng ta cần biết khi nào sử dụng trình thực thi nào. Chúng ta cần chọn ProcessPoolExecutor trong trường hợp khối lượng công việc bị ràng buộc bởi CPU và ThreadPoolExecutor trong trường hợp khối lượng công việc bị ràng buộc bởi I / O.
Nếu chúng ta sử dụng ProcessPoolExecutor, thì chúng ta không cần phải lo lắng về GIL vì nó sử dụng đa xử lý. Hơn nữa, thời gian thực hiện sẽ ít hơn khi so sánh vớiThreadPoolExecution. Hãy xem xét ví dụ tập lệnh Python sau để hiểu điều này.
Thí dụ
import time
import concurrent.futures
value = [8000000, 7000000]
def counting(n):
start = time.time()
while n > 0:
n -= 1
return time.time() - start
def main():
start = time.time()
with concurrent.futures.ProcessPoolExecutor() as executor:
for number, time_taken in zip(value, executor.map(counting, value)):
print('Start: {} Time taken: {}'.format(number, time_taken))
print('Total time taken: {}'.format(time.time() - start))
if __name__ == '__main__':
main()Đầu ra
Start: 8000000 Time taken: 1.5509998798370361
Start: 7000000 Time taken: 1.3259999752044678
Total time taken: 2.0840001106262207
Example- Python script with ThreadPoolExecutor:
import time
import concurrent.futures
value = [8000000, 7000000]
def counting(n):
start = time.time()
while n > 0:
n -= 1
return time.time() - start
def main():
start = time.time()
with concurrent.futures.ThreadPoolExecutor() as executor:
for number, time_taken in zip(value, executor.map(counting, value)):
print('Start: {} Time taken: {}'.format(number, time_taken))
print('Total time taken: {}'.format(time.time() - start))
if __name__ == '__main__':
main()Đầu ra
Start: 8000000 Time taken: 3.8420000076293945
Start: 7000000 Time taken: 3.6010000705718994
Total time taken: 3.8480000495910645Từ kết quả đầu ra của cả hai chương trình trên, chúng ta có thể thấy sự khác biệt về thời gian thực thi khi sử dụng ProcessPoolExecutor và ThreadPoolExecutor.