Moduł `concurrent` w Pythonie

W tym artykule wyjaśnimy moduł concurrent w Pythonie.

Wyjaśniając pojęcia współbieżności i równoległości, pokażemy, jak zaimplementować przetwarzanie asynchroniczne z użyciem modułu concurrent na praktycznych przykładach.

YouTube Video

Moduł concurrent w Pythonie

Przy przyspieszaniu przetwarzania w Pythonie ważne jest, aby mieć świadomość różnic między współbieżnością a równoległością. Moduł concurrent jest ważnym narzędziem do bezpiecznego i prostego obsługiwania przetwarzania asynchronicznego z uwzględnieniem tych różnic.

Różnica między współbieżnością a równoległością

• Współbieżność oznacza zaprojektowanie procesu tak, aby wiele zadań postępowało poprzez przełączanie się między nimi w małych jednostkach pracy. Nawet jeśli zadania nie są faktycznie wykonywane jednocześnie, wykorzystanie "czasu oczekiwania" pozwala usprawnić cały proces.

• Równoległość to mechanizm, który fizycznie wykonuje wiele zadań jednocześnie. Dzięki wykorzystaniu wielu rdzeni CPU przetwarzanie odbywa się równocześnie.

Obie są technikami przyspieszającymi przetwarzanie, jednak współbieżność jest kwestią projektową dotyczącą 'jak przebiegać', a równoległość kwestią wykonania - 'jak się uruchamia', przez co różnią się zasadniczo.

Czym jest moduł concurrent?

concurrent to biblioteka standardowa Pythona, która oferuje wysokopoziomowe API pozwalające bezpiecznie i prosto realizować współbieżność oraz równoległość. Została zaprojektowana tak, aby można było skupić się na 'wykonywaniu zadań' bez przejmowania się niskopoziomową obsługą wątków czy procesów.

Rola ThreadPoolExecutor i ProcessPoolExecutor

Moduł concurrent oferuje dwa główne rozwiązania, w zależności od charakteru zadania.

• ThreadPoolExecutor To rozwiązanie nadaje się do implementacji współbieżnych, szczególnie dla zadań z wieloma oczekiwaniami na operacje wejścia/wyjścia, takich jak komunikacja sieciowa czy operacje na plikach. Przełączając się między zadaniami, efektywnie wykorzystuje czas oczekiwania.

• ProcessPoolExecutor Ta implementacja przeznaczona jest do przetwarzania równoległego i nadaje się do zadań wymagających dużej mocy obliczeniowej. Wykorzystuje wiele procesów, aby w pełni użyć dostępnych rdzeni procesora równolegle.

Główną cechą modułu concurrent jest to, że udostępnia strukturę pozwalającą odpowiednio wybrać między współbieżnością a równoległością w zależności od potrzeb.

Podstawy ThreadPoolExecutor (dla zadań wejścia/wyjścia)

ThreadPoolExecutor nadaje się do zadań związanych z wejściem/wyjściem, takich jak komunikacja sieciowa czy operacje na plikach. Rozdziela zadania pomiędzy wiele wątków, skutecznie wykorzystując czas oczekiwania.

 1from concurrent.futures import ThreadPoolExecutor
 2import time
 3
 4def fetch_data(n):
 5    # Simulate an I/O-bound task
 6    time.sleep(1)
 7    return f"data-{n}"
 8
 9with ThreadPoolExecutor(max_workers=3) as executor:
10    futures = [executor.submit(fetch_data, i) for i in range(5)]
11
12    for future in futures:
13        print(future.result())

• W tym przykładzie wiele zadań wejścia/wyjścia, które oczekują przez sekundę, jest wykonywanych współbieżnie. Korzystając z submit, wywołania funkcji rejestrowane są jako zadania asynchroniczne, a wywołując result(), możesz poczekać na zakończenie i pobrać wynik, co umożliwia zwięzłą implementację przetwarzania współbieżnego z efektywnym wykorzystaniem czasu oczekiwania.

Prosta współbieżność z użyciem map

Jeśli nie jest potrzebne zaawansowane sterowanie, użycie map pozwoli uprościć Twój kod.

 1from concurrent.futures import ThreadPoolExecutor
 2import time
 3
 4def fetch_data(n):
 5    # Simulate an I/O-bound task
 6    time.sleep(1)
 7    return f"data-{n}"
 8
 9with ThreadPoolExecutor(max_workers=3) as executor:
10    results = executor.map(fetch_data, range(5))
11
12    for result in results:
13        print(result)

• W tym przykładzie wiele zadań wejścia/wyjścia jest wykonywanych współbieżnie z użyciem ThreadPoolExecutor.map. Ponieważ map zwraca wyniki w tej samej kolejności, co wejście, kod można napisać niemal jak do przetwarzania sekwencyjnego, a przetwarzanie współbieżne jest możliwe bez świadomości działania asynchronicznego — to duża zaleta.

Podstawy ProcessPoolExecutor (dla zadań CPU)

