Ενδιάμεσο λογισμικό

Το middleware περιβάλλει την εκτέλεση εργασιών σε μια διοχέτευση τύπου κρεμμυδιού. Κάθε middleware μπορεί να εκτελέσει λογική πριν και μετά την εργασία, να διακόψει πρόωρα την εκτέλεση ή να συλλάβει εξαιρέσεις. Το Queuety περιλαμβάνει πέντε ενσωματωμένα middleware και υποστηρίζει προσαρμοσμένα middleware μέσω της διεπαφής Contracts\Middleware.

Το middleware είναι διαθέσιμο για κλάσεις εργασιών που μπορούν να αποσταλούν (κλάσεις που υλοποιούν το Contracts\Job με το trait Dispatchable).

Η διεπαφή Contracts\Middleware

use Queuety\Contracts\Middleware;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

Η παράμετρος $job είναι το στιγμιότυπο της εργασίας που επεξεργάζεται. Καλέστε το $next( $job ) για να συνεχιστεί η εκτέλεση στο επόμενο middleware της διοχέτευσης (ή στη μέθοδο handle() της εργασίας, αν αυτό είναι το τελευταίο middleware).

Προσθήκη middleware σε μια εργασία

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

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class SyncInventoryJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $sku,
    ) {}

    public function handle(): void {
        // Sync inventory for this SKU...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 30 ),
            new RateLimited( max: 100, window: 60 ),
        ];
    }
}

Σειρά εκτέλεσης της διοχέτευσης

Τα middleware εφαρμόζονται με τη σειρά του πίνακα, με το πρώτο middleware να αποτελεί το εξωτερικότερο επίπεδο. Αν η μέθοδος middleware() επιστρέφει το [A, B, C], η σειρά εκτέλεσης είναι:

A::handle() -> B::handle() -> C::handle() -> $job->handle()

Κάθε middleware περιβάλλει το επόμενο. Το A εκτελείται πρώτο πριν από την εργασία και τελευταίο μετά από αυτήν, οπότε είναι ιδανικό για χρονομέτρηση ή περιτυλίγματα try/catch.

Δημιουργία προσαρμοσμένου middleware

Υλοποιήστε τη διεπαφή Contracts\Middleware:

use Queuety\Contracts\Middleware;

class LogExecution implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        $class = get_class( $job );
        error_log( "[Queuety] Starting: {$class}" );

        $start = microtime( true );

        try {
            $next( $job );
        } finally {
            $elapsed = round( microtime( true ) - $start, 3 );
            error_log( "[Queuety] Finished: {$class} in {$elapsed}s" );
        }
    }
}

Χρησιμοποιήστε το σε μια εργασία:

public function middleware(): array {
    return [
        new LogExecution(),
    ];
}

Πρόωρη διακοπή

Ένα middleware μπορεί να παραλείψει εντελώς την εργασία αν δεν καλέσει το $next():

class SkipOnMaintenanceMode implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        if ( wp_is_maintenance_mode() ) {
            return; // Job silently skipped.
        }

        $next( $job );
    }
}

Ενσωματωμένο ενδιάμεσο λογισμικό

ThrottlesExceptions

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

use Queuety\Middleware\ThrottlesExceptions;

public function middleware(): array {
    return [
        new ThrottlesExceptions( max_attempts: 10, decay_minutes: 10 ),
    ];
}

Οι μετρήσεις εξαιρέσεων παρακολουθούνται ανά κλάση χειριστή στον πίνακα κλειδωμάτων του Queuety και είναι κοινές σε όλες τις διεργασίες εργαζομένων. Όταν επιτευχθεί το όριο, το middleware προκαλεί ένα RuntimeException χωρίς να καλέσει την εργασία και ο εργαζόμενος το χειρίζεται όπως κάθε άλλη αποτυχία: η προσπάθεια προσμετράται στο max_attempts, η εργασία επαναλαμβάνεται με την κανονική καθυστέρηση και μεταφέρεται στις θαμμένες εργασίες όταν εξαντληθούν οι προσπάθειες.

Παράδειγμα: αν μια εξωτερική υπηρεσία API είναι εκτός λειτουργίας, οι πρώτες 10 αποτυχίες υποβάλλονται σε κανονική επεξεργασία (καθεμία αυξάνει τον μετρητή εξαιρέσεων). Οι επόμενες προσπάθειες αποτυγχάνουν γρήγορα χωρίς να καλέσουν το API, μέχρι να περάσουν 10 λεπτά χωρίς νέα εξαίρεση· το χρονικό παράθυρο ξεκινά ξανά με κάθε καταγεγραμμένη εξαίρεση.

RateLimited

Επιβάλλει μέγιστο αριθμό εκτελέσεων μέσα σε ένα χρονικό διάστημα. Αν το όριο ξεπεραστεί μέσα στο middleware, προκαλείται ένα RateLimitExceededException· ο εργαζόμενος το χειρίζεται ως συνηθισμένη αποτυχημένη προσπάθεια, η οποία επαναλαμβάνεται με την κανονική καθυστέρηση και προσμετράται στο max_attempts. Μόλις το middleware καταχωρίσει το όριο στον περιοριστή ρυθμού του εργαζομένου, οι επόμενες εργασίες του χειριστή επιστρέφουν επίσης στο pending πριν εκτελεστούν, όσο ισχύει το όριο (βλ. Περιορισμός ρυθμού).

use Queuety\Middleware\RateLimited;

public function middleware(): array {
    return [
        new RateLimited( max: 60, window: 60 ), // 60 per minute
    ];
}

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

Timeout

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

use Queuety\Middleware\Timeout;

public function middleware(): array {
    return [
        new Timeout( seconds: 30 ),
    ];
}

Απαιτεί την επέκταση pcntl PHP. Αν το pcntl δεν είναι διαθέσιμο, το middleware συνεχίζει σιωπηρά (δεν επιβάλλεται χρονικό όριο).

UniqueJob

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

use Queuety\Middleware\UniqueJob;

public function middleware(): array {
    return [
        new UniqueJob(),                         // Lock key defaults to the job class name
        new UniqueJob( key: 'sync_inventory' ),   // Custom lock key
    ];
}

WithoutOverlapping

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

use Queuety\Middleware\WithoutOverlapping;

public function middleware(): array {
    return [
        new WithoutOverlapping( key: 'import_products', release_after: 600 ),
    ];
}

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

Παράδειγμα: συνδυασμός middleware

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;
use Queuety\Middleware\WithoutOverlapping;

class ImportProductsJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly int $supplier_id,
    ) {}

    public function handle(): void {
        // Fetch and import products from the supplier...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 300 ),
            new WithoutOverlapping(
                key: "import_supplier_{$this->supplier_id}",
                release_after: 600,
            ),
            new RateLimited( max: 5, window: 60 ),
        ];
    }
}

Αυτή η εργασία θα:

1. Διακοπεί αν εκτελεστεί για περισσότερο από 5 λεπτά (Timeout)
2. Παραλειφθεί αν εκτελείται ήδη άλλη εισαγωγή δεδομένων για τον ίδιο προμηθευτή (WithoutOverlapping)
3. Περιοριστεί σε 5 εισαγωγές δεδομένων ανά λεπτό για όλους τους προμηθευτές (RateLimited)