Python-Multiprocessing-Protokollierung

Das Python-Modul multiprocessing führt Aufgaben in getrennten Prozessen aus. Sie können einzelne Prozesse starten und auf ihr Ende warten, mit einem Pool dieselbe Funktion auf mehrere Eingaben anwenden und Protokolleinträge der Worker an einen Listener im Elternprozess senden. Die folgenden Beispiele zeigen, wie Sie diese Muster einrichten und erkennen, welcher Prozess eine Meldung erzeugt hat.

Multiprocessing

Beim Multiprocessing werden Aufgaben in getrennten Prozessen mit jeweils eigenem Speicher ausgeführt. Es eignet sich für CPU-intensive Arbeit, die sich in unabhängige Aufgaben aufteilen lässt. Das Betriebssystem plant die Prozesse ein; deshalb kann die Reihenfolge ihrer Ausgaben variieren.

Multiprocessing in Python

Importieren Sie das Modul multiprocessing, um Prozesse zu erstellen. Ein Process erhält eine aufrufbare Funktion als target und ein Tupel mit Argumenten als args. Rufen Sie start() auf, um den Worker zu starten, und join(), um auf sein Ende zu warten. Die Hauptschutzabfrage verhindert, dass Kindprozesse den Code zur Prozesserstellung auf Programmebene erneut ausführen. Das gilt auch für Plattformen, auf denen Worker durch Import des Hauptmoduls gestartet werden.

Führen Sie dieses Beispiel aus:

import multiprocessing


def report(value):
    print(f"worker received: {value}", flush=True)


if __name__ == "__main__":
    process = multiprocessing.Process(target=report, args=(5,))
    process.start()
    process.join()
    print("parent: process finished")

worker received: 5
parent: process finished

Der Worker gibt seine Meldung aus, bevor join() zurückkehrt; deshalb folgt darauf die Meldung des Elternprozesses. Bei mehreren Workern können sich Meldungen überlagern oder in anderer Reihenfolge erscheinen. Verwenden Sie eine Warteschlange, wenn die Ausgaben zentral verarbeitet werden sollen.

Prozess-IDs

Rufen Sie innerhalb eines Workers os.getpid() auf, um den Prozess zu ermitteln, in dem er gerade läuft. Nach start() kann der Elternprozess dieselbe Prozesskennung über das Attribut pid des Process-Objekts auslesen. Das Betriebssystem weist PIDs zu; sie unterscheiden sich von Lauf zu Lauf. Betrachten Sie sie daher als Laufzeitwerte und nicht als feste Ausgabe.

import multiprocessing
import os


def report_pid():
    print(f"worker PID: {os.getpid()}", flush=True)


if __name__ == "__main__":
    process = multiprocessing.Process(target=report_pid)
    process.start()
    print(f"Process.pid: {process.pid}")
    process.join()

Arbeit mit Pool abbilden

Ein Pool verwaltet eine Gruppe von Worker-Prozessen. Seine Methode map() wendet eine Funktion auf jedes Element eines iterierbaren Objekts an und gibt die Ergebnisse in der Reihenfolge der Eingaben zurück. Die Pool-Größe gibt die maximale Anzahl an Workern an; sie garantiert nicht einen Prozess pro Eingabe. Beim Verlassen des Kontextmanagers wird der Pool am Ende des Blocks terminiert; wenn ein ordnungsgemäßes Beenden wichtig ist, schließen Sie den Pool stattdessen explizit für neue Aufgaben und warten Sie mit join() auf das Ende seiner Worker.

import multiprocessing


def square(value):
    return value * value


if __name__ == "__main__":
    with multiprocessing.Pool(processes=2) as pool:
        results = pool.map(square, [2, 3, 4])
    print(results)

Protokollierung mit Multiprocessing in Python

Worker-Prozesse sollten ihre Protokolleinträge an eine gemeinsame Warteschlange senden. Ein QueueListener im Elternprozess leitet diese Einträge an Ausgabe-Handler weiter. So werden Datei- oder Konsolenausgaben in einem Prozess verarbeitet, statt dass mehrere Worker unabhängig in denselben Handler schreiben. Der Logger jedes Workers erhält einen Queue-Handler; der Listener bekommt die Warteschlange und einen oder mehrere Handler.

Dieses vollständige Beispiel richtet einen Pool-Worker mit einem QueueHandler ein. Der Elternprozess startet den Listener, bevor er den Pool erstellt. Nach Abschluss der Zuordnung nimmt der Pool keine neuen Aufgaben mehr an; seine Worker werden beendet, bevor der Listener gestoppt wird. Beim Beenden leeren die Worker die von ihren Queue-Feedern gepufferten Einträge, sodass der Listener sie verarbeiten kann. Dieselbe Aufräumsequenz läuft auch, wenn bei der Zuordnung ein Fehler auftritt: Die bereits erzeugten Worker-Einträge können verarbeitet werden, bevor der Fehler weitergegeben wird. Der Formatierer ergänzt jede Meldung um ihre Protokollierungsstufe, und der Elternprozess schreibt die Einträge nach stdout. Ein einzelner Worker verarbeitet diese Elemente der Reihe nach. Wenn Sie einen Worker anhand seiner Prozess-ID erkennen müssen, verwenden Sie das vorherige PID-Beispiel.

import logging
import multiprocessing
import sys
from logging.handlers import QueueHandler, QueueListener


def configure_worker(log_queue):
    logger = logging.getLogger()
    logger.handlers.clear()
    logger.setLevel(logging.INFO)
    logger.addHandler(QueueHandler(log_queue))


def report(value):
    logger = logging.getLogger(__name__)
    logger.info("item=%s", value)
    return value


if __name__ == "__main__":
    log_queue = multiprocessing.Queue()
    output_handler = logging.StreamHandler(sys.stdout)
    output_handler.setFormatter(logging.Formatter("%(levelname)s %(message)s"))
    listener = QueueListener(log_queue, output_handler)
    listener.start()

    try:
        pool = multiprocessing.Pool(
            processes=1,
            initializer=configure_worker,
            initargs=(log_queue,),
        )
        try:
            results = pool.map(report, [1, 2, 3])
        finally:
            pool.close()
            pool.join()
    finally:
        listener.stop()
        log_queue.close()
        log_queue.join_thread()

INFO item=1
INFO item=2
INFO item=3

Dieses Beispiel wurde mit Exitcode 0 abgeschlossen und erzeugte für jede zugeordnete Eingabe einen INFO-Eintrag. Bei einem Worker erschienen die Einträge in der Reihenfolge der Zuordnung. Das unveränderte Beispiel zeigt die Protokollierungsstufe INFO auf Englisch; sie wurde nicht lokalisiert, damit der Code- und Ausgabeblock bytegenau erhalten bleibt.

Um Einträge statt auf der Konsole in eine Datei zu schreiben, ersetzen Sie logging.StreamHandler(sys.stdout) durch logging.FileHandler("multiprocessing.log", encoding="utf-8"). Der Elternprozess besitzt und schließt diesen Handler. Wenn Sie ihn ausdrücklich verwalten, statt sich auf das Herunterfahren des Interpreters zu verlassen, schließen Sie ihn nach listener.stop(). Für rotierende Protokolle oder mehrere Ausgabeziele übergeben Sie die zusätzlichen Handler an QueueListener und konfigurieren jeden davon mit dem benötigten Formatierer.