Jobs

Jobs vormen de basiseenheid van werk in Queuety. Dispatch een job met een handlernaam en payload; een worker pakt de job op en voert deze op de achtergrond uit.

Jobs dispatchen

use Queuety\Queuety;

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

dispatch() retourneert een PendingJob-builder. U kunt opties koppelen voordat deze automatisch wordt gedispatcht:

use Queuety\Enums\Priority;

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

Roep ->id() aan om direct de job-ID te krijgen:

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

Handlers schrijven

Een handler is elke klasse die de Queuety\Handler-interface implementeert:

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',
        ];
    }
}

De methode handle() ontvangt de jobpayload en voert het werk uit. De methode config() retourneert optionele standaardwaarden:

Deze resource-sleutels maken deel uit van het bredere Queuety-model voor Resourcebeheer, waarin jobmetadata, workflowbudgetten en toelating van workers samenwerken.

Registreer de handler:

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

Of gebruik attributen PHP voor automatische registratie.

Prioriteit

Jobs hebben vier prioriteitsniveaus. Workers halen jobs met een hogere prioriteit als eerste op.

use Queuety\Enums\Priority;

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

Vertraging

Vertraag een job zodat deze pas beschikbaar wordt nadat het opgegeven aantal seconden is verstreken:

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

De job wordt direct ingevoegd, maar workers halen deze pas op nadat de vertraging is verstreken.

Unieke jobs

Voorkom dubbele dispatches voor dezelfde handler en payload. Als er al een wachtende of verwerkende job met dezelfde handler en payload bestaat, wordt de ID van die bestaande job geretourneerd in plaats van een nieuwe aan te maken.

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

Jobafhankelijkheden

Een job kan afhankelijk zijn van een andere job. Een worker haalt de job pas op nadat de afhankelijkheid is voltooid.

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

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

Batchdispatch

Dispatch meerdere jobs in één INSERT met meerdere rijen voor maximale doorvoer:

$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' ] ],
] );

Elk item kan optionele sleutels bevatten: queue, priority, delay, max_attempts.

Opnieuw proberen en back-off

Wanneer een job een uitzondering veroorzaakt, zet Queuety de job terug in pending met een vertraagde available_at totdat deze in totaal max_attempts keer is uitgevoerd (de eerste uitvoering telt mee, dus max_attempts( 3 ) betekent maximaal twee retries). De vertraging vóór elke retry wordt bepaald door de back-offstrategie, waarbij attempt het nummer is van de uitvoering die zojuist is mislukt:

Stel de globale standaard in wp-config.php in:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Of per handler in config():

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

Of per dispatch:

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

Begraven taken

Na alle retrypogingen wordt een job gemarkeerd als begraven. Begraven jobs blijven in de database staan voor inspectie, maar worden niet verwerkt. U kunt ze handmatig opnieuw proberen. Bij een handmatige retry wordt de job weer pending zonder de pogingteller te resetten; een opnieuw geprobeerd begraven job krijgt dus nog één uitvoering en wordt opnieuw begraven als die uitvoering mislukt:

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

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

Of via CLI:

wp queuety retry 42
wp queuety retry-buried

Time-out

Queuety stelt een pcntl_alarm() in voor de maximale uitvoeringstijd en moet de poging laten mislukken met een TimeoutException als die tijd wordt overschreden. Die time-out wordt momenteel niet afgedwongen (#10): de worker schakelt asynchrone signalen niet in, dus een langlopende handler wordt volledig uitgevoerd, terwijl een handler die vastzit in sleep() wordt afgebroken wanneer het alarm afgaat en vervolgens als voltooid wordt gemarkeerd. Configureer de time-out globaal:

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

Jobs die langer dan QUEUETY_STALE_TIMEOUT seconden vastzitten in de status processing, worden automatisch hersteld:

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

Herstel verouderde taken handmatig:

wp queuety recover

Queues

Jobs worden naar benoemde wachtrijen gedispatcht. Elke worker verwerkt één wachtrij of een geordende lijst met wachtrijen (zie Workers):

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

wp queuety work --queue=emails

De standaardwachtrij is default. U kunt voor elke wachtrij aparte workerprocessen uitvoeren om workloads te isoleren.

Snelheidsbeperking

Beperk hoe vaak een handler binnen een tijdvenster kan worden uitgevoerd:

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

Zie Snelheidsbeperking voor meer informatie.

Dispatchbare jobklassen

Queuety v0.6.0 introduceert een op klassen gebaseerd jobAPI-model. In plaats van de handlernaam, payload en handlerklasse gescheiden te houden, definieert u één klasse die alles bevat: de payload (als getypeerde openbare properties), de uitvoeringslogica en optionele middleware.

De Contracts\Job-interface

use Queuety\Contracts\Job;

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

Een Job-klasse bevat de payload in openbare properties en voert de eigen logica uit in handle(). Er is geen aparte config()-methode of handlerregistratie nodig.

De Dispatchable-trait

Voeg de Dispatchable-trait toe om een statische methode dispatch() aan uw jobklasse toe te voegen:

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 );
    }
}

