Jobok

A jobok a Queuety alapvető munkavégzési egységei. Indítson egy jobot kezelőnévvel és payloaddal; egy worker felveszi, majd a háttérben végrehajtja.

Jobok indítása

use Queuety\Queuety;

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

A dispatch() egy PendingJob buildert ad vissza. Az automatikus indítás előtt láncolhat hozzá beállításokat:

use Queuety\Enums\Priority;

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

A job ID-jának azonnali lekéréséhez hívja meg a ->id() metódust:

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

Kezelők írása

Kezelő lehet bármely osztály, amely megvalósítja a Queuety\Handler interfészt:

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

A handle() metódus megkapja a job payloadját, és elvégzi a munkát. A config() metódus opcionális alapértékeket ad vissza:

Ezek az erőforráskulcsok a Queuety tágabb Erőforrás-kezelési modelljének részei, amelyben a jobok metaadatai, a munkafolyamat-keretek és a workerfelvétel együtt működnek.

Regisztrálja a kezelőt:

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

Vagy használja az PHP attribútumokat az automatikus regisztrációhoz.

Prioritás

A joboknak négy prioritási szintjük van. A workerek először a magasabb prioritású jobokat veszik fel.

use Queuety\Enums\Priority;

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

Késleltetés

Késleltesse a jobot, hogy csak a megadott számú másodperc elteltével váljon elérhetővé:

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

A job azonnal bekerül, de a workerek csak a késleltetés letelte után veszik fel.

Egyedi jobok

Akadályozza meg ugyanazon kezelő és payload többszöri indítását. Ha már létezik azonos kezelőjű és payloadú, függőben lévő vagy feldolgozás alatt álló job, az új létrehozása helyett annak meglévő ID-ja kerül visszaadásra.

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

Jobfüggőségek

Egy job függhet egy másik jobtól. A worker csak a függőség befejezése után veszi fel.

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

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

Kötegelt indítás

Indítson több jobot egyetlen, több soros INSERT utasítással a maximális áteresztőképességért:

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

Minden elem opcionálisan tartalmazhatja a következő kulcsokat: queue, priority, delay, max_attempts.

Újrapróbálkozás és késleltetés

Ha egy job kivételt dob, a Queuety visszateszi azt a pending sorba, és késleltetett available_at-t állít be, amíg összesen max_attempts alkalommal le nem futott (az első futás is számít, így a max_attempts( 3 ) legfeljebb két újrapróbálkozást jelent). Az egyes próbálkozások előtti késleltetést a késleltetési stratégia határozza meg; a attempt az éppen sikertelen futás sorszáma:

Állítsa be a globális alapértéket itt: wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Vagy kezelőnként itt: config():

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

Vagy indításonként:

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

Eltemetett feladatok

Az újrapróbálkozási kísérletek kimerítése után a job parkol. A parkolt jobok vizsgálat céljából az adatbázisban maradnak, de nem kerülnek feldolgozásra. Manuálisan újrapróbálhatja őket. Ilyenkor a job ismét pending lesz, próbálkozásszámlálója azonban nem áll vissza, ezért a parkolt job még egyszer lefut, és újabb hiba esetén ismét parkolásra kerül:

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

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

Vagy a CLI használatával:

wp queuety retry 42
wp queuety retry-buried

Időkorlát

