İşler

İşler, Queuety içindeki temel çalışma birimidir. Bir işleyici adı ve yük içeren bir iş dağıtın; çalışan işi alıp arka planda yürütür.

İş dağıtımı

use Queuety\Queuety;

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

dispatch() bir PendingJob oluşturucu döndürür. Otomatik dağıtılmadan önce seçenekleri zincirleyebilirsiniz:

use Queuety\Enums\Priority;

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

İş kimliğini hemen almak için ->id() çağırın:

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

İşleyici yazma

İşleyici, Queuety\Handler arayüzünü uygulayan herhangi bir sınıftır:

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() yöntemi iş yükünü alır ve işi gerçekleştirir. config() yöntemi isteğe bağlı varsayılan değerleri döndürür:

Bu kaynak anahtarları, iş metaverilerinin, iş akışı bütçelerinin ve çalışan kabulünün birlikte çalıştığı Queuety kapsamındaki Kaynak Yönetimi modelinin parçasıdır.

İşleyiciyi kaydedin:

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

Ya da otomatik kayıt için PHP özniteliklerini kullanın.

Öncelik

İşlerin dört öncelik düzeyi vardır. Önceliği yüksek işler, çalışanlar tarafından önce alınır.

use Queuety\Enums\Priority;

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

Gecikme

Bir işi, yalnızca belirtilen saniye sayısı geçtikten sonra kullanılabilir olacak şekilde geciktirin:

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

İş hemen eklenir, ancak gecikme süresi dolana kadar çalışanlar tarafından alınmaz.

Benzersiz işler

Aynı işleyici ve yük için yinelenen gönderimleri önleyin. Aynı işleyici ve yüke sahip bekleyen veya işlenmekte olan bir iş zaten varsa, yeni bir iş oluşturmak yerine mevcut işin kimliği döndürülür.

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

İş bağımlılıkları

Bir iş başka bir işe bağlı olabilir. Bağımlı iş tamamlanana kadar çalışan tarafından alınmaz.

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

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

Toplu gönderim

En yüksek aktarım hızı için tek bir çok satırlı INSERT ile birden fazla iş gönderin:

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

Her öğe isteğe bağlı şu anahtarları içerebilir: queue, priority, delay, max_attempts.

Yeniden deneme ve bekleme süresi

Bir iş istisna fırlattığında, Queuety işi pending durumuna geri alır ve toplamda max_attempts kez çalışana kadar gecikmeli bir available_at uygular (ilk çalıştırma da sayılır; bu nedenle max_attempts( 3 ) en fazla iki yeniden deneme anlamına gelir). Her yeniden denemeden önceki gecikme, son başarısız çalıştırmanın numarası olan attempt değerini kullanan bekleme stratejisine göre belirlenir:

Genel varsayılanı wp-config.php içinde ayarlayın:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Ya da işleyici başına config() içinde ayarlayın:

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

Ya da gönderim başına ayarlayın:

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

Gömülmüş işler

Tüm yeniden deneme girişimleri tükendiğinde iş gömülü olarak işaretlenir. Gömülü işler inceleme için veritabanında kalır ancak işlenmez. Bunları elle yeniden deneyebilirsiniz. Elle yeniden deneme, girişim sayacını sıfırlamadan işi tekrar pending yapar; bu nedenle yeniden denenen gömülü iş bir kez daha çalıştırılır ve bu çalıştırma başarısız olursa tekrar gömülür:

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

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

Ya da CLI aracılığıyla:

wp queuety retry 42
wp queuety retry-buried

Zaman aşımı

