Jobb

Jobb är den grundläggande arbetsenheten i Queuety. Skicka iväg ett jobb med ett hanterarnamn och en datanyttolast, så hämtar en arbetare det och kör det i bakgrunden.

Skicka iväg jobb

use Queuety\Queuety;

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

dispatch() returnerar en byggare av typen PendingJob. Du kan kedja alternativ innan den skickar iväg jobbet automatiskt:

use Queuety\Enums\Priority;

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

Anropa ->id() för att få jobbets ID direkt:

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

Skriva hanterare

En hanterare kan vara vilken klass som helst som implementerar gränssnittet Queuety\Handler:

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

Metoden handle() tar emot jobbets datanyttolast och utför arbetet. Metoden config() returnerar valfria standardvärden:

Dessa resursnycklar ingår i den bredare modellen för QueuetyResurshantering, där jobbmetadata, arbetsflödesbudgetar och arbetarantagning samverkar.

Registrera hanteraren:

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

Eller använd PHP-attribut för automatisk registrering.

Prioritet

Jobb har fyra prioritetsnivåer. Arbetare hämtar jobb med högre prioritet först.

use Queuety\Enums\Priority;

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

Fördröjning

Fördröj ett jobb så att det blir tillgängligt först efter det angivna antalet sekunder:

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

Jobbet infogas omedelbart, men arbetare hämtar det inte förrän fördröjningen har löpt ut.

Unika jobb

Förhindra duplicerade utskick för samma hanterare och nyttolast. Om det redan finns ett väntande eller pågående jobb med samma hanterare och nyttolast returneras det befintliga jobb-ID:t i stället för att skapa ett nytt.

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

Jobbberoenden

Ett jobb kan vara beroende av ett annat jobb. En arbetare hämtar det inte förrän beroendet har slutförts.

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

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

Grupputskick

Skicka ut flera jobb i en enda INSERT med flera rader för högsta möjliga genomströmning:

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

Varje objekt kan innehålla valfria nycklar: queue, priority, delay, max_attempts.

Återförsök och exponentiell väntetid

När ett jobb kastar ett undantag flyttar Queuety tillbaka det till pending med ett fördröjt available_at tills det har körts totalt max_attempts gånger (den första körningen räknas, så max_attempts( 3 ) innebär upp till två återförsök). Väntetiden före varje återförsök bestäms av strategin för exponentiell väntetid, där attempt är numret på körningen som nyss misslyckades:

Ange det globala standardvärdet i wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Eller per hanterare i config():

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

Eller per utskick:

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

Jobb i karantän

När alla återförsök har förbrukats markeras ett jobb som begravt. Begravda jobb ligger kvar i databasen för inspektion men behandlas inte. Du kan försöka igen manuellt. Ett manuellt återförsök gör jobbet pending igen utan att återställa antalet försök, så ett begravt jobb som försöks igen får en körning till och begravs på nytt om den körningen misslyckas:

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

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

Eller via CLI:

wp queuety retry 42
wp queuety retry-buried

Tidsgräns

