Jobs

Jobs sind die zentrale Arbeitseinheit in Queuety. Löse einen Job mit Handler-Namen und Payload aus; ein Worker übernimmt ihn und führt ihn im Hintergrund aus.

Jobs auslösen

use Queuety\Queuety;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );

dispatch() gibt einen PendingJob-Builder zurück. Du kannst Optionen verketten, bevor der Job automatisch ausgelöst wird:

use Queuety\Enums\Priority;

Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] )
    ->on_queue( 'emails' )
    ->with_priority( Priority::High )
    ->delay( 300 )
    ->max_attempts( 5 );

Rufe ->id() auf, um sofort die Job-ID zu erhalten:

$job_id = Queuety::dispatch( 'send_email', $payload )->id();

Handler schreiben

Ein Handler ist eine beliebige Klasse, die das Interface Queuety\Handler implementiert:

use Queuety\Handler;

class SendEmailHandler implements Handler {
    public function handle( array $payload ): void {
        wp_mail( $payload['to'], $payload['subject'], $payload['body'] );
    }

    public function config(): array {
        return [
            'queue'           => 'emails',
            'max_attempts'    => 5,
            'backoff'         => 'exponential',
        ];
    }
}

Die Methode handle() erhält die Job-Payload und führt die Arbeit aus. Die Methode config() gibt optionale Standardwerte zurück:

Diese Ressourcenschlüssel sind Teil des umfassenderen Modells für Ressourcenverwaltung von Queuety, in dem Job-Metadaten, Workflow-Budgets und Worker-Aufnahme zusammenwirken.

Registriere den Handler:

Queuety::register( 'send_email', SendEmailHandler::class );

Oder verwende PHP-Attribute zur automatischen Registrierung.

Priorität

Jobs haben vier Prioritätsstufen. Worker beanspruchen Jobs mit höherer Priorität zuerst.

use Queuety\Enums\Priority;

Queuety::dispatch( 'handler', $payload )->with_priority( Priority::Urgent );

Verzögerung

Verzögere einen Job, damit er erst nach der angegebenen Anzahl Sekunden verfügbar wird:

Queuety::dispatch( 'send_reminder', $payload )->delay( 3600 ); // 1 hour

Der Job wird sofort eingefügt, aber Worker beanspruchen ihn erst nach Ablauf der Verzögerung.

Eindeutige Jobs

Verhindere doppelte Auslösungen für denselben Handler und dieselbe Payload. Wenn bereits ein ausstehender oder verarbeiteter Job mit demselben Handler und derselben Payload existiert, wird dessen Job-ID zurückgegeben, statt einen neuen Job anzulegen.

Queuety::dispatch( 'sync_inventory', [ 'sku' => 'ABC-123' ] )->unique();

Job-Abhängigkeiten

Ein Job kann von einem anderen Job abhängen. Ein Worker beansprucht ihn erst, wenn die Abhängigkeit abgeschlossen ist.

$first_id = Queuety::dispatch( 'import_data', $payload )->id();

Queuety::dispatch( 'send_notification', [ 'import_id' => $first_id ] )
    ->after( $first_id );

Batch-Auslösung

Löse mehrere Jobs in einem einzigen INSERT mit mehreren Zeilen aus, um den Durchsatz zu maximieren:

$job_ids = Queuety::batch( [
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'a@example.com' ] ],
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'b@example.com' ] ],
    [ 'handler' => 'send_email', 'payload' => [ 'to' => 'c@example.com' ] ],
] );

Jeder Eintrag kann optionale Schlüssel enthalten: queue, priority, delay, max_attempts.

Wiederholungen und Backoff

Wenn ein Job eine Ausnahme auslöst, setzt Queuety ihn mit einem verzögerten available_at zurück auf pending, bis er insgesamt max_attempts-mal ausgeführt wurde (der erste Lauf zählt; max_attempts( 3 ) bedeutet also bis zu zwei Wiederholungen). Die Wartezeit vor jedem erneuten Versuch richtet sich nach der Backoff-Strategie. Dabei bezeichnet attempt die Nummer des gerade fehlgeschlagenen Laufs:

