Xử lý đồng thời trong python

Python thì không phải là phương án tối ưu nhất cho việc xử lý các process đồng thời. Tuy nhiên ở bài viết này mình sẽ viết thử 1 scripts nho nhỏ để kiểm chứng hiệu suất của việc xử lý concurrentcy này.

Scripts tải image đơn giản

import urllib.request
import time


url = 'https://picsum.photos/id/{}/200/300'
args = [(n, url.format(n)) for n in range(20)]
start = time.time()

for img_id, url in args:
    res = urllib.request.urlopen(url)
    pic = res.read()
    with open(f'./{img_id}.jpg', 'wb') as f:
        f.write(pic)
    print(f'Picture {img_id} saved!')

msg ='Spend time {:.3f}'
print(msg.format(time.time() - start))

scripts trên đang thực hiện download tuần tự 20 bức ảnh và output xuất ra màn hình

Picture 0 saved!
Picture 1 saved!
...

Picture 19 saved!

Spend time 4.491

Bây giờ mình sẽ viết 1 function cho công việc download image và sử dụng map để call function đó

import urllib.request
import time

def download_img(img_id, url):
    res = urllib.request.urlopen(url)
    pic = res.read()
    with open(f'./{img_id}.jpg', 'wb') as f:
        f.write(pic)
    print(f'Picture {img_id} saved!')

def main():
    url = 'https://picsum.photos/id/{}/200/300'
    args = [(n, url.format(n)) for n in range(20)]
    start = time.time()
    img_ids, urls = zip(*args)
    start = time.time()
    for _ in map(download_img, img_ids, urls):
        pass
    msg ='Spend time {:.3f}'
    print(msg.format(time.time() - start))

if __name__ == '__main__':
    main()

Và output vẫn là tuần tự công việc download image

Picture 0 saved!
Picture 1 saved!
...

Picture 19 saved!

Spend time 4.012

Trong cú pháp trên cách sử dụng map để thay thế cho for hơi khó hiểu. Tuy nhiên nó lại là bước đầu để bạn chuyển dần sang xử lý đồng thời.

Scripts xử lý đồng thời.

Để làm được điều này bạn cần phải biết tới module concurrent.futures bạn có thể tham khảo chi tiết tại https://docs.python.org/3/library/concurrent.futures.html?highlight=concurrent%20futures#executor-objects

Mình sẽ thay thể map bằng concurrency như sau

for _ in map(download_img, img_ids, urls):
    pass

Bằng

from concurrent.futures import ThreadPoolExecutor
...

with ThreadPoolExecutor(max_workers=20) as executor:
    executor.map(download_img, img_ids, urls)

function chi tiết như sau

import urllib.request
import time
from concurrent.futures import ThreadPoolExecutor

def download_img(img_id, url):
    res = urllib.request.urlopen(url)
    pic = res.read()
    with open(f'./{img_id}.jpg', 'wb') as f:
        f.write(pic)
    print(f'Picture {img_id} saved!')

def main():
    url = 'https://picsum.photos/id/{}/200/300'
    args = [(n, url.format(n)) for n in range(20)]
    start = time.time()
    img_ids, urls = zip(*args)
    start = time.time()
    with ThreadPoolExecutor(max_workers=20) as executor:
        executor.map(download_img, img_ids, urls)

    msg ='Spend time {:.3f}'
    print(msg.format(time.time() - start))

if __name__ == '__main__':
    main()

Output nhận được sẽ như sau

Picture 9 saved!
Picture 2 saved!
...
Picture 10 saved!

Spend time 0.209

Kết quả nhận được hoàn toàn khác xa lúc trước. Nếu xử lý tuần tự thì công việc download 20 images sẽ phải mất tới hơn 4s. Tuy nhiên nếu chuyển sang xử lý đồng thời kết quả chưa tới 1s. Thư viện ThreadPoolExecutor đã tận dụng 20 workers của cpu để cùng lúc xử lý song song 20 công việc đó. Output in ra màn hình thứ tự các bức ảnh tải về cũng ko còn theo tuần tự mà là bức ảnh nào được tải về trước sẽ được print ra đầu tiên.

Điều này tưởng như có vẻ là tuyệt vời tuy nhiên nó lại có 1 nhược điểm khá lớn. Khi sử dụng các thread để thực thi các công việc song song, nhưng các công việc đó lại được xử lý chỉ trong 1 core Nếu bạn cần nhiều hơn các công việc xử lý song song điều này có thể dẫn tới tình trạng cpu bị nóng và tê liệt. Rất may python đã cung cấp sẵn 1 thư viện khác thay thế cho nhược điểm đó

ProcessPoolExecutor. Thư viện này sẽ xử lý các công việc song song trên nhiều cpu. Thư viện này giải quyết được vấn đề của thread và nó vẫn ko phải là hoàn hảo. Trong python 1 process có thể chạy với nhiều thread, Để start 1 process nó cần nhiều hơn các tài nguyên xử lý. Bài sau mình sẽ mô tả chi tiết hơn về thư viện ProcessPoolExecutor này.