Jobs

Jobs er den grundlæggende arbejdsenhed i Queuety. Send et job med et handlernavn og en payload, hvorefter en worker henter og udfører det i baggrunden.

Afsendelse af jobs

use Queuety\Queuety;

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

dispatch() returnerer en builder af typen PendingJob. Du kan kæde valgmuligheder på, før den automatisk sender jobbet:

use Queuety\Enums\Priority;

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

Kald ->id() for at få job-id'et med det samme:

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

Skrivning af handlere

En handler er en vilkårlig klasse, der implementerer Queuety\Handler-grænsefladen:

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() modtager jobpayloaden og udfører arbejdet. Metoden config() returnerer valgfrie standardværdier:

Disse ressourcenøgler indgår i Queuetys bredere model for Resource Management, hvor jobmetadata, workflowbudgetter og worker-adgangskontrol fungerer sammen.

Registrér handleren:

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

Eller brug attributterne PHP til automatisk registrering.

Prioritet

Jobs har fire prioritetsniveauer. Workers henter jobs med højere prioritet først.

use Queuety\Enums\Priority;

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

Forsinkelse

Forsink et job, så det først bliver tilgængeligt efter det angivne antal sekunder:

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

Jobbet indsættes med det samme, men workers henter det først, når forsinkelsen er udløbet.

Unikke jobs

Undgå gentagne afsendelser med samme handler og payload. Hvis der allerede findes et afventende eller igangværende job med samme handler og payload, returneres det eksisterende job-id i stedet for at oprette et nyt.

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

Jobafhængigheder

Et job kan afhænge af et andet job. En worker henter det først, når afhængigheden er fuldført.

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

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

Afsendelse af batches

Send flere jobs i én INSERT med flere rækker for at opnå maksimal gennemstrø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' ] ],
] );

Hvert element kan indeholde valgfrie nøgler: queue, priority, delay, max_attempts.

Nye forsøg og forsinkelse

Når et job udløser en undtagelse, flytter Queuety det tilbage til pending med en forsinket available_at, indtil det er kørt max_attempts gange i alt (første kørsel tæller med, så max_attempts( 3 ) betyder op til to nye forsøg). Forsinkelsen før hvert nyt forsøg bestemmes af strategien for forsinkelse, hvor attempt er nummeret på den kørsel, der netop mislykkedes:

Angiv den globale standard i wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Eller pr. handler i config():

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

Eller pr. afsendelse:

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

Parkerede jobs

Når alle forsøg på ny kørsel er brugt, markeres jobbet som lagt til side. Jobs, der er lagt til side, bliver i databasen til gennemsyn, men behandles ikke. Du kan prøve dem igen manuelt. Et manuelt nyt forsøg gør jobbet til pending igen uden at nulstille forsøgsantallet, så et job, der er lagt til side, får én ekstra kørsel og lægges til side igen, hvis den mislykkes:

// 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

Timeout