A(z) Queuety élesít egy pcntl_alarm()-t a maximális végrehajtási időre, és a próbálkozás meghiúsítására szolgál egy TimeoutException-val, ha az időkorlát lejár. Ez az időkorlát jelenleg nincs érvényesítve (#10): a worker nem engedélyezi az aszinkron jelzéseket, ezért a hosszan futó kezelő a végéig fut, az sleep()-ban blokkolt kezelő pedig megszakad, amikor megszólal a riasztás, majd befejezettként lesz megjelölve. Az időkorlát globális beállítása:

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

A processing állapotban a megadott QUEUETY_STALE_TIMEOUT másodpercnél tovább elakadt jobok automatikusan helyreállításra kerülnek:

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

Az elakadt feladatok kézi helyreállítása:

wp queuety recover

Várólisták

A jobok névvel ellátott sorokba kerülnek. Minden worker egy sort vagy sorok rendezett listáját dolgozza fel (lásd: Workerek):

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

wp queuety work --queue=emails

Az alapértelmezett sor a default. A munkaterhelések elkülönítéséhez külön workerfolyamatokat futtathat az egyes sorokhoz.

Sebességkorlátozás

Korlátozza, hogy egy kezelő milyen gyakran hajtható végre egy időablakon belül:

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

Részletekért lásd: Sebességkorlátozás.

Indítható jobosztályok

A Queuety v0.6.0 osztályalapú job API rendszert vezet be. A kezelő nevének, a payloadnak és a kezelőosztálynak a külön tárolása helyett egyetlen, mindent összefogó osztályt definiál: ez tartalmazza a payloadot (típusos publikus tulajdonságokként), a végrehajtási logikát és az opcionális middleware-t.

A Contracts\Job interfész

use Queuety\Contracts\Job;

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

Egy Job osztály publikus tulajdonságokként tárolja a payloadját, és a saját logikáját a handle() metódusban hajtja végre. Nincs szükség külön config() metódusra vagy kezelőregisztrációra.

A Dispatchable trait

Adja hozzá a Dispatchable traitet, hogy statikus dispatch() metódus kerüljön a jobosztályba:

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

Indítsa el konstruktorargumentumok megadásával:

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

A Dispatchable trait belsőleg meghívja a Queuety::dispatch() metódust, ezért minden PendingJob beállítás (queue, priority, delay, max_attempts, unique, after) ugyanúgy működik.

A szerializálás működése

Egy Job példány indításakor a Queuety JobSerializer metódusa a teljesen minősített osztálynevet használja kezelőként, az összes publikus tulajdonságot pedig payloadként. A worker oldalán a szerializáló a payloadkulcsokat a konstruktor paraméterneveihez rendelve állítja vissza a példányt. A backed enumok az alapul szolgáló értékükre szerializálódnak, majd automatikusan visszaállításra kerülnek.

A feladat tulajdonságai

A jobosztályok deklarálhatnak publikus tulajdonságokat az újrapróbálkozások és az időtúllépés viselkedésének szabályozására. A worker reflexióval olvassa ki ezeket, és automatikusan alkalmazza őket.

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

A $backoff tömb növekvő újrapróbálkozási késleltetéseket ad meg. Például a [10, 30, 60, 120] azt jelenti, hogy az első újrapróbálkozás 10 másodpercet, a második 30 másodpercet, a harmadik 60 másodpercet vár, a további próbálkozások pedig 120 másodpercet.

Az erőforrással kapcsolatos tulajdonságok akkor hasznosak, ha a jobosztályok eltérő költségprofilokat képviselnek. Egy gyors gyorsítótár-frissítés és egy nagy szolgáltatói hívás egyaránt lehet „egy job”, de nem feltétlenül kell azonos súllyal számítaniuk, amikor a munkafolyamat-keret vagy a workerfelvétel célja a folyamat biztonságos keretek között tartása.

A teljes modellért, beleértve a munkafolyamat-kereteket és a workeroldali felvételt, lásd: Erőforrás-kezelés.

Sikertelen hook

Ha a job az összes újrapróbálkozási kísérlet kimerítése után a parkolóba kerül, a Queuety meghívja a jobpéldány failed() metódusát, ha az definiálva van. Használja takarításhoz, riasztáshoz vagy naplózáshoz.

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

A failed() metódus megkapja a végleges hibát okozó kivételt. A jobpéldány az eredeti payloaddal szerializálódik vissza, így minden publikus tulajdonság elérhető.

Feltételes indítás

A Dispatchable trait a feltételes indításhoz a dispatch_if() és a dispatch_unless() metódust biztosítja. Mindkettő első argumentuma egy logikai feltétel, amelyet a job konstruktorának argumentumai követnek.

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

A dispatch_if() akkor indítja el a jobot, ha a feltétel true, és a PendingJob értéket adja vissza. Ha a feltétel false, akkor null értéket ad vissza.

A dispatch_unless() akkor indítja el a jobot, ha a feltétel false, és a PendingJob értéket adja vissza. Ha a feltétel true, akkor null értéket ad vissza.

Mindkét metódus támogatja a PendingJob beállítások láncolását indításkor:

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

Szinkron indítás

A Queuety::dispatch_sync() használatával a jobot azonnal, a jelenlegi folyamatban hajthatja végre anélkül, hogy hozzáadná a sorhoz. Ez teszteléshez és olyan jobokhoz hasznos, amelyeket közvetlenül kell futtatni.

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

A feladat handle() metódusa közvetlenül hívódik meg. Nem jön létre adatbázissor, nincs szükség munkásra, és a köztes szoftverek sem futnak le.

Jobok láncolása

A jobok láncolása egymás után hajt végre egy sorozatot. A lánc minden jobja az előző job sikeres befejezésétől függ. Ha bármelyik job meghiúsul, a lánc további jobjai függőben maradnak.

A Queuety::chain() használata

use Queuety\Queuety;

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

A Queuety::chain() egy ChainBuilder objektumot ad vissza, amely folyékony metódusokat biztosít a sor beállításához és egy hibakezelő megadásához. A dispatch() metódus minden jobot egymást követő depends_on kapcsolatokkal hoz létre, és az első job ID-ját adja vissza.

A with_chain() használata

A Dispatchable trait egy with_chain() metódust biztosít. Ez jelenleg egyenértékű a Queuety::chain( $jobs ) metódussal: a hívás címzettje nem kerül be a láncba (#13), ezért az első jobot is sorolja fel a tömbben:

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

Lánchibák kezelése

Regisztráljon hibakezelő osztályt a catch() használatával. Az osztálynak rendelkeznie kell handle() metódussal. Ez akkor hívódik meg, amikor a lánc bármelyik jobja véglegesen a parkolóba kerül. A worker a következőket támogatja: handle(), handle( Job $job ) és handle( Job $job, \Throwable $exception ).

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

Middleware hozzáadása

A jobosztályok definiálhatnak egy middleware() metódust, amely a végrehajtást körülvevő middleware-t ad vissza:

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

A beépített middleware-ekről és az egyéni middleware írásáról lásd: Middleware.

Összehasonlítás a kezelőalapú jobokkal

Mindkét megközelítés teljes mértékben támogatott. A meglévő, kezelőalapú feladatok változtatás nélkül tovább működnek. A két stílust ugyanabban a projektben is vegyítheti.

A feladat életciklusa

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