Tâches

Les tâches sont l'unité de travail fondamentale dans Queuety. Déclenchez une tâche avec le nom d'un gestionnaire et des données utiles ; un worker la récupère et l'exécute en arrière-plan.

Déclencher des tâches

use Queuety\Queuety;

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

dispatch() renvoie un générateur PendingJob. Vous pouvez enchaîner des options avant son déclenchement automatique :

use Queuety\Enums\Priority;

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

Pour obtenir immédiatement l'ID de la tâche, appelez ->id() :

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

Écrire des gestionnaires

Un gestionnaire peut être n'importe quelle classe qui implémente l'interface 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',
        ];
    }
}

La méthode handle() reçoit les données utiles de la tâche et effectue le traitement. La méthode config() renvoie des valeurs par défaut facultatives :

Ces clés de ressources font partie du modèle plus général de gestion des ressources de Queuety, où les métadonnées des tâches, les budgets de workflow et l'admission des workers fonctionnent de concert.

Enregistrez le gestionnaire :

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

Ou utilisez les attributs PHP pour l’enregistrement automatique.

Priorité

Les tâches ont quatre niveaux de priorité. Les tâches les plus prioritaires sont prises en charge en premier par les workers.

use Queuety\Enums\Priority;

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

Délai

Retardez une tâche pour qu’elle ne soit disponible qu’après le nombre de secondes indiqué :

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

La tâche est insérée immédiatement, mais les workers ne la prendront en charge qu’une fois le délai écoulé.

Tâches uniques

Empêchez les envois en double pour un même gestionnaire et une même charge utile. Si une tâche en attente ou en cours de traitement existe déjà avec le même gestionnaire et la même charge utile, son ID est renvoyé au lieu d’en créer une nouvelle.

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

Dépendances entre tâches

Une tâche peut dépendre d’une autre. Un worker ne la prendra en charge qu’une fois la dépendance terminée.

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

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

Envoi par lot

Envoyez plusieurs tâches en une seule instruction INSERT à plusieurs lignes pour un débit maximal :

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

Chaque élément peut inclure les clés facultatives suivantes : queue, priority, delay, max_attempts.

Nouvelle tentative et délai exponentiel

Lorsqu’une tâche lève une exception, Queuety la remet dans pending avec un available_at différé jusqu’à ce qu’elle ait été exécutée max_attempts fois au total (la première exécution compte, donc max_attempts( 3 ) signifie jusqu’à deux nouvelles tentatives). Le délai avant chaque nouvelle tentative est déterminé par la stratégie de temporisation, où attempt désigne le numéro de l’exécution qui vient d’échouer :

Définissez la valeur par défaut globale dans wp-config.php :

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Ou définissez-la pour chaque gestionnaire dans config() :

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

Ou pour chaque envoi :

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

Tâches en attente définitive

Après avoir épuisé toutes les tentatives, une tâche est marquée comme enterrée. Les tâches enterrées restent dans la base de données pour inspection, mais ne sont pas traitées. Vous pouvez les relancer manuellement. Une relance manuelle remet la tâche dans pending sans réinitialiser son nombre de tentatives. Une tâche enterrée relancée bénéficie donc d’une exécution supplémentaire et sera de nouveau enterrée si cette exécution échoue :

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

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

Ou via CLI :

wp queuety retry 42
wp queuety retry-buried

Délai d'attente

