Einführung in die parallele und gleichzeitige Programmierung in Python

() translation by (you can also view the original English article)

Python ist eine der beliebtesten Sprachen für Datenverarbeitung und Datenwissenschaft im Allgemeinen. Das Ökosystem bietet viele Bibliotheken und Frameworks, die Hochleistungsrechnen ermöglichen. Das parallele Programmieren in Python kann sich jedoch als recht schwierig erweisen.

In diesem Tutorial werden wir untersuchen, warum Parallelität besonders im Python-Kontext schwierig ist, und dafür werden wir Folgendes durchgehen:

Warum ist Parallelität in Python schwierig? (Hinweis: Dies liegt an der GIL - der globalen Interpretersperre).
Threads vs. Prozesse: Verschiedene Wege zur Erreichung von Parallelität. Wann übereinander verwenden?
Parallel vs. Concurrent: Warum können wir uns in einigen Fällen eher mit Parallelität als mit Parallelität zufrieden geben?
Erstellen eines einfachen, aber praktischen Beispiels unter Verwendung der verschiedenen diskutierten Techniken.

Global Interpreter Lock

Das Global Interpreter Lock (GIL) ist eines der umstrittensten Themen in der Python-Welt. In CPython, der beliebtesten Implementierung von Python, ist die GIL ein Mutex, der die Thread-Sicherheit gewährleistet. Die GIL erleichtert die Integration in externe Bibliotheken, die nicht threadsicher sind, und beschleunigt nicht parallelen Code. Dies ist jedoch mit Kosten verbunden. Aufgrund der GIL können wir durch Multithreading keine echte Parallelität erreichen. Grundsätzlich können zwei verschiedene native Threads desselben Prozesses Python-Code nicht gleichzeitig ausführen.

Die Dinge sind jedoch nicht so schlimm, und hier ist der Grund: Dinge, die außerhalb des GIL-Bereichs passieren, können parallel sein. In diese Kategorie fallen lang laufende Aufgaben wie E/A und glücklicherweise Bibliotheken wie numpy.

Threads vs. Prozesse

Python ist also nicht wirklich multithreaded. Aber was ist ein Thread? Machen wir einen Schritt zurück und betrachten die Dinge in der Perspektive.

Ein Prozess ist eine grundlegende Betriebssystemabstraktion. Es ist ein Programm, das ausgeführt wird - mit anderen Worten, Code, der ausgeführt wird. Auf einem Computer werden immer mehrere Prozesse ausgeführt, die parallel ausgeführt werden.

Ein Prozess kann mehrere Threads haben. Sie führen denselben Code aus, der zum übergeordneten Prozess gehört. Im Idealfall laufen sie parallel, aber nicht unbedingt. Der Grund, warum Prozesse nicht ausreichen, liegt darin, dass Anwendungen reagieren und auf Benutzeraktionen warten müssen, während sie die Anzeige aktualisieren und eine Datei speichern.

Wenn das noch etwas unklar ist, hier ein Cheatsheet:

PROZESSE	GEWINDE
Prozesse teilen keinen Speicher	Threads teilen sich den Speicher
Laich-/Wechselprozesse sind teuer	Das Laichen/Wechseln von Threads ist kostengünstiger
Prozesse erfordern mehr Ressourcen	Threads erfordern weniger Ressourcen (werden manchmal als Lightweight-Prozesse bezeichnet).
Keine Speichersynchronisation erforderlich	Sie müssen Synchronisationsmechanismen verwenden, um sicherzustellen, dass Sie die Daten korrekt verarbeiten

Es gibt nicht ein Rezept, das alles bietet. Die Auswahl hängt stark vom Kontext und der Aufgabe ab, die Sie erreichen möchten.

Parallel vs. Gleichzeitig

Jetzt gehen wir noch einen Schritt weiter und tauchen in die Parallelität ein. Parallelität wird oft missverstanden und mit Parallelität verwechselt. Das ist nicht der Fall. Parallelität bedeutet, dass unabhängiger Code so geplant wird, dass er kooperativ ausgeführt wird. Nutzen Sie die Tatsache, dass ein Teil des Codes auf E/A-Vorgänge wartet, und führen Sie während dieser Zeit einen anderen, aber unabhängigen Teil des Codes aus.

