Εργασίες

Οι εργασίες είναι η βασική μονάδα εκτέλεσης στο Queuety. Αποστείλετε μια εργασία με όνομα χειριστή και φορτίο, και ένας worker θα την αναλάβει και θα την εκτελέσει στο παρασκήνιο.

Αποστολή εργασιών

use Queuety\Queuety;

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

Το dispatch() επιστρέφει ένα εργαλείο δημιουργίας PendingJob. Μπορείτε να συνδέσετε επιλογές πριν από την αυτόματη αποστολή του:

use Queuety\Enums\Priority;

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

Για να λάβετε αμέσως το αναγνωριστικό εργασίας, καλέστε το ->id():

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

Σύνταξη χειριστών

Χειριστής είναι κάθε κλάση που υλοποιεί τη διεπαφή 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',
        ];
    }
}

Η μέθοδος handle() λαμβάνει το φορτίο της εργασίας και εκτελεί την εργασία. Η μέθοδος config() επιστρέφει προαιρετικές προεπιλεγμένες τιμές:

Αυτά τα κλειδιά πόρων αποτελούν μέρος του ευρύτερου μοντέλου Διαχείρισης πόρων του Queuety, όπου συνεργάζονται τα μεταδεδομένα εργασιών, οι προϋπολογισμοί ροών εργασίας και η αποδοχή εργασιών από worker.

Καταχωρίστε τον χειριστή:

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

Ή χρησιμοποιήστε γνωρίσματα PHP attributes για αυτόματη καταχώριση.

Προτεραιότητα

Οι εργασίες έχουν τέσσερα επίπεδα προτεραιότητας. Οι εργαζόμενες διεργασίες αναλαμβάνουν πρώτα τις εργασίες με υψηλότερη προτεραιότητα.

use Queuety\Enums\Priority;

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

Καθυστέρηση

Καθυστερήστε μια εργασία ώστε να γίνει διαθέσιμη μόνο αφού περάσει ο καθορισμένος αριθμός δευτερολέπτων:

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

Η εργασία εισάγεται αμέσως, αλλά οι εργαζόμενες διεργασίες δεν θα την αναλάβουν μέχρι να παρέλθει η καθυστέρηση.

Μοναδικές εργασίες

Αποτρέψτε την αποστολή διπλότυπων εργασιών για τον ίδιο χειριστή και το ίδιο ωφέλιμο φορτίο. Αν υπάρχει ήδη εργασία σε αναμονή ή υπό επεξεργασία με τον ίδιο χειριστή και ωφέλιμο φορτίο, επιστρέφεται το αναγνωριστικό της υπάρχουσας εργασίας αντί να δημιουργηθεί νέα.

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

Εξαρτήσεις εργασιών

Μια εργασία μπορεί να εξαρτάται από μια άλλη. Δεν θα αναληφθεί από εργαζόμενη διεργασία μέχρι να ολοκληρωθεί η εξαρτώμενη εργασία.

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

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

Μαζική αποστολή

Στείλτε πολλές εργασίες με ένα INSERT πολλών γραμμών για μέγιστη απόδοση:

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

Κάθε στοιχείο μπορεί να περιλαμβάνει προαιρετικά κλειδιά: queue, priority, delay, max_attempts.

Επανάληψη και σταδιακή καθυστέρηση

Όταν μια εργασία προκαλεί εξαίρεση, το Queuety την επαναφέρει στο pending με καθυστερημένο available_at μέχρι να εκτελεστεί συνολικά max_attempts φορές (η πρώτη εκτέλεση μετράει, επομένως το max_attempts( 3 ) σημαίνει έως δύο επαναλήψεις). Η καθυστέρηση πριν από κάθε επανάληψη καθορίζεται από τη στρατηγική σταδιακής καθυστέρησης, όπου το attempt είναι ο αριθμός της εκτέλεσης που μόλις απέτυχε:

Ορίστε την καθολική προεπιλογή στο wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Ή ανά χειριστή στο config():

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

Ή ανά αποστολή:

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

Θαμμένες εργασίες

Αφού εξαντληθούν όλες οι προσπάθειες επανάληψης, μια εργασία χαρακτηρίζεται ως θαμμένη. Οι θαμμένες εργασίες παραμένουν στη βάση δεδομένων για επιθεώρηση, αλλά δεν υποβάλλονται σε επεξεργασία. Μπορείτε να τις επαναλάβετε χειροκίνητα. Η χειροκίνητη επανάληψη επαναφέρει την εργασία σε κατάσταση pending χωρίς να μηδενίσει τον αριθμό προσπαθειών της, οπότε μια εργασία που επαναλαμβάνεται ενώ είναι θαμμένη εκτελείται μία ακόμη φορά και θάβεται ξανά αν αποτύχει:

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

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

Ή μέσω του CLI:

wp queuety retry 42
wp queuety retry-buried

Χρονικό όριο