Przy obciążających CPU obliczeniach należy stosować procesy zamiast wątków. Dzięki temu można uniknąć ograniczenia Global Interpreter Lock (GIL).

 1from concurrent.futures import ProcessPoolExecutor
 2
 3def heavy_calculation(n):
 4    # Simulate a CPU-bound task
 5    total = 0
 6    for i in range(10_000_000):
 7        total += i * n
 8    return total
 9
10if __name__ == "__main__":
11    with ProcessPoolExecutor(max_workers=4) as executor:
12        results = executor.map(heavy_calculation, range(4))
13
14        for result in results:
15            print(result)

W tym przykładzie obciążające CPU obliczenia są wykonywane równolegle z użyciem ProcessPoolExecutor. Ponieważ wymaga to tworzenia procesów, potrzebny jest warunek __main__, który umożliwia bezpieczne przetwarzanie równoległe z wykorzystaniem wielu rdzeni CPU.

Przetwarzanie według kolejności ukończenia z as_completed

as_completed jest przydatne, gdy chcesz obsłużyć wyniki w kolejności ich ukończenia.

 1from concurrent.futures import ThreadPoolExecutor, as_completed
 2import time
 3
 4def fetch_data(n):
 5    # Simulate an I/O-bound task
 6    time.sleep(1)
 7    return f"data-{n}"
 8
 9with ThreadPoolExecutor(max_workers=3) as executor:
10    futures = [executor.submit(fetch_data, i) for i in range(5)]
11
12    for future in as_completed(futures):
13        print(future.result())

• W tym przykładzie wiele zadań asynchronicznych jest uruchamianych jednocześnie, a wyniki pobierane są w kolejności wykonywania. Stosowanie as_completed pozwala szybko obsługiwać wyniki niezależnie od kolejności zadań, co jest przydatne np. do pokazywania postępu czy konieczności sekwencyjnej obsługi.

Obsługa wyjątków

W concurrent wyjątki zgłaszane są podczas wywołania result().

 1from concurrent.futures import ThreadPoolExecutor
 2
 3def risky_task(n):
 4    # Simulate a task that may fail for a specific input
 5    if n == 3:
 6        raise ValueError("Something went wrong")
 7    return n * 2
 8
 9with ThreadPoolExecutor() as executor:
10    futures = [executor.submit(risky_task, i) for i in range(5)]
11
12    for future in futures:
13        try:
14            print(future.result())
15        except Exception as e:
16            print("Error:", e)

• Przykład ten pokazuje, że nawet jeśli niektóre zadania zgłoszą wyjątki, pozostałe kontynuują pracę i możesz obsłużyć wyjątki indywidualnie podczas pobierania wyników. Dzięki użyciu Future z concurrent istotne jest, że zarówno powodzenia, jak i niepowodzenia w przetwarzaniu asynchronicznym można bezpiecznie obsłużyć.

Wskazówki dotyczące wyboru między wątkami a procesami

Aby efektywnie stosować współbieżność i równoległość, istotny jest właściwy wybór metody zależnie od charakteru zadania.

W praktyce poniższe kryteria mogą pomóc w podjęciu decyzji.

• W przypadku procesów z dużą liczbą operacji oczekiwania na wejście/wyjście, takich jak komunikacja lub operacje na plikach, użyj ThreadPoolExecutor.
• Do zadań obciążających CPU, wymagających intensywnych obliczeń, użyj ProcessPoolExecutor.
• Jeśli masz wiele prostych zadań, użycie map pozwala napisać bardziej zwięzły kod.
• Jeśli istotne jest precyzyjne sterowanie kolejnością realizacji lub obsługa wyjątków, połącz submit z as_completed.

Zalety korzystania z concurrent

Dzięki użyciu modułu concurrent, można obsługiwać przetwarzanie asynchroniczne bezpiecznie i intuicyjnie.

Główne zalety to:.

• Nie musisz zajmować się niskopoziomową obsługą wątków i procesów.
• Jest dostępny jako część biblioteki standardowej Pythona, więc możesz korzystać z niej z pełnym zaufaniem.
• Kod staje się bardziej czytelny i łatwy w utrzymaniu.
• To idealny pierwszy krok do nauki współbieżności i równoległości.

Samo pamiętanie o tych wskazówkach znacznie zmniejsza liczbę błędów przy użyciu concurrent.

Podsumowanie

Moduł concurrent to standardowa opcja do praktycznej współbieżności i równoległości w Pythonie. Pozwala zwiększyć wydajność bez dużych zmian w treści przetwarzania, co jest dużą zaletą w praktyce. Dzięki concurrent możesz zwięźle zaimplementować przetwarzanie asynchroniczne, bezpiecznie zarządzając obsługą wyjątków i kontrolą wykonania.

Możesz śledzić ten artykuł, korzystając z Visual Studio Code na naszym kanale YouTube. Proszę również sprawdzić nasz kanał YouTube.

YouTube Video