In Python können wir über Greenlets ein leichtes gleichzeitiges Verhalten erzielen. Aus Sicht der Parallelisierung ist die Verwendung von Threads oder Greenlets gleichwertig, da keiner von beiden parallel ausgeführt wird. Die Herstellung von Greenlets ist noch günstiger als die von Threads. Aus diesem Grund werden Greenlets häufig für die Ausführung einer Vielzahl einfacher E/A-Aufgaben verwendet, wie sie normalerweise in Netzwerken und Webservern zu finden sind.

Nachdem wir nun den Unterschied zwischen parallelen und gleichzeitigen Threads und Prozessen kennen, können wir veranschaulichen, wie unterschiedliche Aufgaben für die beiden Paradigmen ausgeführt werden. Folgendes werden wir tun: Wir werden mehrmals eine Aufgabe außerhalb der GIL und eine innerhalb der GIL ausführen. Wir führen sie seriell aus, verwenden Threads und Prozesse. Definieren wir die Aufgaben:

import os
import time
import threading
import multiprocessing

NUM_WORKERS = 4

def only_sleep():
    """ Do nothing, wait for a timer to expire """
    print("PID: %s, Process Name: %s, Thread Name: %s" % (
        os.getpid(),
        multiprocessing.current_process().name,
        threading.current_thread().name)
    )
    time.sleep(1)


def crunch_numbers():
    """ Do some computations """
    print("PID: %s, Process Name: %s, Thread Name: %s" % (
        os.getpid(),
        multiprocessing.current_process().name,
        threading.current_thread().name)
    )
    x = 0
    while x < 10000000:
        x += 1

Wir haben zwei Aufgaben erstellt. Beide haben eine lange Laufzeit, aber nur crunch_numbers führt aktiv Berechnungen durch. Lassen Sie uns only_sleep seriell, multithreaded und mit mehreren Prozessen ausführen und die Ergebnisse vergleichen:

## Run tasks serially
start_time = time.time()
for _ in range(NUM_WORKERS):
    only_sleep()
end_time = time.time()

print("Serial time=", end_time - start_time)

# Run tasks using threads
start_time = time.time()
threads = [threading.Thread(target=only_sleep) for _ in range(NUM_WORKERS)]
[thread.start() for thread in threads]
[thread.join() for thread in threads]
end_time = time.time()

print("Threads time=", end_time - start_time)

# Run tasks using processes
start_time = time.time()
processes = [multiprocessing.Process(target=only_sleep()) for _ in range(NUM_WORKERS)]
[process.start() for process in processes]
[process.join() for process in processes]
end_time = time.time()

print("Parallel time=", end_time - start_time)

Hier ist die Ausgabe, die ich habe (Ihre sollte ähnlich sein, obwohl PIDs und Zeiten etwas variieren):

PID: 95726, Process Name: MainProcess, Thread Name: MainThread
PID: 95726, Process Name: MainProcess, Thread Name: MainThread
PID: 95726, Process Name: MainProcess, Thread Name: MainThread
PID: 95726, Process Name: MainProcess, Thread Name: MainThread
Serial time= 4.018089056015015

PID: 95726, Process Name: MainProcess, Thread Name: Thread-1
PID: 95726, Process Name: MainProcess, Thread Name: Thread-2
PID: 95726, Process Name: MainProcess, Thread Name: Thread-3
PID: 95726, Process Name: MainProcess, Thread Name: Thread-4
Threads time= 1.0047411918640137

PID: 95728, Process Name: Process-1, Thread Name: MainThread
PID: 95729, Process Name: Process-2, Thread Name: MainThread
PID: 95730, Process Name: Process-3, Thread Name: MainThread
PID: 95731, Process Name: Process-4, Thread Name: MainThread
Parallel time= 1.014023780822754

Hier einige Beobachtungen:

