Worker

Worker sind dauerhaft laufende Prozesse, die die Datenbank auf ausstehende Jobs prüfen, diese ausführen und Workflows fortsetzen. Der Befehl wp queuety work läuft innerhalb von WP-CLI, während die Warteschlange direkt aus MySQL über den jeweils besten verfügbaren Queuety-Datenbanktreiber abgerufen wird.

Das Plugin installiert außerdem einen standardmäßigen WordPress-cron-Fallback, der einmal pro Minute läuft und für Umgebungen ohne Shell-Zugriff gedacht ist. Diese Seite behandelt dedizierte WP-CLI-Worker, die empfohlene Bereitstellungsart für einen höheren Durchsatz.

Worker starten

wp queuety work

Damit wird ein einzelner Worker für die Warteschlange default gestartet. Er läuft in einer Schleife: Job abrufen, ausführen, wiederholen. Ist die Warteschlange leer, wartet er QUEUETY_WORKER_SLEEP Sekunden (Standard: 1), bevor er erneut abfragt.

Optionen

# Process a specific queue
wp queuety work --queue=emails

# Process at most one job and exit
wp queuety work --once

# Run 4 concurrent workers
wp queuety work --workers=4

# Scale between 2 and 6 workers as backlog changes
wp queuety work --queue=providers --min-workers=2 --max-workers=6

Prioritäten für mehrere Warteschlangen

Ein einzelner Worker kann mehrere Warteschlangen in Prioritätsreihenfolge verarbeiten. Übergib --queue eine durch Kommas getrennte Liste von Warteschlangennamen:

wp queuety work --queue=critical,default,low

Der Worker versucht, in der angegebenen Reihenfolge einen Job aus jeder Warteschlange abzurufen. Zuerst prüft er critical. Wenn critical einen Job enthält, verarbeitet er ihn. Ist critical leer, wechselt er zu default und danach zu low. Das bedeutet, dass Warteschlangen mit höherer Priorität immer geleert werden, bevor die mit niedrigerer Priorität an der Reihe sind.

Funktionsweise der Prioritätsreihenfolge

Bei jedem Durchlauf versucht der Worker, einen Job aus der ersten Warteschlange der Liste abzurufen. Ist diese leer oder pausiert, geht er zur nächsten Warteschlange über, bis er einen Job findet oder die Liste abgearbeitet hat. Wird nirgends ein Job gefunden, wartet der Worker und versucht es erneut. Bei diesem strikten Prioritätsmodell können anhaltend viele Jobs aus critical dazu führen, dass default und low nie zum Zug kommen. Wenn Warteschlangen mit niedrigerer Priorität garantiert abgearbeitet werden müssen, verwende stattdessen dedizierte Worker.

Mehrere Warteschlangen in PHP verwenden

Die Methoden Worker::run() und Worker::flush() akzeptieren entweder eine Zeichenfolge oder ein Array:

// Comma-separated string
Queuety::worker()->run( 'critical,default,low' );

// Array
Queuety::worker()->run( [ 'critical', 'default', 'low' ] );

// flush() also supports multi-queue
$count = Queuety::worker()->flush( 'critical,default,low' );

Bereitstellungsstrategien

Dedizierte Worker pro Warteschlange bieten dir volle Kontrolle über Nebenläufigkeit und Isolation. Jede Warteschlange erhält eine eigene Worker-process(es) und kann nicht durch andere Warteschlangen ausgebremst werden:

[program:queuety-critical]
command=wp queuety work --queue=critical
numprocs=4

[program:queuety-default]
command=wp queuety work --queue=default
numprocs=2

[program:queuety-low]
command=wp queuety work --queue=low
numprocs=1

Prioritätsreihenfolge mit einem einzelnen Worker ist einfacher zu verwalten und eignet sich gut, wenn Warteschlangen mit niedrigerer Priorität Verzögerungen verkraften:

[program:queuety-worker]
command=wp queuety work --queue=critical,default,low
numprocs=2

Du kannst auch beide Ansätze kombinieren: dedizierte Worker für critical, um den Durchsatz sicherzustellen, und einen Worker mit Prioritätsreihenfolge für alles andere:

[program:queuety-critical]
command=wp queuety work --queue=critical
numprocs=2

[program:queuety-fallback]
command=wp queuety work --queue=default,low
numprocs=2

Ressourcenbewusste Zulassung

Worker erledigen mehr, als nur die nächste ausstehende Zeile zu beanspruchen. Bevor ein beanspruchter Job startet, kann Queuety grobe Zulassungsprüfungen anhand der Parallelitätsgruppe des Jobs und des aktuellen Ausführungsprofils für diesen Handler durchführen.

Das umfassendere Modell und das Zusammenspiel mit Job-Metadaten und Workflow-Budgets findest du unter Ressourcenverwaltung.

Eine Parallelitätsgruppe legt eine gemeinsame Obergrenze für mehrere Handler oder Jobklassen fest. Wenn du mehrere ressourcenintensive Anbieter derselben Gruppe und demselben Limit zuordnest, gibt ein Worker neu beanspruchte Arbeit wieder frei, statt sie zu starten, sobald die Gruppe ausgelastet ist. So überlasten unabhängige Warteschlangen nicht dasselbe nachgelagerte System, nur weil mehrere Worker untätig sind.

Aktuelle Ausführungsprotokolle fließen ebenfalls in ein einfaches Zulassungsprofil ein. Wenn die Ressourcenfreigabe aktiviert ist, prüft Queuety die durchschnittliche Dauer und den Spitzenwert des Speicherbedarfs für den Handler, den es gleich ausführen soll. Wenn der aktuelle Worker bereits nahe an seiner konfigurierten Speichergrenze ist, eine Warteschlange oder Anbietergruppe ihr gewichtetes Budget erreicht hat, der aktuelle Container für das beobachtete Profil des Handlers nicht genug freien Speicher zu haben scheint oder ein Lauf im Stil von --once nahe an sein Zeitlimit kommt, stellt Queuety den Job zurück, statt eine Arbeit zu starten, die wahrscheinlich nicht ordnungsgemäß abgeschlossen wird.

Du aktivierst die Richtlinie in der Arbeitsdefinition und nicht über den Worker-Befehl:

class CallProviderStep implements \Queuety\Step {
    public function handle( array $state ): array {
        // ...
    }

    public function config(): array {
        return [
            'concurrency_group' => 'providers',
            'concurrency_limit' => 3,
            'cost_units' => 5,
        ];
    }
}

Dieselben Metadaten können über öffentliche Eigenschaften an ausführbare Jobklassen oder über PendingJob::concurrency_group() und PendingJob::cost_units() an einen einzelnen Dispatch angehängt werden.

Das Zulassungsprofil auf der Worker-Seite wird über diese Konstanten gesteuert:

Worker-Lebenszyklus

Ein Worker beendet sich ordnungsgemäß nach Abschluss seines aktuellen Jobs, sobald eine der folgenden Bedingungen erfüllt ist:

Nach dem Anhalten wird der Worker sauber beendet. In einer Produktionsumgebung startet ihn ein Prozessmanager automatisch neu.