Dispatch deze door constructorargumenten mee te geven:

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

De Dispatchable-trait roept intern Queuety::dispatch() aan, zodat alle opties van PendingJob (wachtrij, prioriteit, vertraging, max_attempts, uniek, after) precies hetzelfde werken.

Hoe serialisatie werkt

Wanneer u een instantie van Job dispatcht, haalt Queuety's JobSerializer de volledig gekwalificeerde klassenaam op als handler en alle openbare properties als payload. Aan de workerzijde reconstrueert de serializer de instantie door payloadsleutels te koppelen aan parameternamen van de constructor. Backed enums worden geserialiseerd naar hun onderliggende waarde en automatisch hersteld.

Taakeigenschappen

Jobklassen kunnen openbare properties declareren om retry- en time-outgedrag te bepalen. De worker leest deze via reflection uit en past ze automatisch toe.

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...
    }
}

De array $backoff biedt oplopende vertragingen tussen retries. Zo betekent [10, 30, 60, 120] dat de eerste retry 10 seconden wacht, de tweede 30 seconden, de derde 60 seconden en elke volgende retry 120 seconden wacht.

De resourcegerelateerde properties zijn nuttig wanneer jobklassen verschillende kostenprofielen vertegenwoordigen. Een snelle cachevernieuwing en een grote provider-aanroep kunnen allebei "één job" zijn, maar hoeven niet op dezelfde manier mee te tellen wanneer een workflowbudget of toelatingsbeslissing van een worker het proces binnen veilige grenzen probeert te houden.

Zie voor het volledige model, inclusief workflowbudgetten en toelating aan de workerzijde, Resourcebeheer.

Hook bij mislukking

Wanneer een job na alle retrypogingen wordt begraven, roept Queuety een methode failed() aan op de jobinstantie, als die is gedefinieerd. Gebruik deze voor opruimen, waarschuwingen of logging.

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()}" );
    }
}

De methode failed() ontvangt de uitzondering die de uiteindelijke mislukking heeft veroorzaakt. De jobinstantie wordt gedeserialiseerd met de oorspronkelijke payload, zodat alle openbare properties beschikbaar zijn.

Voorwaardelijk dispatchen

De Dispatchable-trait biedt dispatch_if() en dispatch_unless() voor voorwaardelijk dispatchen. Beide accepteren een booleaanse voorwaarde als eerste argument, gevolgd door de constructorargumenten van de job.

// 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() dispatcht wanneer de voorwaarde true is en retourneert de PendingJob. De methode retourneert null wanneer de voorwaarde false is.

dispatch_unless() dispatcht wanneer de voorwaarde false is en retourneert de PendingJob. De methode retourneert null wanneer de voorwaarde true is.

Beide methoden ondersteunen het koppelen van opties PendingJob wanneer ze dispatchen:

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

Synchroon dispatchen

Gebruik Queuety::dispatch_sync() om een job direct in het huidige proces uit te voeren zonder deze aan de wachtrij toe te voegen. Dit is handig voor testen en voor jobs die direct moeten worden uitgevoerd.

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' ] );

De methode handle() van de job wordt rechtstreeks aangeroepen. Er wordt geen databaserecord aangemaakt, er is geen worker bij betrokken en middleware wordt niet toegepast.

Jobs koppelen

Bij het koppelen van jobs wordt een reeks jobs na elkaar uitgevoerd. Elke job in de keten is afhankelijk van de succesvolle voltooiing van de vorige. Als een job mislukt, blijven de daaropvolgende jobs in de keten in de wachtrij staan.

Queuety::chain() gebruiken

use Queuety\Queuety;

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

Queuety::chain() retourneert een ChainBuilder met fluent-methoden om de wachtrij en een foutafhandelaar in te stellen. De methode dispatch() maakt alle jobs aan met sequentiële depends_on-relaties en retourneert de ID van de eerste job.

with_chain() gebruiken

De Dispatchable-trait biedt een methode with_chain(). Deze is momenteel gelijk aan Queuety::chain( $jobs ): de klasse waarop deze wordt aangeroepen, wordt niet aan de keten toegevoegd (#13), dus vermeld ook de eerste job in de array:

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

Foutafhandeling van ketens

Registreer een klasse voor foutafhandeling met catch(). De handlerklasse moet een methode handle() hebben. Deze wordt aangeroepen wanneer een job in de keten definitief wordt begraven. De worker ondersteunt handle(), handle( Job $job ) en handle( Job $job, \Throwable $exception ).

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

Middleware toevoegen

Jobklassen kunnen een methode middleware() definiëren om middleware terug te geven die de uitvoering omhult:

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 ),
        ];
    }
}

Zie Middleware voor meer informatie over ingebouwde middleware en het schrijven van aangepaste middleware.

Vergelijking met handlergebaseerde jobs

Beide benaderingen worden volledig ondersteund. Bestaande op handlers gebaseerde taken blijven zonder wijzigingen werken. U kunt beide stijlen in hetzelfde project combineren.

Taaklevenscyclus

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