Im Fall des seriellen Ansatzes sind die Dinge ziemlich offensichtlich. Wir führen die Aufgaben nacheinander aus. Alle vier Läufe werden von demselben Thread desselben Prozesses ausgeführt.
Mithilfe von Prozessen reduzieren wir die Ausführungszeit auf ein Viertel der ursprünglichen Zeit, einfach weil die Aufgaben parallel ausgeführt werden. Beachten Sie, wie jede Aufgabe in einem anderen Prozess und auf dem MainThread dieses Prozesses ausgeführt wird.
Mit Threads nutzen wir die Tatsache, dass die Aufgaben gleichzeitig ausgeführt werden können. Die Ausführungszeit wird ebenfalls auf ein Viertel reduziert, obwohl nichts parallel läuft. So geht's: Wir erzeugen den ersten Thread und er wartet darauf, dass der Timer abläuft. Wir unterbrechen die Ausführung und lassen sie warten, bis der Timer abgelaufen ist. In dieser Zeit erzeugen wir den zweiten Thread. Wir wiederholen dies für alle Threads. In einem Moment läuft der Timer des ersten Threads ab, sodass wir die Ausführung darauf umschalten und ihn beenden. Der Algorithmus wird für den zweiten und für alle anderen Threads wiederholt. Am Ende ist das Ergebnis, als ob die Dinge parallel laufen würden. Sie werden auch feststellen, dass die vier verschiedenen Threads von demselben Prozess abzweigen und darin leben: MainProcess.
Möglicherweise stellen Sie sogar fest, dass der Thread-Ansatz schneller ist als der wirklich parallele. Das liegt am Overhead der Laichprozesse. Wie bereits erwähnt, ist das Laichen und Umschalten ein teurer Vorgang.

Lassen Sie uns die gleiche Routine ausführen, aber diesmal die Aufgabe crunch_numbers ausführen:

start_time = time.time()
for _ in range(NUM_WORKERS):
    crunch_numbers()
end_time = time.time()

print("Serial time=", end_time - start_time)

start_time = time.time()
threads = [threading.Thread(target=crunch_numbers) for _ in range(NUM_WORKERS)]
[thread.start() for thread in threads]
[thread.join() for thread in threads]
end_time = time.time()

print("Threads time=", end_time - start_time)


start_time = time.time()
processes = [multiprocessing.Process(target=crunch_numbers) for _ in range(NUM_WORKERS)]
[process.start() for process in processes]
[process.join() for process in processes]
end_time = time.time()

print("Parallel time=", end_time - start_time)

Hier ist die Ausgabe, die ich habe:

PID: 96285, Process Name: MainProcess, Thread Name: MainThread
PID: 96285, Process Name: MainProcess, Thread Name: MainThread
PID: 96285, Process Name: MainProcess, Thread Name: MainThread
PID: 96285, Process Name: MainProcess, Thread Name: MainThread
Serial time= 2.705625057220459
PID: 96285, Process Name: MainProcess, Thread Name: Thread-1
PID: 96285, Process Name: MainProcess, Thread Name: Thread-2
PID: 96285, Process Name: MainProcess, Thread Name: Thread-3
PID: 96285, Process Name: MainProcess, Thread Name: Thread-4
Threads time= 2.6961309909820557
PID: 96289, Process Name: Process-1, Thread Name: MainThread
PID: 96290, Process Name: Process-2, Thread Name: MainThread
PID: 96291, Process Name: Process-3, Thread Name: MainThread
PID: 96292, Process Name: Process-4, Thread Name: MainThread
Parallel time= 0.8014059066772461

Der Hauptunterschied liegt hier im Ergebnis des Multithread-Ansatzes. Dieses Mal funktioniert es sehr ähnlich wie der serielle Ansatz, und hier ist der Grund: Da es Berechnungen durchführt und Python keine echte Parallelität ausführt, werden die Threads im Grunde nacheinander ausgeführt und führen zu einer Ausführung, bis sie alle fertig sind.

Das Python Parallel/Concurrent Programming Ecosystem

Python verfügt über umfangreiche APIs für die parallele/gleichzeitige Programmierung. In diesem Tutorial behandeln wir die beliebtesten, aber Sie müssen wissen, dass es für jeden Bedarf in diesem Bereich wahrscheinlich bereits etwas gibt, das Ihnen helfen kann, Ihr Ziel zu erreichen.

Im nächsten Abschnitt erstellen wir eine praktische Anwendung in vielen Formen unter Verwendung aller vorgestellten Bibliotheken. Hier sind ohne weiteres die Module/Bibliotheken, die wir behandeln werden:

threading: Die Standardmethode zum Arbeiten mit Threads in Python. Es handelt sich um einen übergeordneten API-Wrapper über die vom _thread-Modul bereitgestellten Funktionen, bei dem es sich um eine übergeordnete Schnittstelle zur Thread-Implementierung des Betriebssystems handelt.
concurrent.futures: Ein Modulteil der Standardbibliothek, der eine noch übergeordnete Abstraktionsschicht über Threads bereitstellt. Die Threads werden als asynchrone Aufgaben modelliert.
multiprocessing: Ähnlich wie das threading-Modul, bietet eine sehr ähnliche Schnittstelle, verwendet jedoch Prozesse anstelle von Threads.
gevent und greenlets: Greenlets, auch als Micro-Threads bezeichnet, sind Ausführungseinheiten, die gemeinsam geplant werden können und gleichzeitig Aufgaben ohne großen Aufwand ausführen können.
celery: Eine übergeordnete Warteschlange für verteilte Aufgaben. Die Aufgaben werden gleichzeitig in die Warteschlange gestellt und unter Verwendung verschiedener Paradigmen wie multiprocessing oder gevent ausgeführt.

Erstellen einer praktischen Anwendung

Die Theorie zu kennen ist schön und gut, aber der beste Weg zu lernen ist, etwas Praktisches zu bauen, oder? In diesem Abschnitt werden wir eine klassische Art von Anwendung erstellen, die alle verschiedenen Paradigmen durchläuft.

Lassen Sie uns eine Anwendung erstellen, die die Verfügbarkeit von Websites überprüft. Es gibt viele solcher Lösungen, die bekanntesten sind wahrscheinlich Jetpack Monitor und Uptime Robot. Der Zweck dieser Apps besteht darin, Sie zu benachrichtigen, wenn Ihre Website nicht verfügbar ist, damit Sie schnell Maßnahmen ergreifen können. So funktionieren sie:

Die Anwendung durchsucht sehr häufig eine Liste von Website-URLs und prüft, ob diese Websites aktiv sind.
Jede Website sollte alle 5-10 Minuten überprüft werden, damit die Ausfallzeit nicht wesentlich ist.
Anstatt eine klassische HTTP-GET-Anforderung auszuführen, wird eine HEAD-Anforderung ausgeführt, sodass Ihr Datenverkehr nicht wesentlich beeinträchtigt wird.
Wenn der HTTP-Status in den Gefahrenbereichen (400+, 500+) liegt, wird der Eigentümer benachrichtigt.
Der Eigentümer wird entweder per E-Mail, SMS oder Push-Benachrichtigung benachrichtigt.

Deshalb ist es wichtig, das Problem parallel/gleichzeitig anzugehen. Wenn die Liste der Websites wächst, garantiert uns das serielle Durchgehen der Liste nicht, dass jede Website etwa alle fünf Minuten überprüft wird. Die Websites können stundenlang nicht verfügbar sein und der Eigentümer wird nicht benachrichtigt.

Beginnen wir mit dem Schreiben einiger Dienstprogramme:

# utils.py

import time
import logging
import requests


class WebsiteDownException(Exception):
    pass


def ping_website(address, timeout=20):
    """
    Check if a website is down. A website is considered down 
    if either the status_code >= 400 or if the timeout expires
    
    Throw a WebsiteDownException if any of the website down conditions are met
    """
    try:
        response = requests.head(address, timeout=timeout)
        if response.status_code >= 400:
            logging.warning("Website %s returned status_code=%s" % (address, response.status_code))
            raise WebsiteDownException()
    except requests.exceptions.RequestException:
        logging.warning("Timeout expired for website %s" % address)
        raise WebsiteDownException()
        

def notify_owner(address):
    """ 
    Send the owner of the address a notification that their website is down 
    
    For now, we're just going to sleep for 0.5 seconds but this is where 
    you would send an email, push notification or text-message
    """
    logging.info("Notifying the owner of %s website" % address)
    time.sleep(0.5)
    

def check_website(address):
    """
    Utility function: check if a website is down, if so, notify the user
    """
    try:
        ping_website(address)
    except WebsiteDownException:
        notify_owner(address)

Wir benötigen tatsächlich eine Website-Liste, um unser System auszuprobieren. Erstelle deine eigene Liste oder benutze meine:

# websites.py