Lege den globalen Standard in wp-config.php fest:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Oder pro Handler in config():

public function config(): array {
    return [ 'backoff' => 'exponential', 'max_attempts' => 5 ];
}

Oder per Dispatch:

Queuety::dispatch( 'handler', $payload )->max_attempts( 10 );

Zurückgestellte Jobs

Wenn alle Wiederholungsversuche ausgeschöpft sind, wird ein Job als begraben markiert. Begrabene Jobs bleiben zur Prüfung in der Datenbank, werden aber nicht verarbeitet. Du kannst sie manuell erneut versuchen. Bei einem manuellen erneuten Versuch wird der Job wieder zu pending, ohne den Versuchszähler zurückzusetzen. Ein erneut versuchter begrabener Job erhält also einen weiteren Lauf und wird wieder begraben, wenn dieser Lauf fehlschlägt:

// Retry a specific buried job
Queuety::retry( $job_id );

// Retry all buried jobs
$count = Queuety::retry_buried();

Oder über CLI:

wp queuety retry 42
wp queuety retry-buried

Timeout

Queuety startet einen pcntl_alarm() für die maximale Ausführungszeit und soll den Versuch mit einem TimeoutException fehlschlagen lassen, wenn diese überschritten wird. Dieses Timeout wird derzeit nicht erzwungen (#10): Der Worker aktiviert keine asynchronen Signale, sodass ein lang laufender Handler bis zum Ende ausgeführt wird. Ein in sleep() blockierter Handler wird beim Auslösen des Alarms abgebrochen und anschließend als abgeschlossen markiert. Konfiguriere das Timeout global:

define( 'QUEUETY_MAX_EXECUTION_TIME', 300 ); // 5 minutes

Jobs, die länger als QUEUETY_STALE_TIMEOUT Sekunden den Status processing haben, werden automatisch wiederhergestellt:

define( 'QUEUETY_STALE_TIMEOUT', 600 ); // 10 minutes

Veraltete Jobs manuell wiederherstellen:

wp queuety recover

Warteschlangen

Jobs werden an benannte Warteschlangen gesendet. Jeder Worker verarbeitet eine Warteschlange oder eine geordnete Liste von Warteschlangen (siehe Workers):

Queuety::dispatch( 'handler', $payload )->on_queue( 'emails' );

wp queuety work --queue=emails

Die Standardwarteschlange ist default. Du kannst für jede Warteschlange einen eigenen Worker-Prozess ausführen, um Workloads voneinander zu isolieren.

Ratenbegrenzung

Begrenze, wie oft ein Handler innerhalb eines Zeitfensters ausgeführt werden kann:

Queuety::dispatch( 'call_api', $payload )
    ->rate_limit( 10, 60 ); // max 10 executions per 60 seconds

Weitere Details findest du unter Ratenbegrenzung.

Dispatchbare Jobklassen

Queuety v0.6.0 führt ein klassenbasiertes Job-API ein. Statt Handlernamen, Payload und Handlerklasse getrennt zu verwalten, definierst du eine einzelne Klasse, die alles kapselt: die Payload (als typisierte öffentliche Eigenschaften), die Ausführungslogik und optionale Middleware.

Die Schnittstelle Contracts\Job

use Queuety\Contracts\Job;

interface Job {
    public function handle(): void;
}

Eine Job-Klasse speichert ihre Payload in öffentlichen Eigenschaften und führt ihre eigene Logik in handle() aus. Eine separate Methode config() oder Handlerregistrierung ist nicht erforderlich.

Das Trait Dispatchable

Füge deiner Jobklasse das Trait Dispatchable hinzu, damit sie eine statische Methode dispatch() erhält:

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class SendEmailJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $to,
        public readonly string $subject,
        public readonly string $body,
    ) {}

    public function handle(): void {
        wp_mail( $this->to, $this->subject, $this->body );
    }
}

Sende den Job, indem du die Konstruktorargumente übergibst:

