Job

Job adalah unit kerja inti di Queuety. Kirim job dengan nama handler dan payload, lalu worker akan mengambil dan menjalankannya di latar belakang.

Mengirim job

use Queuety\Queuety;

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

dispatch() mengembalikan builder PendingJob. Anda dapat merangkai opsi sebelum job dikirim secara otomatis:

use Queuety\Enums\Priority;

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

Untuk langsung mendapatkan ID job, panggil ->id():

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

Menulis handler

Handler adalah kelas apa pun yang mengimplementasikan antarmuka 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',
        ];
    }
}

Metode handle() menerima payload job dan menjalankan pekerjaan. Metode config() mengembalikan nilai default opsional:

Kunci sumber daya ini merupakan bagian dari model Manajemen Sumber Daya Queuety yang lebih luas, yang menyatukan metadata job, anggaran alur kerja, dan penerimaan worker.

Daftarkan handler:

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

Atau gunakan atribut PHP untuk pendaftaran otomatis.

Prioritas

Job memiliki empat tingkat prioritas. Worker mengambil job berprioritas lebih tinggi terlebih dahulu.

use Queuety\Enums\Priority;

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

Penundaan

Tunda job agar hanya tersedia setelah jumlah detik yang ditentukan:

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

Job langsung dimasukkan, tetapi worker tidak akan mengambilnya sampai waktu penundaan berlalu.

Job unik

Cegah pengiriman duplikat untuk handler dan payload yang sama. Jika sudah ada job tertunda atau sedang diproses dengan handler dan payload yang sama, ID job yang ada akan dikembalikan alih-alih membuat job baru.

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

Dependensi job

Sebuah job dapat bergantung pada job lain. Worker tidak akan mengambilnya sampai dependensinya selesai.

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

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

Pengiriman batch

Kirim beberapa job dalam satu INSERT multi-baris untuk throughput maksimum:

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

Setiap item dapat menyertakan kunci opsional: queue, priority, delay, max_attempts.

Percobaan ulang dan jeda

Saat job memunculkan pengecualian, Queuety mengembalikannya ke pending dengan available_at tertunda hingga total dijalankan sebanyak max_attempts kali (eksekusi pertama ikut dihitung, jadi max_attempts( 3 ) berarti hingga dua kali percobaan ulang). Penundaan sebelum setiap percobaan ulang ditentukan oleh strategi jeda, dengan attempt sebagai nomor eksekusi yang baru saja gagal:

Tetapkan default global di wp-config.php:

define( 'QUEUETY_RETRY_BACKOFF', 'linear' );

Atau per handler di config():

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

Atau per pengiriman:

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

Pekerjaan yang dikarantina

Setelah semua percobaan ulang habis, job ditandai sebagai dikubur. Job yang dikubur tetap berada di database untuk diperiksa, tetapi tidak diproses. Anda dapat mencoba ulang secara manual. Percobaan ulang manual membuat job pending kembali tanpa mengatur ulang jumlah percobaannya, sehingga job yang dikubur dan dicoba ulang akan mendapat satu kali eksekusi lagi dan dikubur kembali jika eksekusi itu gagal:

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

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

Atau melalui CLI:

wp queuety retry 42
wp queuety retry-buried

Batas waktu