WEBSITE_LIST = [
    'https://envato.com',
    'http://amazon.co.uk',
    'http://amazon.com',
    'http://facebook.com',
    'http://google.com',
    'http://google.fr',
    'http://google.es',
    'http://google.co.uk',
    'http://internet.org',
    'http://gmail.com',
    'http://stackoverflow.com',
    'http://github.com',
    'http://heroku.com',
    'http://really-cool-available-domain.com',
    'http://djangoproject.com',
    'http://rubyonrails.org',
    'http://basecamp.com',
    'http://trello.com',
    'http://yiiframework.com',
    'http://shopify.com',
    'http://another-really-interesting-domain.co',
    'http://airbnb.com',
    'http://instagram.com',
    'http://snapchat.com',
    'http://youtube.com',
    'http://baidu.com',
    'http://yahoo.com',
    'http://live.com',
    'http://linkedin.com',
    'http://yandex.ru',
    'http://netflix.com',
    'http://wordpress.com',
    'http://bing.com',
]

Normalerweise speichern Sie diese Liste zusammen mit den Kontaktinformationen des Eigentümers in einer Datenbank, damit Sie Kontakt mit ihnen aufnehmen können. Da dies nicht das Hauptthema dieses Tutorials ist, werden wir der Einfachheit halber nur diese Python-Liste verwenden.

Wenn Sie wirklich gut aufgepasst haben, haben Sie möglicherweise zwei wirklich lange Domains in der Liste bemerkt, die keine gültigen Websites sind (ich hoffe, niemand hat sie gekauft, als Sie dies lesen, um mir das Gegenteil zu beweisen!). Ich habe diese beiden Domains hinzugefügt, um sicherzustellen, dass bei jedem Lauf einige Websites nicht verfügbar sind. Nennen wir auch unsere App UptimeSquirrel.

Serieller Ansatz

Versuchen wir zunächst den seriellen Ansatz und sehen, wie schlecht er funktioniert. Wir werden dies als Basis betrachten.

# serial_squirrel.py

import time


start_time = time.time()

for address in WEBSITE_LIST:
    check_website(address)
        
end_time = time.time()        

print("Time for SerialSquirrel: %ssecs" % (end_time - start_time))

# WARNING:root:Timeout expired for website http://really-cool-available-domain.com
# WARNING:root:Timeout expired for website http://another-really-interesting-domain.co
# WARNING:root:Website http://bing.com returned status_code=405
# Time for SerialSquirrel: 15.881232261657715secs

Threading-Ansatz

Wir werden mit der Implementierung des Threaded-Ansatzes etwas kreativer. Wir verwenden eine Warteschlange, um die Adressen einzufügen und Arbeitsthreads zu erstellen, um sie aus der Warteschlange zu entfernen und zu verarbeiten. Wir werden warten, bis die Warteschlange leer ist, was bedeutet, dass alle Adressen von unseren Arbeitsthreads verarbeitet wurden.

# threaded_squirrel.py

import time
from queue import Queue
from threading import Thread

NUM_WORKERS = 4
task_queue = Queue()

def worker():
    # Constantly check the queue for addresses
    while True:
        address = task_queue.get()
        check_website(address)
        
        # Mark the processed task as done
        task_queue.task_done()

start_time = time.time()
        
# Create the worker threads
threads = [Thread(target=worker) for _ in range(NUM_WORKERS)]

# Add the websites to the task queue
[task_queue.put(item) for item in WEBSITE_LIST]

# Start all the workers
[thread.start() for thread in threads]

# Wait for all the tasks in the queue to be processed
task_queue.join()

        
end_time = time.time()        

print("Time for ThreadedSquirrel: %ssecs" % (end_time - start_time))

# WARNING:root:Timeout expired for website http://really-cool-available-domain.com
# WARNING:root:Timeout expired for website http://another-really-interesting-domain.co
# WARNING:root:Website http://bing.com returned status_code=405
# Time for ThreadedSquirrel: 3.110753059387207secs

concurrent.futures

Wie bereits erwähnt, ist concurrent.futures eine API auf hoher Ebene für die Verwendung von Threads. Der Ansatz, den wir hier verfolgen, impliziert die Verwendung eines ThreadPoolExecutors. Wir werden Aufgaben an den Pool senden und Futures zurückerhalten. Dies sind Ergebnisse, die uns in Zukunft zur Verfügung stehen werden. Natürlich können wir warten, bis alle Futures zu tatsächlichen Ergebnissen werden.

# future_squirrel.py

import time
import concurrent.futures

NUM_WORKERS = 4

start_time = time.time()

with concurrent.futures.ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
    futures = {executor.submit(check_website, address) for address in WEBSITE_LIST}
    concurrent.futures.wait(futures)