Queuety déclenche une pcntl_alarm() pour la durée maximale d’exécution et est censé faire échouer la tentative avec une TimeoutException si cette durée est dépassée. Ce délai n’est actuellement pas appliqué (#10) : le worker n’active pas les signaux asynchrones. Un gestionnaire de longue durée s’exécute donc jusqu’au bout, et un gestionnaire bloqué dans sleep() est interrompu au déclenchement de l’alarme, puis marqué comme terminé. Configurez le délai d’expiration globalement :

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

Les tâches dont le statut processing persiste plus de QUEUETY_STALE_TIMEOUT secondes sont automatiquement récupérées :

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

Récupérez manuellement les tâches obsolètes :

wp queuety recover

Files d’attente

Les tâches sont envoyées vers des files d’attente nommées. Chaque worker traite une file ou une liste de files dans un ordre défini (voir Workers) :

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

wp queuety work --queue=emails

La file d’attente par défaut est default. Vous pouvez exécuter des processus worker distincts pour chaque file afin d’isoler les charges de travail.

Limitation du débit

Limitez la fréquence à laquelle un gestionnaire peut s’exécuter sur une période donnée :

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

Consultez Limitation du débit pour en savoir plus.

Classes de tâches envoyables

Queuety v0.6.0 introduit un API de tâches fondé sur des classes. Au lieu de séparer le nom du gestionnaire, la charge utile et la classe du gestionnaire, vous définissez une seule classe qui regroupe tout : la charge utile (sous forme de propriétés publiques typées), la logique d’exécution et les middlewares facultatifs.

L’interface Contracts\Job

use Queuety\Contracts\Job;

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

Une classe Job stocke sa charge utile dans des propriétés publiques et exécute sa propre logique dans handle(). Aucune méthode config() distincte ni inscription d’un gestionnaire n’est nécessaire.

Le trait Dispatchable

Ajoutez le trait Dispatchable pour obtenir une méthode statique dispatch() dans votre classe de tâche :

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

Envoyez-la en passant les arguments du constructeur :

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

Le trait Dispatchable appelle Queuety::dispatch() en interne. Toutes les options PendingJob (queue, priority, delay, max_attempts, unique, after) fonctionnent donc de la même manière.

Fonctionnement de la sérialisation

Lorsque vous envoyez une instance de Job, le JobSerializer de Queuety extrait le nom de classe qualifié complet pour en faire le gestionnaire, ainsi que toutes les propriétés publiques pour constituer la charge utile. Côté worker, le sérialiseur reconstruit l’instance en faisant correspondre les clés de la charge utile aux noms des paramètres du constructeur. Les énumérations adossées sont sérialisées sous leur valeur sous-jacente et restaurées automatiquement.

Propriétés des tâches

Les classes de tâches peuvent déclarer des propriétés publiques pour contrôler le comportement des nouvelles tentatives et des délais d’expiration. Le worker les lit par réflexion et les applique automatiquement.

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

Le tableau $backoff définit des délais croissants entre les nouvelles tentatives. Par exemple, [10, 30, 60, 120] signifie que la première nouvelle tentative attend 10 secondes, la deuxième 30 secondes, la troisième 60 secondes, et que les suivantes attendent 120 secondes.

Les propriétés liées aux ressources sont utiles lorsque les classes de tâches représentent différents niveaux de coût. Un rafraîchissement rapide du cache et un appel coûteux à un fournisseur peuvent tous deux constituer « une tâche », mais ils ne devraient pas forcément être comptabilisés de la même manière lorsqu’un budget de workflow ou une décision d’admission des workers vise à maintenir le processus dans des limites sûres.

Pour découvrir le modèle complet, y compris les budgets de workflow et l’admission côté worker, consultez Gestion des ressources.

Hook d’échec

Lorsqu’une tâche est enterrée après avoir épuisé toutes ses tentatives, Queuety appelle une méthode failed() sur l’instance de la tâche, si elle est définie. Utilisez-la pour le nettoyage, les alertes ou la journalisation.

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

La méthode failed() reçoit l’exception à l’origine de l’échec final. L’instance de la tâche est désérialisée avec sa charge utile d’origine ; toutes ses propriétés publiques sont donc disponibles.

Envoi conditionnel

Le trait Dispatchable fournit dispatch_if() et dispatch_unless() pour effectuer des envois conditionnels. Les deux méthodes acceptent une condition booléenne comme premier argument, suivie des arguments du constructeur de la tâche.

// 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() envoie la tâche si la condition est true et renvoie le PendingJob. Elle renvoie null si la condition est false.

dispatch_unless() envoie la tâche si la condition est false et renvoie le PendingJob. Elle renvoie null si la condition est true.

Les deux méthodes acceptent les options PendingJob chaînables lors de l’envoi :

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

Envoi synchrone

Utilisez Queuety::dispatch_sync() pour exécuter immédiatement une tâche dans le processus actuel, sans l’ajouter à la file d’attente. C’est utile pour les tests et les tâches qui doivent s’exécuter en ligne.

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

La méthode handle() de la tâche est appelée directement. Aucune ligne n’est créée dans la base de données, aucun worker n’intervient et aucun intergiciel n’est appliqué.

Enchaînement de tâches

L’enchaînement de tâches exécute une série de tâches l’une après l’autre. Chaque tâche de la chaîne dépend de la réussite de la précédente. Si une tâche échoue, les suivantes restent en attente.

Utiliser 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() renvoie un ChainBuilder avec des méthodes chaînables pour définir la file d’attente et un gestionnaire d’échec. La méthode dispatch() crée toutes les tâches avec des relations depends_on séquentielles et renvoie l’ID de la première tâche.

Utiliser with_chain()

Le trait Dispatchable fournit une méthode with_chain(). Elle est actuellement équivalente à Queuety::chain( $jobs ) : la classe sur laquelle elle est appelée n’est pas ajoutée à la chaîne (#13). Il faut donc aussi indiquer la première tâche dans le tableau :

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

Gestion des échecs de chaîne

Enregistrez une classe de gestion des échecs avec catch(). Cette classe doit disposer d’une méthode handle(). Elle est appelée lorsqu’une tâche de la chaîne est définitivement enterrée. Le worker prend en charge handle(), handle( Job $job ) et handle( Job $job, \Throwable $exception ).

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

Ajouter des middlewares

Les classes de tâches peuvent définir une méthode middleware() qui renvoie les middlewares chargés d’encapsuler l’exécution :

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

Consultez Middlewares pour en savoir plus sur les middlewares intégrés et la création de middlewares personnalisés.

Comparaison avec les tâches fondées sur des gestionnaires

Les deux approches sont entièrement prises en charge. Les tâches existantes fondées sur des gestionnaires continuent de fonctionner sans modification. Vous pouvez combiner les deux styles dans un même projet.

Cycle de vie d’une tâche

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