Το Queuety ενεργοποιεί ένα pcntl_alarm() για τον μέγιστο χρόνο εκτέλεσης και προορίζεται να αποτύχει την προσπάθεια με ένα TimeoutException όταν ξεπεραστεί αυτό το όριο. Αυτό το χρονικό όριο δεν επιβάλλεται προς το παρόν (#10): η εργαζόμενη διεργασία δεν ενεργοποιεί ασύγχρονα σήματα, επομένως ένας χειριστής μακράς διάρκειας εκτελείται μέχρι τέλους, ενώ ένας χειριστής που έχει μπλοκαριστεί στο sleep() διακόπτεται όταν ενεργοποιηθεί ο συναγερμός και έπειτα σημειώνεται ως ολοκληρωμένος. Διαμορφώστε το χρονικό όριο καθολικά:

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

Οι εργασίες που παραμένουν κολλημένες σε κατάσταση processing για περισσότερο από QUEUETY_STALE_TIMEOUT δευτερόλεπτα ανακτώνται αυτόματα:

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

Ανακτήστε παρωχημένες εργασίες με μη αυτόματο τρόπο:

wp queuety recover

Ουρές

Οι εργασίες αποστέλλονται σε επώνυμες ουρές. Κάθε εργαζόμενη διεργασία επεξεργάζεται μία ουρά ή μια ταξινομημένη λίστα ουρών (βλ. Εργαζόμενες διεργασίες):

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

wp queuety work --queue=emails

Η προεπιλεγμένη ουρά είναι η default. Μπορείτε να εκτελείτε ξεχωριστές διεργασίες εργαζομένων για κάθε ουρά, ώστε να απομονώνετε τους φόρτους εργασίας.

Περιορισμός ρυθμού

Περιορίστε πόσο συχνά μπορεί να εκτελείται ένας χειριστής μέσα σε ένα χρονικό διάστημα:

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

Για λεπτομέρειες, δείτε την ενότητα Περιορισμός ρυθμού.

Κλάσεις εργασιών προς αποστολή

Το Queuety v0.6.0 εισάγει μια κλάση εργασίας με βάση το API. Αντί να διαχωρίζετε το όνομα του χειριστή, το ωφέλιμο φορτίο και την κλάση χειριστή, ορίζετε μία κλάση που τα περιλαμβάνει όλα: το ωφέλιμο φορτίο (ως τυποποιημένες δημόσιες ιδιότητες), τη λογική εκτέλεσης και προαιρετικό ενδιάμεσο λογισμικό.

Η διεπαφή Contracts\Job

use Queuety\Contracts\Job;

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

Μια κλάση Job διατηρεί το ωφέλιμο φορτίο της ως δημόσιες ιδιότητες και εκτελεί τη δική της λογική στο handle(). Δεν απαιτείται ξεχωριστή μέθοδος config() ή εγγραφή χειριστή.

Το trait Dispatchable

Προσθέστε το trait Dispatchable για να αποκτήσετε μια στατική μέθοδο dispatch() στην κλάση εργασίας σας:

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

Στείλτε την περνώντας ορίσματα στον κατασκευαστή:

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

Το trait Dispatchable καλεί εσωτερικά το Queuety::dispatch(), επομένως όλες οι επιλογές του PendingJob (ουρά, προτεραιότητα, καθυστέρηση, max_attempts, unique, after) λειτουργούν ακριβώς με τον ίδιο τρόπο.

Πώς λειτουργεί η σειριοποίηση

Όταν στέλνετε ένα στιγμιότυπο Job, το Queuety του JobSerializer εξάγει το πλήρως προσδιορισμένο όνομα της κλάσης ως χειριστή και όλες τις δημόσιες ιδιότητες ως ωφέλιμο φορτίο. Στην πλευρά της εργαζόμενης διεργασίας, ο σειριοποιητής ανακατασκευάζει το στιγμιότυπο αντιστοιχίζοντας τα κλειδιά του ωφέλιμου φορτίου στα ονόματα των παραμέτρων του κατασκευαστή. Τα backed enum σειριοποιούνται στην υποκείμενη τιμή τους και επαναφέρονται αυτόματα.

Ιδιότητες εργασιών

Οι κλάσεις εργασιών μπορούν να δηλώνουν δημόσιες ιδιότητες για να ελέγχουν τη συμπεριφορά επανάληψης και χρονικού ορίου. Η εργαζόμενη διεργασία τις διαβάζει μέσω reflection και τις εφαρμόζει αυτόματα.

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

Ο πίνακας $backoff παρέχει αυξανόμενες καθυστερήσεις επανάληψης. Για παράδειγμα, [10, 30, 60, 120] σημαίνει ότι η πρώτη επανάληψη περιμένει 10 δευτερόλεπτα, η δεύτερη 30 δευτερόλεπτα, η τρίτη 60 δευτερόλεπτα και οι επόμενες επαναλήψεις περιμένουν 120 δευτερόλεπτα.

Οι ιδιότητες που σχετίζονται με πόρους είναι χρήσιμες όταν οι κλάσεις εργασιών αντιπροσωπεύουν διαφορετικά προφίλ κόστους. Μια γρήγορη ανανέωση προσωρινής μνήμης και μια μεγάλη κλήση παρόχου μπορεί να είναι και οι δύο «μία εργασία», αλλά δεν είναι απαραίτητο να μετρούν το ίδιο όταν ένας προϋπολογισμός ροής εργασίας ή μια απόφαση εισαγωγής εργαζομένου προσπαθεί να διατηρήσει τη διεργασία εντός ασφαλών ορίων.

Για το πλήρες μοντέλο, συμπεριλαμβανομένων των προϋπολογισμών ροών εργασίας και της εισαγωγής από την πλευρά των εργαζομένων, δείτε τη Διαχείριση πόρων.

Χειριστής αποτυχίας

Όταν μια εργασία θάβεται αφού εξαντλήσει όλες τις προσπάθειες επανάληψης, το Queuety καλεί μια μέθοδο failed() στο στιγμιότυπο της εργασίας, αν έχει οριστεί. Χρησιμοποιήστε την για καθαρισμό, ειδοποιήσεις ή καταγραφή.

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

Η μέθοδος failed() λαμβάνει την εξαίρεση που προκάλεσε την τελική αποτυχία. Το στιγμιότυπο της εργασίας αποσειριοποιείται με το αρχικό ωφέλιμο φορτίο του, οπότε είναι διαθέσιμες όλες οι δημόσιες ιδιότητες.

Υπό όρους αποστολή

Το trait Dispatchable παρέχει τα dispatch_if() και dispatch_unless() για αποστολή υπό όρους. Και τα δύο δέχονται μια συνθήκη boolean ως πρώτο όρισμα, ακολουθούμενη από τα ορίσματα του κατασκευαστή της εργασίας.

// 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() στέλνει την εργασία όταν η συνθήκη είναι true και επιστρέφει το PendingJob. Επιστρέφει null όταν η συνθήκη είναι false.

Το dispatch_unless() στέλνει την εργασία όταν η συνθήκη είναι false και επιστρέφει το PendingJob. Επιστρέφει null όταν η συνθήκη είναι true.

Και οι δύο μέθοδοι υποστηρίζουν τη σύνδεση επιλογών PendingJob κατά την αποστολή:

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

Σύγχρονη αποστολή

Χρησιμοποιήστε το Queuety::dispatch_sync() για να εκτελέσετε μια εργασία αμέσως στην τρέχουσα διεργασία, χωρίς να την προσθέσετε στην ουρά. Είναι χρήσιμο για δοκιμές και για εργασίες που πρέπει να εκτελεστούν ενσωματωμένα.

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

Η μέθοδος handle() της εργασίας καλείται απευθείας. Δεν δημιουργείται εγγραφή στη βάση δεδομένων, δεν συμμετέχει worker και δεν εφαρμόζεται ενδιάμεσο λογισμικό.

Αλυσίδες εργασιών

Οι αλυσίδες εργασιών εκτελούν μια ακολουθία εργασιών τη μία μετά την άλλη. Κάθε εργασία στην αλυσίδα εξαρτάται από την επιτυχή ολοκλήρωση της προηγούμενης. Αν αποτύχει οποιαδήποτε εργασία, οι επόμενες εργασίες στην αλυσίδα παραμένουν σε αναμονή.

Χρήση του 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() επιστρέφει ένα ChainBuilder με μεθόδους αλυσιδωτής κλήσης για τον ορισμό της ουράς και ενός χειριστή αποτυχίας. Η μέθοδος dispatch() δημιουργεί όλες τις εργασίες με διαδοχικές σχέσεις depends_on και επιστρέφει το αναγνωριστικό της πρώτης εργασίας.

Χρήση του with_chain()

Το trait Dispatchable παρέχει μια μέθοδο with_chain(). Προς το παρόν είναι ισοδύναμη με το Queuety::chain( $jobs ): η κλάση στην οποία καλείται δεν προστίθεται στην αλυσίδα (#13), επομένως συμπεριλάβετε και την πρώτη εργασία στον πίνακα:

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

Χειρισμός αποτυχίας αλυσίδας

Καταχωρίστε μια κλάση χειριστή αποτυχίας χρησιμοποιώντας το catch(). Η κλάση χειριστή πρέπει να έχει μέθοδο handle(). Καλείται όταν οποιαδήποτε εργασία στην αλυσίδα θαφτεί οριστικά. Η εργαζόμενη διεργασία υποστηρίζει τα handle(), handle( Job $job ) και handle( Job $job, \Throwable $exception ).

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

Προσθήκη ενδιάμεσου λογισμικού

Οι κλάσεις εργασιών μπορούν να ορίσουν μια μέθοδο middleware() που επιστρέφει ενδιάμεσο λογισμικό το οποίο περιβάλλει την εκτέλεση:

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

Για λεπτομέρειες σχετικά με το ενσωματωμένο ενδιάμεσο λογισμικό και τη δημιουργία προσαρμοσμένου, δείτε την ενότητα Ενδιάμεσο λογισμικό.

Σύγκριση με εργασίες βάσει χειριστών

Υποστηρίζονται πλήρως και οι δύο προσεγγίσεις. Οι υπάρχουσες εργασίες βάσει χειριστών συνεχίζουν να λειτουργούν χωρίς αλλαγές. Μπορείτε να συνδυάσετε και τα δύο στυλ στο ίδιο έργο.

Κύκλος ζωής εργασίας

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