Python có hỗ trợ xử lý song song không?

Phiên bản đồng bộ

Chúng tôi sẽ bắt đầu với phiên bản không đồng thời của tác vụ này. Lưu ý rằng chương trình này yêu cầu mô-đun

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Như bạn có thể thấy, đây là một chương trình khá ngắn.

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Có thể chỉ cần sử dụng trực tiếp

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Sơ đồ xử lý cho chương trình này sẽ giống như sơ đồ giới hạn I/O trong phần trước

Ghi chú. Lưu lượng mạng phụ thuộc vào nhiều yếu tố có thể thay đổi từ giây này sang giây khác. Tôi đã thấy thời gian của các bài kiểm tra này tăng gấp đôi từ lần chạy này sang lần chạy khác do sự cố mạng

Tại sao phiên bản đồng bộ lại thành công

Điều tuyệt vời về phiên bản mã này là, thật dễ dàng. Nó tương đối dễ viết và gỡ lỗi. Nó cũng đơn giản hơn để suy nghĩ về. Chỉ có một dòng suy nghĩ chạy qua nó, vì vậy bạn có thể dự đoán bước tiếp theo là gì và nó sẽ hoạt động như thế nào

Các vấn đề với phiên bản đồng bộ

Vấn đề lớn ở đây là nó tương đối chậm so với các giải pháp khác mà chúng tôi sẽ cung cấp. Đây là một ví dụ về kết quả cuối cùng trên máy của tôi

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Ghi chú. Kết quả của bạn có thể thay đổi đáng kể. Khi chạy tập lệnh này, tôi thấy thời gian thay đổi từ 14. 2 đến 21. 9 giây. Đối với bài viết này, tôi đã chạy nhanh nhất trong ba lần chạy. Sự khác biệt giữa các phương pháp vẫn sẽ rõ ràng

Tuy nhiên, chậm hơn không phải lúc nào cũng là một vấn đề lớn. Nếu chương trình bạn đang chạy chỉ mất 2 giây với phiên bản đồng bộ và hiếm khi chạy, thì có lẽ không đáng để thêm đồng thời. Bạn có thể dừng lại ở đây

Nếu chương trình của bạn được chạy thường xuyên thì sao?

Phiên bản $ ./io_non_concurrent.py [most output skipped] Downloaded 160 in 14.289619207382202 seconds 7

Như bạn có thể đoán, viết một chương trình theo luồng cần nhiều nỗ lực hơn. Tuy nhiên, bạn có thể ngạc nhiên khi thấy cần thêm ít nỗ lực như thế nào đối với các trường hợp đơn giản. Đây là chương trình tương tự trông như thế nào với

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

Khi bạn thêm

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Trong phiên bản này, bạn đang tạo một

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Bạn đã biết về phần

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Thật hữu ích, thư viện tiêu chuẩn triển khai

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Khi bạn có một

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Những người trong số các bạn đến từ các ngôn ngữ khác, hoặc thậm chí Python 2, có lẽ đang tự hỏi đâu là các đối tượng và hàm thông thường quản lý các chi tiết mà bạn đã quen khi xử lý

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tất cả những thứ này vẫn còn ở đó và bạn có thể sử dụng chúng để đạt được sự kiểm soát chi tiết về cách chạy các chuỗi của bạn. Nhưng, bắt đầu với Python 3. 2, thư viện tiêu chuẩn đã thêm một trừu tượng cấp cao hơn có tên là

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Một thay đổi thú vị khác trong ví dụ của chúng ta là mỗi luồng cần tạo đối tượng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đây là một trong những vấn đề thú vị và khó khăn với

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Có một số chiến lược để làm cho truy cập dữ liệu an toàn theo luồng tùy thuộc vào dữ liệu là gì và cách bạn sử dụng dữ liệu đó. Một trong số đó là sử dụng các cấu trúc dữ liệu an toàn theo luồng như

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

Các đối tượng này sử dụng các nguyên hàm cấp thấp như

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Một chiến lược khác để sử dụng ở đây là thứ gọi là lưu trữ cục bộ luồng.

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Khi

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

Cuối cùng, một lưu ý nhanh về việc chọn số lượng chủ đề. Bạn có thể thấy rằng mã ví dụ sử dụng 5 chủ đề. Vui lòng thử với con số này và xem thời gian tổng thể thay đổi như thế nào. Bạn có thể mong đợi rằng có một luồng cho mỗi lần tải xuống sẽ là nhanh nhất, nhưng ít nhất trên hệ thống của tôi thì không phải vậy. Tôi đã tìm thấy kết quả nhanh nhất ở đâu đó trong khoảng từ 5 đến 10 chủ đề. Nếu bạn tăng cao hơn mức đó, thì chi phí bổ sung cho việc tạo và hủy chuỗi sẽ xóa bất kỳ khoản tiết kiệm thời gian nào