Queuety mengaktifkan pcntl_alarm() selama waktu eksekusi maksimum dan dirancang untuk menggagalkan percobaan dengan TimeoutException jika batas waktu terlampaui. Batas waktu tersebut saat ini belum diterapkan (#10): worker tidak mengaktifkan sinyal asinkron, sehingga handler yang berjalan lama tetap berjalan hingga selesai, sedangkan handler yang terblokir di sleep() terhenti saat alarm berbunyi lalu ditandai selesai. Konfigurasikan batas waktu secara global:

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

Job yang terjebak dalam status processing lebih lama dari QUEUETY_STALE_TIMEOUT detik akan dipulihkan secara otomatis:

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

Pulihkan job usang secara manual:

wp queuety recover

Antrean

Job dikirim ke antrean bernama. Setiap worker memproses satu antrean atau daftar antrean berurutan (lihat Worker):

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

wp queuety work --queue=emails

Antrean default adalah default. Anda dapat menjalankan proses worker terpisah untuk setiap antrean guna mengisolasi beban kerja.

Pembatasan laju

Batasi seberapa sering handler dapat dijalankan dalam jangka waktu tertentu:

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

Lihat Pembatasan Laju untuk detailnya.

Kelas job yang dapat di-dispatch

Queuety v0.6.0 memperkenalkan API job berbasis kelas. Alih-alih memisahkan nama handler, payload, dan kelas handler, Anda dapat menentukan satu kelas yang merangkum semuanya: payload (sebagai properti publik bertipe), logika eksekusi, dan middleware opsional.

Antarmuka Contracts\Job

use Queuety\Contracts\Job;

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

Kelas Job menyimpan payload sebagai properti publik dan menjalankan logikanya sendiri di handle(). Metode config() terpisah atau pendaftaran handler tidak diperlukan.

Trait Dispatchable

Tambahkan trait Dispatchable untuk mendapatkan metode statis dispatch() pada kelas job Anda:

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

Kirim job dengan memberikan argumen konstruktor:

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

Trait Dispatchable memanggil Queuety::dispatch() secara internal, sehingga semua opsi PendingJob (queue, priority, delay, max_attempts, unique, after) berfungsi sama persis.

Cara kerja serialisasi

Saat Anda mengirim instance Job, QueuetyJobSerializer mengekstrak nama kelas berkualifikasi penuh sebagai handler dan semua properti publik sebagai payload. Di sisi worker, serializer membangun kembali instance dengan mencocokkan kunci payload dengan nama parameter konstruktor. Enum backed diserialisasi menjadi nilai dasarnya dan dipulihkan secara otomatis.

Properti pekerjaan

Kelas job dapat mendeklarasikan properti publik untuk mengendalikan perilaku percobaan ulang dan batas waktu. Worker membacanya melalui refleksi dan menerapkannya secara otomatis.

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

Array $backoff menyediakan penundaan percobaan ulang yang makin lama. Misalnya, [10, 30, 60, 120] berarti percobaan ulang pertama menunggu 10 detik, kedua 30 detik, ketiga 60 detik, dan percobaan selanjutnya menunggu 120 detik.

Properti terkait sumber daya berguna ketika kelas job mewakili profil biaya yang berbeda. Penyegaran cache yang cepat dan pemanggilan penyedia berskala besar sama-sama dapat disebut "satu job", tetapi belum tentu semestinya dihitung dengan cara yang sama ketika anggaran alur kerja atau keputusan penerimaan worker berupaya menjaga proses tetap dalam batas aman.

Untuk model lengkap, termasuk anggaran alur kerja dan penerimaan di sisi worker, lihat Manajemen Sumber Daya.

Hook kegagalan

Saat job dikubur setelah semua percobaan ulang habis, Queuety memanggil metode failed() pada instance job jika metode tersebut didefinisikan. Gunakan untuk pembersihan, pemberian peringatan, atau pencatatan log.

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

Metode failed() menerima pengecualian yang menyebabkan kegagalan terakhir. Instance job dideserialisasi dengan payload aslinya, sehingga semua properti publik tersedia.

Pengiriman bersyarat

Trait Dispatchable menyediakan dispatch_if() dan dispatch_unless() untuk pengiriman bersyarat. Keduanya menerima kondisi boolean sebagai argumen pertama, diikuti argumen konstruktor job.

// 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() mengirim job jika kondisi bernilai true dan mengembalikan PendingJob. Metode ini mengembalikan null jika kondisi bernilai false.

dispatch_unless() mengirim job jika kondisi bernilai false dan mengembalikan PendingJob. Metode ini mengembalikan null jika kondisi bernilai true.

Kedua metode mendukung perangkaian opsi PendingJob saat mengirim job:

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

Pengiriman sinkron

Gunakan Queuety::dispatch_sync() untuk menjalankan job segera dalam proses saat ini tanpa menambahkannya ke antrean. Ini berguna untuk pengujian dan job yang harus dijalankan secara inline.

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

Metode handle() milik job dipanggil secara langsung. Tidak ada baris database yang dibuat, worker tidak dilibatkan, dan middleware tidak diterapkan.

Perantaian job

Perantaian job menjalankan serangkaian job satu per satu. Setiap job dalam rantai bergantung pada keberhasilan penyelesaian job sebelumnya. Jika ada job yang gagal, job berikutnya dalam rantai tetap tertunda.

Menggunakan 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() mengembalikan ChainBuilder dengan metode fluent untuk mengatur antrean dan handler kegagalan. Metode dispatch() membuat semua job dengan relasi depends_on berurutan dan mengembalikan ID job pertama.

Menggunakan with_chain()

Trait Dispatchable menyediakan metode with_chain(). Saat ini, metode tersebut setara dengan Queuety::chain( $jobs ): kelas tempat metode dipanggil tidak ditambahkan ke rantai (#13), jadi sertakan juga job pertama dalam array:

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

Penanganan kegagalan rantai

Daftarkan kelas handler kegagalan menggunakan catch(). Kelas handler harus memiliki metode handle(). Metode ini dipanggil saat job mana pun dalam rantai dikubur secara permanen. Worker mendukung handle(), handle( Job $job ), dan handle( Job $job, \Throwable $exception ).

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

Menambahkan middleware

Kelas job dapat mendefinisikan metode middleware() untuk mengembalikan middleware yang membungkus eksekusi:

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

Lihat Middleware untuk detail tentang middleware bawaan dan cara menulis middleware khusus.

Perbandingan dengan job berbasis handler

Kedua pendekatan didukung sepenuhnya. Job berbasis handler yang sudah ada tetap berfungsi tanpa perubahan. Anda dapat mencampur kedua gaya dalam proyek yang sama.

Siklus hidup job

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