SendEmailJob::dispatch( 'user@example.com', 'Welcome!', 'Hello there.' )
    ->on_queue( 'emails' )
    ->with_priority( Priority::High );

Das Trait Dispatchable ruft intern Queuety::dispatch() auf, sodass alle Optionen von PendingJob (queue, priority, delay, max_attempts, unique, after) genauso funktionieren.

So funktioniert die Serialisierung

Wenn du eine Instanz von Job sendest, extrahiert Queuetys JobSerializer den vollständig qualifizierten Klassennamen als Handler und alle öffentlichen Eigenschaften als Payload. Auf der Worker-Seite rekonstruiert der Serializer die Instanz, indem er die Payload-Schlüssel den Namen der Konstruktorparameter zuordnet. Backed Enums werden in ihren zugrunde liegenden Wert serialisiert und automatisch wiederhergestellt.

Job-Eigenschaften

Jobklassen können öffentliche Eigenschaften definieren, um Wiederholungs- und Timeout-Verhalten zu steuern. Der Worker liest diese per Reflection aus und wendet sie automatisch an.

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class ImportProductsJob implements Job {
    use Dispatchable;

    public int $tries = 5;
    public int $timeout = 120;
    public array $backoff = [ 10, 30, 60, 120 ];
    public string $concurrency_group = 'imports';
    public int $concurrency_limit = 2;
    public int $cost_units = 4;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Import products...
    }
}

Das Array $backoff legt zunehmend längere Verzögerungen zwischen Wiederholungsversuchen fest. Zum Beispiel bedeutet [10, 30, 60, 120]: Vor dem ersten Wiederholungsversuch wird 10 Sekunden gewartet, vor dem zweiten 30 Sekunden, vor dem dritten 60 Sekunden und vor allen weiteren 120 Sekunden.

Eigenschaften zu Ressourcen sind nützlich, wenn Jobklassen unterschiedliche Kostenprofile abbilden. Eine schnelle Cache-Aktualisierung und ein umfangreicher Aufruf bei einem Anbieter können beide „ein Job“ sein, sollten aber nicht unbedingt gleich gewichtet werden, wenn ein Workflow-Budget oder eine Worker-Zulassungsentscheidung den Prozess innerhalb sicherer Grenzen halten soll.

Das vollständige Modell einschließlich Workflow-Budgets und Zulassung auf Worker-Seite findest du unter Ressourcenverwaltung.

Hook bei Fehlschlag

Wenn ein Job nach Ausschöpfen aller Wiederholungsversuche begraben wird, ruft Queuety eine Methode failed() auf der Jobinstanz auf, sofern eine definiert ist. Nutze sie für Bereinigungen, Warnmeldungen oder Protokollierung.

use Queuety\Contracts\Job;
use Queuety\Dispatchable;

class ProcessPaymentJob implements Job {
    use Dispatchable;

    public int $tries = 3;

    public function __construct(
        public readonly int $order_id,
    ) {}

    public function handle(): void {
        // Process the payment...
    }

    public function failed( \Throwable $exception ): void {
        // Notify the team that payment processing failed permanently.
        error_log( "Payment failed for order {$this->order_id}: {$exception->getMessage()}" );
    }
}

Die Methode failed() erhält die Ausnahme, die zum endgültigen Fehlschlag geführt hat. Die Jobinstanz wird mit ihrer ursprünglichen Payload deserialisiert, sodass alle öffentlichen Eigenschaften verfügbar sind.

Bedingter Dispatch

Das Trait Dispatchable stellt dispatch_if() und dispatch_unless() für bedingtes Senden bereit. Beide erwarten als erstes Argument eine boolesche Bedingung, gefolgt von den Konstruktorargumenten des Jobs.

// Only dispatch if the user has opted in to notifications.
SendEmailJob::dispatch_if( $user->wants_notifications, $user->email, 'Update', $body );

// Dispatch unless the feature flag is disabled.
SyncInventoryJob::dispatch_unless( $maintenance_mode, $sku );

dispatch_if() sendet den Job, wenn die Bedingung true ist, und gibt das PendingJob zurück. Wenn die Bedingung false ist, gibt es null zurück.