Câu trả lời khó ở đây là số luồng chính xác không phải là hằng số từ nhiệm vụ này sang nhiệm vụ khác. Một số thử nghiệm là bắt buộc

Tại sao phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

nó nhanh. Đây là lần chạy thử nghiệm nhanh nhất của tôi. Hãy nhớ rằng phiên bản không đồng thời mất hơn 14 giây

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đây là sơ đồ thời gian thực hiện của nó trông như thế nào

Nó sử dụng nhiều luồng để có nhiều yêu cầu mở đến các trang web cùng một lúc, cho phép chương trình của bạn chồng chéo thời gian chờ đợi và nhận được kết quả cuối cùng nhanh hơn. dippee. đó là mục tiêu

Các vấn đề với Phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Chà, như bạn có thể thấy từ ví dụ, cần thêm một chút mã để thực hiện điều này và bạn thực sự phải suy nghĩ về dữ liệu nào được chia sẻ giữa các chuỗi

Các luồng có thể tương tác theo những cách tinh vi và khó phát hiện. Những tương tác này có thể gây ra các điều kiện chạy đua thường dẫn đến các lỗi ngẫu nhiên, không liên tục và có thể khá khó tìm. Những bạn chưa quen với khái niệm điều kiện cuộc đua có thể muốn mở rộng và đọc phần bên dưới

Điều kiện cuộc đuaHiển thị/Ẩn

Điều kiện chủng tộc là toàn bộ các loại lỗi tinh vi có thể và thường xuyên xảy ra trong mã đa luồng. Điều kiện cạnh tranh xảy ra do lập trình viên không có quyền truy cập dữ liệu được bảo vệ đầy đủ để ngăn chặn các luồng can thiệp lẫn nhau. Bạn cần thực hiện thêm các bước khi viết mã theo luồng để đảm bảo mọi thứ an toàn theo luồng

Điều đang xảy ra ở đây là hệ điều hành đang kiểm soát khi nào luồng của bạn chạy và khi nào nó được hoán đổi để cho một luồng khác chạy. Việc hoán đổi chuỗi này có thể xảy ra tại bất kỳ thời điểm nào, ngay cả khi thực hiện các bước phụ của câu lệnh Python. Như một ví dụ nhanh, nhìn vào chức năng này

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đoạn mã này khá giống với cấu trúc bạn đã sử dụng trong ví dụ về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Để tăng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Bởi vì hệ điều hành không biết gì về mã của bạn và có thể hoán đổi các luồng tại bất kỳ thời điểm nào trong quá trình thực thi, nên việc hoán đổi này có thể xảy ra sau khi một luồng đã đọc giá trị nhưng trước khi nó có cơ hội ghi lại giá trị đó. Nếu mã mới đang chạy cũng sửa đổi

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Như bạn có thể tưởng tượng, gặp phải tình huống chính xác này là khá hiếm. Bạn có thể chạy chương trình này hàng ngàn lần và không bao giờ gặp sự cố. Đó là điều làm cho loại sự cố này khá khó gỡ lỗi vì nó có thể khá khó tái tạo và có thể khiến các lỗi trông ngẫu nhiên xuất hiện

Ví dụ khác, tôi muốn nhắc bạn rằng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản $ ./io_non_concurrent.py [most output skipped] Downloaded 160 in 14.289619207382202 seconds 6

Trước khi bạn xem xét mã ví dụ của

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đây sẽ là phiên bản đơn giản hóa của

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Khái niệm chung về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Trạng thái sẵn sàng sẽ chỉ ra rằng một tác vụ có việc phải làm và sẵn sàng để chạy và trạng thái chờ có nghĩa là tác vụ đang đợi một số thứ bên ngoài kết thúc, chẳng hạn như hoạt động mạng

Vòng lặp sự kiện đơn giản hóa của bạn duy trì hai danh sách nhiệm vụ, một danh sách cho mỗi trạng thái này. Nó chọn một trong các tác vụ đã sẵn sàng và khởi động lại để chạy. Nhiệm vụ đó được kiểm soát hoàn toàn cho đến khi nó hợp tác trao lại quyền kiểm soát cho vòng lặp sự kiện

Khi tác vụ đang chạy trao quyền điều khiển trở lại vòng lặp sự kiện, vòng lặp sự kiện sẽ đặt tác vụ đó vào danh sách sẵn sàng hoặc danh sách chờ, sau đó duyệt qua từng tác vụ trong danh sách chờ để xem liệu tác vụ đó đã sẵn sàng cho một thao tác I/O chưa . Nó biết rằng các tác vụ trong danh sách sẵn sàng vẫn sẵn sàng vì nó biết chúng chưa chạy