Queuety ställer in ett pcntl_alarm() för den maximala körningstiden och är avsett att få försöket att misslyckas med ett TimeoutException när tiden överskrids. Den tidsgränsen tillämpas för närvarande inte (#10): arbetaren aktiverar inte asynkrona signaler, så en långkörande hanterare kör klart, och en hanterare som blockeras i sleep() avbryts när larmet utlöses och markeras sedan som slutförd. Konfigurera tidsgränsen globalt:

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

Jobb som har fastnat i statusen processing i mer än QUEUETY_STALE_TIMEOUT sekunder återställs automatiskt:

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

Återställ föråldrade jobb manuellt:

wp queuety recover

Köer

Jobb skickas till namngivna köer. Varje arbetare behandlar en kö eller en ordnad lista med köer (se Arbetare):

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

wp queuety work --queue=emails

Standardkön är default. Du kan köra separata arbetarprocesser för varje kö för att isolera arbetsbelastningar.

Hastighetsbegränsning

Begränsa hur ofta en hanterare kan köras under en tidsperiod:

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

Läs Hastighetsbegränsning för mer information.

Utskickbara jobbklasser

Queuety v0.6.0 introducerar jobb API baserade på klasser. I stället för att skilja på hanterarens namn, nyttolast och hanterarklass definierar du en enda klass som kapslar in allt: nyttolasten (som typade offentliga egenskaper), exekveringslogiken och valfri mellanprogramvara.

Gränssnittet Contracts\Job

use Queuety\Contracts\Job;

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

En klass av typen Job har sin nyttolast som offentliga egenskaper och kör sin egen logik i handle(). Ingen separat metod config() eller registrering av hanterare krävs.

Egenskapen Dispatchable

Lägg till egenskapen Dispatchable för att få en statisk metod dispatch() i din jobbklass:

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

Skicka ut jobbet genom att ange argument till konstruktorn:

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

Egenskapen Dispatchable anropar Queuety::dispatch() internt, så alla alternativ i PendingJob (kö, prioritet, fördröjning, max_attempts, unique, after) fungerar på samma sätt.

Så fungerar serialiseringen

När du skickar ut en instans av Job hämtar Queuetys JobSerializer det fullständigt kvalificerade klassnamnet som hanterare och alla offentliga egenskaper som nyttolast. På arbetarsidan återskapar serialiseraren instansen genom att matcha nyttolastens nycklar mot konstruktorns parameternamn. Backade enum-värden serialiseras till sina underliggande värden och återställs automatiskt.

Jobbegenskaper

Jobbklasser kan deklarera offentliga egenskaper för att styra återförsök och tidsgränser. Arbetaren läser dessa via reflection och tillämpar dem automatiskt.

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

Arrayen $backoff anger ökande väntetider mellan återförsök. Till exempel innebär [10, 30, 60, 120] att det första återförsöket väntar 10 sekunder, det andra 30 sekunder, det tredje 60 sekunder och eventuella ytterligare återförsök 120 sekunder.

De resursrelaterade egenskaperna är användbara när jobbklasser representerar olika kostnadsprofiler. En snabb cacheuppdatering och ett stort anrop till en leverantör kan båda vara "ett jobb", men de bör inte nödvändigtvis räknas likadant när en arbetsflödesbudget eller ett beslut om arbetarinsläpp ska hålla processen inom en säker gräns.

Se Resurshantering för den fullständiga modellen, inklusive arbetsflödesbudgetar och arbetarinsläpp.

Felhanteringskrok

När ett jobb begravs efter att alla återförsök har förbrukats anropar Queuety en metod failed() på jobbinstansen, om en sådan har definierats. Använd den för rensning, aviseringar eller loggning.

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

Metoden failed() tar emot undantaget som orsakade det slutliga felet. Jobbinstansen deserialiseras med sin ursprungliga nyttolast, så alla offentliga egenskaper är tillgängliga.

Villkorat utskick

Egenskapen Dispatchable tillhandahåller dispatch_if() och dispatch_unless() för villkorliga utskick. Båda tar emot ett booleskt villkor som första argument, följt av jobbets konstruktionsargument.

// 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() skickar ut jobbet när villkoret är true och returnerar PendingJob. Den returnerar null när villkoret är false.

dispatch_unless() skickar ut jobbet när villkoret är false och returnerar PendingJob. Den returnerar null när villkoret är true.

Båda metoderna har stöd för kedjade alternativ i PendingJob när de skickar ut jobb:

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

Synkront utskick

Använd Queuety::dispatch_sync() för att köra ett jobb omedelbart i den aktuella processen utan att lägga till det i kön. Det är användbart vid testning och för jobb som måste köras direkt.

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

Jobbets metod handle() anropas direkt. Ingen databasrad skapas, ingen arbetare medverkar och ingen mellanprogramvara tillämpas.

Jobbkedjor

Jobbkedjor kör en sekvens av jobb i tur och ordning. Varje jobb i kedjan är beroende av att föregående jobb slutförs utan fel. Om ett jobb misslyckas förblir efterföljande jobb i kedjan väntande.

Använda 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() returnerar en ChainBuilder med kedjbara metoder för att ange kön och en felhanterare. Metoden dispatch() skapar alla jobb med sekventiella relationer av typen depends_on och returnerar det första jobbets ID.

Använda with_chain()

Egenskapen Dispatchable tillhandahåller en metod with_chain(). Den motsvarar för närvarande Queuety::chain( $jobs ): klassen den anropas på läggs inte till i kedjan (#13), så ta även med det första jobbet i arrayen:

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

Hantering av kedjefel

Registrera en klass för felhantering med catch(). Klassen för felhantering måste ha en metod handle(). Den anropas när ett jobb i kedjan begravs permanent. Arbetaren har stöd för handle(), handle( Job $job ) och handle( Job $job, \Throwable $exception ).

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

Lägga till mellanprogramvara

Jobbklasser kan definiera en metod middleware() som returnerar mellanprogramvara för att omsluta körningen:

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

Se Mellanprogramvara för mer information om inbyggd mellanprogramvara och hur du skriver egen.

Jämförelse med hanterarbaserade jobb

Båda metoderna stöds fullt ut. Befintliga hanterarbaserade jobb fortsätter att fungera utan ändringar. Du kan blanda båda stilarna i samma projekt.

Jobbets livscykel

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