dispatch_unless() sendet den Job, wenn die Bedingung false ist, und gibt das PendingJob zurück. Wenn die Bedingung true ist, gibt es null zurück.

Beide Methoden unterstützen das Verketten von PendingJob-Optionen beim Senden:

SendEmailJob::dispatch_if( $should_send, $to, $subject, $body )
    ?->on_queue( 'emails' )
    ?->with_priority( Priority::High );

Synchrone Ausführung

Nutze Queuety::dispatch_sync(), um einen Job sofort im aktuellen Prozess auszuführen, ohne ihn zur Warteschlange hinzuzufügen. Das ist hilfreich beim Testen und für Jobs, die direkt ausgeführt werden müssen.

use Queuety\Queuety;

// With a Job instance
Queuety::dispatch_sync( new SendEmailJob( to: 'user@example.com', subject: 'Hello', body: 'Hi!' ) );

// With a handler name and payload
Queuety::dispatch_sync( 'send_email', [ 'to' => 'user@example.com' ] );

Die Methode handle() des Jobs wird direkt aufgerufen. Es wird kein Datenbankeintrag erstellt, kein Worker eingesetzt und keine Middleware angewendet.

Jobverkettung

Bei einer Jobverkettung wird eine Folge von Jobs nacheinander ausgeführt. Jeder Job in der Kette hängt davon ab, dass der vorherige erfolgreich abgeschlossen wird. Wenn ein Job fehlschlägt, bleiben die nachfolgenden Jobs in der Kette ausstehend.

Mit Queuety::chain()

use Queuety\Queuety;

$first_job_id = Queuety::chain( [
    new FetchDataJob( $url ),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->catch( ChainFailedHandler::class )
    ->dispatch();

Queuety::chain() gibt ein ChainBuilder mit Fluent-Methoden zum Festlegen der Warteschlange und eines Fehlerhandlers zurück. Die Methode dispatch() erstellt alle Jobs mit sequenziellen depends_on-Beziehungen und gibt die ID des ersten Jobs zurück.

Mit with_chain()

Das Trait Dispatchable stellt eine Methode with_chain() bereit. Sie entspricht derzeit Queuety::chain( $jobs ): Die Klasse, auf der sie aufgerufen wird, wird nicht zur Kette hinzugefügt (#13). Führe daher auch den ersten Job im Array auf:

FetchDataJob::with_chain( [
    new FetchDataJob(),
    new ProcessDataJob(),
    new NotifyCompleteJob(),
] )
    ->on_queue( 'pipeline' )
    ->dispatch();

Fehlerbehandlung bei Ketten

Registriere mit catch() eine Fehlerhandlerklasse. Die Handlerklasse muss über eine Methode handle() verfügen. Sie wird aufgerufen, wenn ein Job in der Kette endgültig begraben wird. Der Worker unterstützt handle(), handle( Job $job ) und handle( Job $job, \Throwable $exception ).

class ChainFailedHandler {
    public function handle(): void {
        // Alert, clean up, etc.
    }
}

Middleware hinzufügen

Jobklassen können eine Methode middleware() definieren, die Middleware zurückgibt, um die Ausführung zu umschließen:

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class CallOpenAIJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $prompt,
    ) {}

    public function handle(): void {
        // Call the API...
    }

    public function middleware(): array {
        return [
            new RateLimited( max: 60, window: 60 ),
            new Timeout( seconds: 120 ),
        ];
    }
}

Details zu integrierter Middleware und zum Schreiben eigener Middleware findest du unter Middleware.

Vergleich mit handlerbasierten Jobs

Beide Ansätze werden vollständig unterstützt. Bestehende handlerbasierte Jobs funktionieren unverändert weiter. Du kannst beide Stile im selben Projekt kombinieren.

Job-Lebenszyklus

dispatch() -> pending -> processing -> completed
                  |           |
                  |           +-> exception -> pending (retry after backoff)
                  |                       |
                  |                       +-> buried (max attempts reached)
                  |
                  +-> delayed (available_at in the future)