Sau khi tất cả các tác vụ đã được sắp xếp lại vào đúng danh sách, vòng lặp sự kiện sẽ chọn tác vụ tiếp theo để chạy và quy trình lặp lại. Vòng lặp sự kiện đơn giản hóa của bạn chọn nhiệm vụ đã đợi lâu nhất và chạy nhiệm vụ đó. Quá trình này lặp lại cho đến khi vòng lặp sự kiện kết thúc

Một điểm quan trọng của

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đó là một cái nhìn cấp cao về những gì đang xảy ra với

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Bây giờ hãy nói về hai từ khóa mới đã được thêm vào Python.

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Dễ dàng nhất để nghĩ về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Một ngoại lệ cho điều này mà bạn sẽ thấy trong đoạn mã tiếp theo là câu lệnh

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Như tôi chắc rằng bạn có thể tưởng tượng, có một số phức tạp trong việc quản lý sự tương tác giữa vòng lặp sự kiện và các tác vụ. Đối với các nhà phát triển bắt đầu với

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Quay lại mã

Bây giờ bạn đã có hiểu biết cơ bản về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản này phức tạp hơn một chút so với hai phiên bản trước. Nó có cấu trúc tương tự, nhưng có một chút công việc thiết lập các tác vụ hơn là tạo

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Bạn có thể chia sẻ phiên trên tất cả các nhiệm vụ, vì vậy, phiên được tạo ở đây dưới dạng trình quản lý bối cảnh. Các tác vụ có thể chia sẻ phiên vì tất cả chúng đều đang chạy trên cùng một luồng. Không có cách nào một nhiệm vụ có thể làm gián đoạn nhiệm vụ khác trong khi phiên ở trạng thái xấu

Bên trong trình quản lý ngữ cảnh đó, nó tạo một danh sách các tác vụ bằng cách sử dụng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Mã

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tuy nhiên, có một thay đổi nhỏ nhưng quan trọng nằm trong các chi tiết ở đây. Hãy nhớ làm thế nào chúng ta nói về số lượng chủ đề để tạo ra?

Một trong những ưu điểm thú vị của

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Cuối cùng, bản chất của

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Nếu bạn đã cập nhật lên Python 3. 7, các nhà phát triển lõi Python đã đơn giản hóa cú pháp này cho bạn. Thay vì

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tại sao phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Nó thực sự nhanh. Trong các thử nghiệm trên máy của tôi, đây là phiên bản mã nhanh nhất với biên độ tốt

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

Biểu đồ thời gian thực hiện trông khá giống với những gì đang xảy ra trong ví dụ về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Việc thiếu một trình bao bọc đẹp mắt như

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Ngoài ra, có một lập luận phổ biến rằng việc phải thêm

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Vấn đề mở rộng quy mô cũng xuất hiện lớn ở đây. Chạy ví dụ

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Các vấn đề với phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Có một số vấn đề với

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Một vấn đề khác, tế nhị hơn, là tất cả các lợi thế của đa nhiệm hợp tác sẽ bị loại bỏ nếu một trong các nhiệm vụ không hợp tác. Một lỗi nhỏ trong mã có thể khiến một tác vụ bị tắt và giữ bộ xử lý trong một thời gian dài, làm chết các tác vụ khác cần chạy. Không có cách nào để vòng lặp sự kiện đột nhập nếu một tác vụ không trao lại quyền kiểm soát cho nó

Với ý nghĩ đó, chúng ta hãy tiến tới một cách tiếp cận hoàn toàn khác với đồng thời,

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản $ ./io_non_concurrent.py [most output skipped] Downloaded 160 in 14.289619207382202 seconds 9

Không giống như các phương pháp trước đây, phiên bản mã

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Điều này ngắn hơn nhiều so với ví dụ

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tóm tắt về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Cho đến thời điểm này, tất cả các ví dụ về đồng thời trong bài viết này chỉ chạy trên một CPU hoặc lõi trong máy tính của bạn. Lý do cho điều này liên quan đến thiết kế hiện tại của CPython và một thứ gọi là Khóa thông dịch viên toàn cầu hoặc GIL

Bài viết này sẽ không đi sâu vào cách thức và lý do của GIL. Bây giờ đủ để biết rằng các phiên bản đồng bộ,

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Như bạn có thể tưởng tượng, việc khởi chạy một trình thông dịch Python riêng không nhanh bằng việc bắt đầu một luồng mới trong trình thông dịch Python hiện tại. Đây là một hoạt động nặng nhọc và đi kèm với một số hạn chế và khó khăn, nhưng đối với vấn đề chính xác, nó có thể tạo ra sự khác biệt lớn