Queuety starter en pcntl_alarm() for den maksimale afviklingstid og skal få forsøget til at mislykkes med en TimeoutException, hvis tiden overskrides. Timeouten håndhæves i øjeblikket ikke (#10): workeren aktiverer ikke asynkrone signaler, så en langvarig handler kører færdig, mens en handler, der blokerer i sleep(), afbrydes, når alarmen udløses, og markeres derefter som fuldført. Konfigurér timeouten globalt:

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

Jobs, der sidder fast i status processing i længere tid end QUEUETY_STALE_TIMEOUT sekunder, gendannes automatisk:

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

Gendan forældede jobs manuelt:

wp queuety recover

Køer

Jobs sendes til navngivne køer. Hver worker behandler én kø eller en sorteret liste over køer (se Workers):

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

wp queuety work --queue=emails

Standardkøen er default. Du kan køre separate worker-processer for hver kø for at isolere arbejdsbelastninger.

Hastighedsbegrænsning

Begræns, hvor ofte en handler kan køre inden for et tidsvindue:

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

Se Rate Limiting for flere oplysninger.

Jobklasser, der kan sendes

Queuety v0.6.0 introducerer en klassebaseret job-API. I stedet for at adskille handlernavn, payload og handlerklasse definerer du én klasse, der samler det hele: payloaden (som typede offentlige egenskaber), afviklingslogikken og valgfri middleware.

Grænsefladen Contracts\Job

use Queuety\Contracts\Job;

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

En Job-klasse gemmer sin payload som offentlige egenskaber og udfører sin egen logik i handle(). Der kræves ingen separat metode config() eller handlerregistrering.

Traitet Dispatchable

Tilføj traitet Dispatchable for at få en statisk metode dispatch() i din jobklasse:

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

Send jobbet ved at angive konstruktørargumenter:

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

Traitet Dispatchable kalder Queuety::dispatch() internt, så alle PendingJob-valgmuligheder (queue, priority, delay, max_attempts, unique, after) fungerer på præcis samme måde.

Sådan fungerer serialisering

Når du sender en instans af Job, udtrækker Queuetys JobSerializer det fuldt kvalificerede klassenavn som handler og alle offentlige egenskaber som payload. På worker-siden genskaber serializeren instansen ved at matche payloadnøgler med konstruktørens parameternavne. Backed enums serialiseres til deres underliggende værdi og gendannes automatisk.

Jobegenskaber

Jobklasser kan erklære offentlige egenskaber, der styrer adfærd ved gentagne forsøg og timeouts. Workeren læser dem med reflection og anvender dem automatisk.

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

Arrayet $backoff angiver stigende forsinkelser mellem nye forsøg. For eksempel betyder [10, 30, 60, 120], at det første nye forsøg venter 10 sekunder, det andet 30 sekunder, det tredje 60 sekunder, og at eventuelle efterfølgende forsøg venter 120 sekunder.

Ressourceegenskaberne er nyttige, når jobklasser repræsenterer forskellige omkostningsprofiler. En hurtig cacheopdatering og et stort udbyderkald kan begge være »ét job«, men de bør ikke nødvendigvis tælle ens, når et workflowbudget eller en beslutning om worker-adgang skal holde processen inden for sikre rammer.

Se Resource Management for hele modellen, herunder workflowbudgetter og adgangskontrol på workersiden.

Hook ved fejl

Når et job lægges til side efter at have brugt alle forsøg på ny kørsel, kalder Queuety metoden failed() på jobinstansen, hvis den er defineret. Brug den til oprydning, alarmer eller logning.

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() modtager den undtagelse, der forårsagede den endelige fejl. Jobinstansen deserialiseres med sin oprindelige payload, så alle offentlige egenskaber er tilgængelige.

Betinget afsendelse

Traitet Dispatchable indeholder dispatch_if() og dispatch_unless() til betinget afsendelse. Begge tager en boolesk betingelse som første argument efterfulgt af jobbets konstruktørargumenter.

// 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() sender jobbet, når betingelsen er true, og returnerer PendingJob. Den returnerer null, når betingelsen er false.

dispatch_unless() sender jobbet, når betingelsen er false, og returnerer PendingJob. Den returnerer null, når betingelsen er true.

Begge metoder understøtter kædning af PendingJob-valgmuligheder ved afsendelse:

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

Synkron afsendelse

Brug Queuety::dispatch_sync() til at udføre et job straks i den aktuelle proces uden at føje det til køen. Det er nyttigt til test og for jobs, der skal køre direkte.

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 metode handle() kaldes direkte. Der oprettes ingen databaserække, ingen worker involveres, og middleware anvendes ikke.

Kædning af jobs

Jobkædning udfører en række jobs efter hinanden. Hvert job i kæden afhænger af, at det foregående job gennemføres korrekt. Hvis et job fejler, forbliver de efterfølgende jobs i kæden afventende.

Brug af 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() returnerer en ChainBuilder med flydende metoder til at angive køen og en fejlhåndtering. Metoden dispatch() opretter alle jobs med sekventielle depends_on-relationer og returnerer ID'et på det første job.

Brug af with_chain()

Trait'et Dispatchable stiller metoden with_chain() til rådighed. Den svarer i øjeblikket til Queuety::chain( $jobs ): Klassen, den kaldes på, føjes ikke til kæden (#13), så angiv også det første job i arrayet:

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

Håndtering af kædefejl

Registrér en klasse til fejlhåndtering med catch(). Handlerklassen skal have en handle()-metode. Den kaldes, når et job i kæden ender permanent i karantæne. Workeren understøtter handle(), handle( Job $job ) og handle( Job $job, \Throwable $exception ).

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

Tilføjelse af middleware

Jobklasser kan definere en middleware()-metode, der returnerer middleware til at omslutte kørslen:

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 Middleware for detaljer om indbygget middleware og om at skrive brugerdefineret middleware.

Sammenligning med handlerbaserede jobs

Begge tilgange understøttes fuldt ud. Eksisterende handlerbaserede jobs fungerer videre uden ændringer. Du kan blande begge stilarter i samme projekt.

Joblivscyklus

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