Queuety, maksimum yürütme süresi için bir pcntl_alarm() başlatır ve süre aşıldığında girişimi bir TimeoutException ile başarısız kılmak üzere tasarlanmıştır. Bu zaman aşımı şu anda uygulanmıyor (#10): çalışan asenkron sinyalleri etkinleştirmediğinden uzun süren bir işleyici tamamlanana kadar çalışır; sleep() içinde engellenen bir işleyici ise alarm çaldığında kesilir ve ardından tamamlandı olarak işaretlenir. Zaman aşımını genel olarak yapılandırın:

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

processing durumunda QUEUETY_STALE_TIMEOUT saniyeden uzun süre takılı kalan işler otomatik olarak kurtarılır:

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

Eski işleri elle kurtarın:

wp queuety recover

Kuyruklar

İşler adlandırılmış kuyruklara gönderilir. Her çalışan bir kuyruğu veya sıralı bir kuyruk listesini işler (bkz. Çalışanlar):

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

wp queuety work --queue=emails

Varsayılan kuyruk default şeklindedir. İş yüklerini yalıtmak için her kuyrukta ayrı çalışan süreçleri çalıştırabilirsiniz.

Hız sınırlama

Bir işleyicinin belirli bir zaman aralığında ne sıklıkla çalışabileceğini sınırlayın:

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

Ayrıntılar için Hız Sınırlama bölümüne bakın.

Gönderilebilir iş sınıfları

Queuety v0.6.0, sınıf tabanlı iş API yaklaşımını sunar. İşleyici adını, yükü ve işleyici sınıfını ayrı tutmak yerine her şeyi kapsayan tek bir sınıf tanımlarsınız: yük (türlü genel özellikler olarak), yürütme mantığı ve isteğe bağlı ara katman yazılımı.

Contracts\Job arayüzü

use Queuety\Contracts\Job;

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

Bir Job sınıfı, yükünü genel özellikler olarak tutar ve kendi mantığını handle() içinde yürütür. Ayrı bir config() yöntemi veya işleyici kaydı gerekmez.

Dispatchable özelliği

İş sınıfınıza statik bir dispatch() yöntemi eklemek için Dispatchable özelliğini ekleyin:

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

Kurucu bağımsız değişkenlerini ileterek gönderin:

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

Dispatchable özelliği dahili olarak Queuety::dispatch() çağırır; bu nedenle tüm PendingJob seçenekleri (queue, priority, delay, max_attempts, unique, after) aynı şekilde çalışır.

Serileştirme nasıl çalışır

Bir Job örneğini gönderdiğinizde, Queuety içindeki JobSerializer tam nitelikli sınıf adını işleyici, tüm genel özellikleri ise yük olarak çıkarır. Çalışan tarafında serileştirici, yük anahtarlarını kurucu parametre adlarıyla eşleştirerek örneği yeniden oluşturur. Değer taşıyan enum'lar temel değerlerine serileştirilir ve otomatik olarak geri yüklenir.

İş özellikleri

İş sınıfları, yeniden deneme ve zaman aşımı davranışını denetlemek için genel özellikler tanımlayabilir. Çalışan bunları yansıma yoluyla okur ve otomatik olarak uygular.

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 dizisi, artan yeniden deneme gecikmeleri sağlar. Örneğin [10, 30, 60, 120], ilk yeniden denemenin 10 saniye, ikincinin 30 saniye, üçüncünün 60 saniye bekleyeceği; sonraki yeniden denemelerin ise 120 saniye bekleyeceği anlamına gelir.

Kaynaklarla ilgili özellikler, farklı maliyet profillerine sahip işleri temsil eden iş sınıfları için yararlıdır. Hızlı bir önbellek yenilemesi ile büyük bir sağlayıcı çağrısı da "bir iş" olabilir; ancak iş akışı bütçesi veya çalışan kabul kararı süreci güvenli sınırlar içinde tutmaya çalışırken bunların aynı ağırlıkta sayılması gerekmeyebilir.

İş akışı bütçeleri ve çalışan tarafındaki kabul dâhil tüm model için Kaynak Yönetimi bölümüne bakın.

Başarısız kancası

Bir iş tüm yeniden deneme girişimleri tükendikten sonra gömüldüğünde Queuety, tanımlanmışsa iş örneğinde bir failed() yöntemi çağırır. Bunu temizlik, uyarı gönderme veya günlük kaydı için kullanın.

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() yöntemi son başarısızlığa neden olan istisnayı alır. İş örneği özgün yüküyle serileştirmeden çıkarılır; böylece tüm genel özellikler kullanılabilir olur.

Koşullu gönderim

Dispatchable özelliği, koşullu gönderim için dispatch_if() ve dispatch_unless() yöntemlerini sağlar. Her ikisi de ilk bağımsız değişken olarak bir Boole koşulunu, ardından iş kurucusunun bağımsız değişkenlerini alır.

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

Koşul true olduğunda dispatch_if() gönderim yapar ve PendingJob değerini döndürür. Koşul false olduğunda null döndürür.

Koşul false olduğunda dispatch_unless() gönderim yapar ve PendingJob değerini döndürür. Koşul true olduğunda null döndürür.

Her iki yöntem de gönderim sırasında PendingJob seçeneklerinin zincirleme kullanılmasını destekler:

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

Eşzamanlı gönderim

Bir işi kuyruğa eklemeden geçerli süreçte hemen yürütmek için Queuety::dispatch_sync() kullanın. Bu, testler ve satır içi çalışması gereken işler için yararlıdır.

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

İşin handle() yöntemi doğrudan çağrılır. Veritabanı satırı oluşturulmaz, çalışan kullanılmaz ve ara katman uygulanmaz.

İş zincirleme

İş zincirleme, bir iş dizisini art arda yürütür. Zincirdeki her iş, önceki işin başarıyla tamamlanmasına bağlıdır. Herhangi bir iş başarısız olursa zincirdeki sonraki işler beklemede kalır.

Queuety::chain() kullanımı

use Queuety\Queuety;

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

Queuety::chain(), kuyruğu ve başarısızlık işleyicisini ayarlamak için akıcı yöntemlere sahip bir ChainBuilder döndürür. dispatch() yöntemi, tüm işleri sıralı depends_on ilişkileriyle oluşturur ve ilk işin kimliğini döndürür.

with_chain() kullanımı

Dispatchable özelliği bir with_chain() yöntemi sağlar. Bu yöntem şu anda Queuety::chain( $jobs ) ile eşdeğerdir: çağrıldığı sınıf zincire eklenmez (#13), bu nedenle dizideki ilk işi de listeleyin:

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

Zincir başarısızlığını ele alma

catch() kullanarak bir başarısızlık işleyici sınıfı kaydedin. İşleyici sınıfında bir handle() yöntemi bulunmalıdır. Zincirdeki herhangi bir iş kalıcı olarak gömüldüğünde çağrılır. Çalışan şu öğeleri destekler: handle(), handle( Job $job ) ve handle( Job $job, \Throwable $exception ).

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

Ara katman yazılımı ekleme

İş sınıfları, yürütmeyi saran ara katman yazılımını döndürmek için bir middleware() yöntemi tanımlayabilir:

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

Yerleşik ara katman yazılımları ve özel ara katman yazılımları yazma hakkında ayrıntılar için Ara Katman Yazılımı bölümüne bakın.

İşleyici tabanlı işlerle karşılaştırma

Her iki yaklaşım da tamamen desteklenir. Mevcut işleyici tabanlı işler değişiklik yapılmadan çalışmaya devam eder. Aynı projede her iki stili de kullanabilirsiniz.

İş yaşam döngüsü

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