Mã số

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Mã này có một số thay đổi nhỏ so với phiên bản đồng bộ của chúng tôi. Cái đầu tiên là trong

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Điều xảy ra ở đây là

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Dòng tạo ra

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đối với vấn đề này, việc tăng số lượng quy trình không giúp mọi thứ nhanh hơn. Nó thực sự làm mọi thứ chậm lại vì chi phí thiết lập và phá bỏ tất cả các quy trình đó lớn hơn lợi ích của việc thực hiện các yêu cầu I/O song song

Tiếp theo chúng ta có phần

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tham số chức năng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đó thực sự là tất cả để có nó. Phần còn lại của mã khá giống với những gì bạn đã thấy trước đây

Tại sao phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Các vấn đề với phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản ví dụ này yêu cầu một số thiết lập bổ sung và đối tượng toàn cầu

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Cuối cùng, nó rõ ràng là chậm hơn phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Điều đó không có gì đáng ngạc nhiên, vì các vấn đề liên quan đến I/O không thực sự là lý do tại sao

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản đồng bộ giới hạn CPU

Bây giờ hãy xem phiên bản không đồng thời của ví dụ

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Mã này gọi

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Không giống như các ví dụ giới hạn I/O, các ví dụ giới hạn CPU thường khá nhất quán trong thời gian chạy của chúng. Cái này mất khoảng 7. 8 giây trên máy của tôi

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Rõ ràng chúng ta có thể làm tốt hơn thế này. Đây là tất cả chạy trên một CPU không có đồng thời. Hãy xem những gì chúng ta có thể làm để làm cho nó tốt hơn

Phiên bản $ ./io_non_concurrent.py [most output skipped] Downloaded 160 in 14.289619207382202 seconds 7 và $ ./io_non_concurrent.py [most output skipped] Downloaded 160 in 14.289619207382202 seconds 6

Bạn nghĩ việc viết lại mã này bằng cách sử dụng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Nếu bạn trả lời “Hoàn toàn không”, hãy cho mình một chiếc bánh quy. Nếu bạn trả lời, "Nó sẽ làm nó chậm lại", hãy cho mình hai chiếc bánh quy

Đây là lý do tại sao. Trong ví dụ về giới hạn I/O của bạn ở trên, phần lớn thời gian tổng thể được sử dụng để chờ các thao tác chậm kết thúc.

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tuy nhiên, đối với sự cố liên quan đến CPU, không có sự chờ đợi. CPU đang quay nhanh nhất có thể để giải quyết vấn đề. Trong Python, cả luồng và tác vụ đều chạy trên cùng một CPU trong cùng một quy trình. Điều đó có nghĩa là một CPU đang thực hiện tất cả công việc của mã không đồng thời cộng với công việc bổ sung là thiết lập các luồng hoặc tác vụ. Phải mất hơn 10 giây

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Tôi đã viết phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản $ ./io_non_concurrent.py [most output skipped] Downloaded 160 in 14.289619207382202 seconds 9 giới hạn CPU

Bây giờ bạn cuối cùng đã đạt đến nơi mà

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đây là mã trông như thế nào

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Một chút mã này phải thay đổi từ phiên bản không đồng thời. Bạn phải

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Đây chỉ là những gì bạn đã làm cho mã I/O-bound

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Như đã đề cập ở trên, tham số tùy chọn

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Ngoài ra, như chúng tôi đã đề cập trong phần đầu tiên về

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

import concurrent.futures
import requests
import threading
import time


thread_local = threading.local()


def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session


def download_site(url):
    session = get_session()
    with session.get(url) as response:
        print(f"Read {len(response.content)} from {url}")


def download_all_sites(sites):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        executor.map(download_site, sites)


if __name__ == "__main__":
    sites = [
        "https://www.jython.org",
        "http://olympus.realpython.org/dice",
    ] * 80
    start_time = time.time()
    download_all_sites(sites)
    duration = time.time() - start_time
    print(f"Downloaded {len(sites)} in {duration} seconds")

Tại sao phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Này, đó chính xác là những gì tôi đã nói vào lần cuối cùng chúng ta xem xét

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Điều đó tốt hơn nhiều so với những gì chúng ta đã thấy với các tùy chọn khác

Các vấn đề với phiên bản

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Có một số nhược điểm khi sử dụng

$ ./io_non_concurrent.py
   [most output skipped]
Downloaded 160 in 14.289619207382202 seconds

Ngoài ra, nhiều giải pháp yêu cầu giao tiếp nhiều hơn giữa các quy trình. Điều này có thể thêm một số phức tạp vào giải pháp của bạn mà một chương trình không đồng thời sẽ không cần phải xử lý