Signale werden noch nicht ordnungsgemäß verarbeitet (#4). Ein einzelner wp queuety work-Prozess installiert keinen Signal-Handler, daher wird er durch SIGTERM oder SIGINT (Ctrl+C) sofort beendet, auch mitten in einem Job. Dieser Job bleibt processing, bis die Wiederherstellung veralteter Jobs ihn nach QUEUETY_STALE_TIMEOUT Sekunden zurücksetzt. Sorge daher dafür, dass Handler sicher erneut ausgeführt werden können. Ein Worker-Pool leitet SIGTERM an seine Kindprozesse weiter, aber diese reagieren nicht darauf. Der Pool läuft also weiter, bis sie aus einem der oben genannten Gründe stoppen. Wenn du einen Pool stoppen musst, konfiguriere deinen Prozessmanager so, dass er nach einer Wartezeit SIGKILL sendet.

Parallelität mit mehreren Workern

Das Flag --workers=N startet mit pcntl_fork() N Kindprozesse. Jeder Kindprozess stellt eine eigene Datenbankverbindung her und verarbeitet Jobs unabhängig. Der Elternprozess überwacht die Kindprozesse und startet abgestürzte Prozesse neu.

wp queuety work --queue=default --workers=4

Anforderungen: Die Erweiterungen pcntl und posix PHP müssen installiert sein (ohne pcntl schlägt der Befehl mit „pcntl extension is required for --workers=N.“ fehl). Jeder Kindprozess verwendet eine eigene Datenbankverbindung, daher muss deine MySQL-Einstellung max_connections für den Pool ausreichen. Für adaptive Pools (--min-workers/--max-workers) gelten dieselben Anforderungen.

Adaptive Worker-Pools

Adaptive Pools halten eine Mindestanzahl an Workern bereit und skalieren dann innerhalb eines konfigurierten Bereichs hoch und herunter:

wp queuety work --queue=providers --min-workers=2 --max-workers=6

Der Elternprozess überwacht den abrufbaren Rückstand in den ausgewählten Warteschlangen und passt die Anzahl der Kindprozesse daran an, begrenzt durch das konfigurierte Maximum. Wenn es in der Warteschlange ruhiger wird, wartet der Pool das Leerlaufintervall ab, bevor er verkleinert wird. So führt ein kurzer Anstieg der Auslastung nicht sofort zu Änderungen an der Prozessliste.

Die Aufwärtsskalierung bleibt vorsichtig. Wenn Queuety Telemetriedaten zum Speicher von Container oder Host auslesen kann, fügt es keine weiteren Kindprozesse hinzu, falls der freie Speicher nicht für den vollen Bedarf eines weiteren Workers ausreicht.

Die Skalierungsschleife auf Elternprozess-Seite wird über Folgendes gesteuert:

Wiederherstellung veralteter Jobs

Stürzt ein Worker während der Verarbeitung eines Jobs ab, bleibt der Job im Status processing. Ein Worker mit langer Laufzeit prüft etwa einmal pro Minute, ob sich Jobs länger als QUEUETY_STALE_TIMEOUT Sekunden im Status processing befinden. Ein veralteter Job mit verbleibenden Versuchen wird nach der Wartezeit erneut ausgeführt. Ein Job, der alle Versuche aufgebraucht hat, wird zurückgestellt und sein Workflow als fehlgeschlagen markiert. --once-Worker und der WP-Cron-Fallback führen diese Prüfung nicht aus.

Veraltete Jobs manuell wiederherstellen:

wp queuety recover

Leeren-Modus

Alle ausstehenden Jobs verarbeiten und beenden. Nützlich für Tests oder einmalige Stapelverarbeitung:

wp queuety flush
wp queuety flush --queue=emails

Bereitstellung in der Produktion

Verwende in der Produktion einen Prozessmanager, damit die Worker weiterlaufen. Hier sind zwei gängige Möglichkeiten.

Supervisord

[program:queuety-default]
command=wp queuety work --queue=default
directory=/var/www/html
autostart=true
autorestart=true
numprocs=2
process_name=%(program_name)s_%(process_num)02d
stdout_logfile=/var/log/queuety-default.log
stderr_logfile=/var/log/queuety-default-error.log

[program:queuety-emails]
command=wp queuety work --queue=emails
directory=/var/www/html
autostart=true
autorestart=true
numprocs=1
stdout_logfile=/var/log/queuety-emails.log
stderr_logfile=/var/log/queuety-emails-error.log

Systemd

[Unit]
Description=Queuety Worker (default queue)
After=mysql.service

[Service]
Type=simple
User=www-data
WorkingDirectory=/var/www/html
ExecStart=/usr/local/bin/wp queuety work --queue=default
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Aktivieren und starten:

sudo systemctl enable queuety-worker
sudo systemctl start queuety-worker

Tipps für die Bereitstellung

Führe separate Worker für einzelne Warteschlangen aus, wenn du eine stärkere Isolierung brauchst. Verwende --workers=N bei CPU-intensiven Workloads und bevorzuge einzeln überwachte Worker, wenn der Workload hauptsächlich auf Ein- und Ausgabe basiert. Setze QUEUETY_WORKER_MAX_JOBS, wenn regelmäßige Neustarts Speicher freigeben sollen, und behalte die Anzahl von buried über Webhooks oder Metriken im Blick.