end_time = time.time()        

print("Time for FutureSquirrel: %ssecs" % (end_time - start_time))

# WARNING:root:Timeout expired for website http://really-cool-available-domain.com
# WARNING:root:Timeout expired for website http://another-really-interesting-domain.co
# WARNING:root:Website http://bing.com returned status_code=405
# Time for FutureSquirrel: 1.812899112701416secs

Der Multiprocessing-Ansatz

Die multiprozessor-Bibliothek bietet eine fast Drop-In-Ersatz-API für die threading-Bibliothek. In diesem Fall werden wir einen Ansatz verfolgen, der dem von concurrent.futures ähnlicher ist. Wir richten ein multiprocessing.Pool ein und senden ihm Aufgaben, indem wir eine Funktion der Adressliste zuordnen (denken Sie an die klassische Python-map-Funktion).

# multiprocessing_squirrel.py

import time
import socket
import multiprocessing

NUM_WORKERS = 4

start_time = time.time()

with multiprocessing.Pool(processes=NUM_WORKERS) as pool:
    results = pool.map_async(check_website, WEBSITE_LIST)
    results.wait()

end_time = time.time()        

print("Time for MultiProcessingSquirrel: %ssecs" % (end_time - start_time))

# WARNING:root:Timeout expired for website http://really-cool-available-domain.com
# WARNING:root:Timeout expired for website http://another-really-interesting-domain.co
# WARNING:root:Website http://bing.com returned status_code=405
# Time for MultiProcessingSquirrel: 2.8224599361419678secs

Gevent

Gevent ist eine beliebte Alternative, um eine massive Parallelität zu erreichen. Es gibt einige Dinge, die Sie wissen müssen, bevor Sie es verwenden:

Code, der gleichzeitig von Greenlets ausgeführt wird, ist deterministisch. Im Gegensatz zu den anderen vorgestellten Alternativen garantiert dieses Paradigma, dass Sie für zwei identische Läufe immer die gleichen Ergebnisse in der gleichen Reihenfolge erhalten.
Sie müssen Standardfunktionen von Affen-Patches ausführen, damit diese mit gevent zusammenarbeiten. Das meine ich damit. Normalerweise blockiert ein Socket-Vorgang. Wir warten auf den Abschluss der Operation. Wenn wir uns in einer Multithread-Umgebung befinden, wechselt der Scheduler einfach zu einem anderen Thread, während der andere auf E/A wartet. Da wir uns nicht in einer Multithread-Umgebung befinden, patcht gevent die Standardfunktionen so, dass sie nicht mehr blockieren und die Kontrolle an den gevent-Scheduler zurückgeben.

Führen Sie zum Installieren von gevent Folgendes aus: pip install gevent

So verwenden Sie gevent, um unsere Aufgabe mit einem gevent.pool.Pool auszuführen:

# green_squirrel.py

import time
from gevent.pool import Pool
from gevent import monkey

# Note that you can spawn many workers with gevent since the cost of creating and switching is very low
NUM_WORKERS = 4

# Monkey-Patch socket module for HTTP requests
monkey.patch_socket()

start_time = time.time()

pool = Pool(NUM_WORKERS)
for address in WEBSITE_LIST:
    pool.spawn(check_website, address)

# Wait for stuff to finish
pool.join()
        
end_time = time.time()        

print("Time for GreenSquirrel: %ssecs" % (end_time - start_time))
# Time for GreenSquirrel: 3.8395519256591797secs

Celery

Sellerie ist ein Ansatz, der sich größtenteils von dem unterscheidet, was wir bisher gesehen haben. Es ist kampferprobt in sehr komplexen und leistungsstarken Umgebungen. Das Einrichten von Sellerie erfordert etwas mehr Basteln als alle oben genannten Lösungen.

Zuerst müssen wir Celery installieren:

pip install celery

Aufgaben sind die zentralen Konzepte innerhalb des Sellerieprojekts. Alles, was Sie in Sellerie ausführen möchten, muss eine Aufgabe sein. Sellerie bietet große Flexibilität beim Ausführen von Aufgaben: Sie können sie synchron oder asynchron, in Echtzeit oder geplant, auf demselben Computer oder auf mehreren Computern und unter Verwendung von Threads, Prozessen, Eventlet oder Gevent ausführen.

Die Anordnung wird etwas komplexer sein. Sellerie verwendet andere Dienste zum Senden und Empfangen von Nachrichten. Diese Nachrichten sind normalerweise Aufgaben oder Ergebnisse von Aufgaben. Wir werden Redis in diesem Tutorial für diesen Zweck verwenden. Redis ist eine gute Wahl, da es sehr einfach zu installieren und zu konfigurieren ist und Sie es möglicherweise bereits in Ihrer Anwendung für andere Zwecke verwenden, z. B. für Caching und pub/sub.

Sie können Redis installieren, indem Sie den Anweisungen auf der Redis-Schnellstartseite folgen. Vergessen Sie nicht, die redis Python-Bibliothek, pip install redis und das für die Verwendung von Redis und Celery erforderliche Bundle zu installieren: pip install cellery[redis].

Starten Sie den Redis-Server wie folgt: $ redis-server

Um mit dem Erstellen von Sellerie zu beginnen, müssen wir zuerst eine Sellerie-Anwendung erstellen. Danach muss Sellerie wissen, welche Art von Aufgaben er ausführen kann. Um dies zu erreichen, müssen wir Aufgaben in der Sellerie-Anwendung registrieren. Wir machen das mit dem @app.task Dekorator:

# celery_squirrel.py

import time
from utils import check_website
from data import WEBSITE_LIST
from celery import Celery
from celery.result import ResultSet

app = Celery('celery_squirrel',
             broker='redis://localhost:6379/0',
             backend='redis://localhost:6379/0')

@app.task
def check_website_task(address):
    return check_website(address)

if __name__ == "__main__":
    start_time = time.time()

    # Using `delay` runs the task async
    rs = ResultSet([check_website_task.delay(address) for address in WEBSITE_LIST])
    
    # Wait for the tasks to finish
    rs.get()

    end_time = time.time()

    print("CelerySquirrel:", end_time - start_time)
    # CelerySquirrel: 2.4979639053344727

Keine Panik, wenn nichts passiert. Denken Sie daran, Sellerie ist ein Dienst, und wir müssen ihn ausführen. Bisher haben wir die Aufgaben nur in Redis platziert, aber Celery nicht gestartet, um sie auszuführen. Dazu müssen wir diesen Befehl in dem Ordner ausführen, in dem sich unser Code befindet:

celery worker -A do_celery --loglevel=debug --concurrency=4

Führen Sie nun das Python-Skript erneut aus und sehen Sie, was passiert. Beachten Sie Folgendes: Beachten Sie, wie wir die Redis-Adresse zweimal an unsere Redis-Anwendung übergeben haben. Der broker-Parameter gibt an, wo die Aufgaben an Celery übergeben werden, und im backend legt Celery die Ergebnisse ab, damit wir sie in unserer App verwenden können. Wenn wir kein Ergebnis-backend angeben, können wir nicht wissen, wann die Aufgabe verarbeitet wurde und was das Ergebnis war.

Beachten Sie außerdem, dass sich die Protokolle jetzt in der Standardausgabe des Sellerieprozesses befinden. Überprüfen Sie sie daher unbedingt im entsprechenden Terminal.

Schlussfolgerungen

Ich hoffe, dies war eine interessante Reise für Sie und eine gute Einführung in die Welt der parallelen / gleichzeitigen Programmierung in Python. Dies ist das Ende der Reise, und wir können einige Schlussfolgerungen ziehen:

Es gibt verschiedene Paradigmen, die uns helfen, Hochleistungs-Computing in Python zu erreichen.
Für das Multithread-Paradigma haben wir die Bibliotheken threading und concurrent.futures.
multiprocessing bietet eine sehr ähnliche Schnittstelle zum threading, jedoch für Prozesse und nicht für Threads.
Denken Sie daran, dass Prozesse eine echte Parallelität erzielen, deren Erstellung jedoch teurer ist.
Denken Sie daran, dass in einem Prozess möglicherweise mehr Threads ausgeführt werden.
Verwechseln Sie nicht parallel mit gleichzeitig. Denken Sie daran, dass nur der parallele Ansatz Multi-Core-Prozessoren nutzt, während die gleichzeitige Programmierung Aufgaben intelligent plant, sodass das Warten auf lang laufende Vorgänge während der parallelen eigentlichen Berechnung erfolgt.