Worker

Worker adalah proses berjalan lama yang memeriksa basis data untuk mencari job tertunda, menjalankannya, lalu melanjutkan alur kerja. Perintah wp queuety work berjalan di dalam WP-CLI, sementara antrean diklaim langsung dari MySQL melalui driver DB Queuety terbaik yang tersedia.

Plugin ini juga memasang fallback WordPress cron default yang berjalan sekali per menit untuk lingkungan tanpa akses shell. Halaman ini membahas worker WP-CLI khusus, yaitu mode deployment yang direkomendasikan untuk throughput lebih tinggi.

Menjalankan worker

wp queuety work

Perintah ini memulai satu worker pada antrean default. Worker berjalan dalam loop: mengambil job, menjalankannya, lalu mengulangi proses. Saat antrean kosong, worker tidur selama QUEUETY_WORKER_SLEEP detik (default: 1) sebelum memeriksa lagi.

Opsi

# Process a specific queue
wp queuety work --queue=emails

# Process at most one job and exit
wp queuety work --once

# Run 4 concurrent workers
wp queuety work --workers=4

# Scale between 2 and 6 workers as backlog changes
wp queuety work --queue=providers --min-workers=2 --max-workers=6

Prioritas beberapa antrean

Satu worker dapat memproses beberapa antrean sesuai urutan prioritas. Teruskan daftar nama antrean yang dipisahkan koma ke --queue:

wp queuety work --queue=critical,default,low

Worker mencoba mengambil job dari setiap antrean sesuai urutan dalam daftar. Worker memeriksa critical terlebih dahulu. Jika critical memiliki job, worker memprosesnya. Jika critical kosong, worker beralih ke default, lalu ke low. Artinya, antrean berprioritas lebih tinggi selalu dikosongkan sebelum antrean berprioritas lebih rendah.

Cara kerja urutan prioritas

Pada setiap iterasi, worker mencoba mengambil job dari antrean pertama dalam daftar. Jika antrean itu kosong atau dijeda, worker beralih ke antrean berikutnya dan terus mencoba sampai menemukan job atau menghabiskan seluruh daftar. Jika tidak ada job yang ditemukan, worker tidur lalu mencoba lagi. Model ini menerapkan prioritas ketat, sehingga aliran critical job yang terus-menerus akan membuat antrean default dan low kelaparan. Jika antrean berprioritas lebih rendah memerlukan throughput yang terjamin, gunakan worker khusus.

Menggunakan beberapa antrean di PHP

Metode Worker::run() dan Worker::flush() menerima string atau array:

// Comma-separated string
Queuety::worker()->run( 'critical,default,low' );

// Array
Queuety::worker()->run( [ 'critical', 'default', 'low' ] );

// flush() also supports multi-queue
$count = Queuety::worker()->flush( 'critical,default,low' );

Strategi deployment

Worker khusus untuk setiap antrean memberi Anda kendali penuh atas konkurensi dan isolasi. Setiap antrean memiliki process(es) worker sendiri dan tidak dapat diabaikan oleh antrean lain:

[program:queuety-critical]
command=wp queuety work --queue=critical
numprocs=4

[program:queuety-default]
command=wp queuety work --queue=default
numprocs=2

[program:queuety-low]
command=wp queuety work --queue=low
numprocs=1

Urutan prioritas dengan satu worker lebih mudah dikelola dan cocok jika antrean berprioritas lebih rendah dapat menoleransi penundaan:

[program:queuety-worker]
command=wp queuety work --queue=critical,default,low
numprocs=2

Anda juga dapat menggabungkan kedua pendekatan: worker khusus untuk critical guna menjamin throughput, serta worker dengan urutan prioritas untuk antrean lainnya:

[program:queuety-critical]
command=wp queuety work --queue=critical
numprocs=2

[program:queuety-fallback]
command=wp queuety work --queue=default,low
numprocs=2

Penerimaan berdasarkan sumber daya

Worker tidak hanya mengambil baris tertunda berikutnya. Sebelum job yang telah diklaim dimulai, Queuety dapat menerapkan pemeriksaan penerimaan kasar berdasarkan grup konkurensi job dan profil eksekusi terbaru handler tersebut.

Untuk memahami model secara lebih luas, termasuk kaitannya dengan metadata job dan anggaran alur kerja, lihat Manajemen Sumber Daya.

Grup konkurensi memberi beberapa handler atau kelas job batas maksimum yang sama. Jika Anda menetapkan grup dan batas yang sama untuk beberapa penyedia yang membutuhkan banyak sumber daya, worker akan melepas pekerjaan yang baru diklaim alih-alih memulainya setelah grup mencapai batas. Ini mencegah antrean yang terpisah membanjiri satu sistem hilir hanya karena ada beberapa worker yang menganggur.

Log eksekusi terbaru juga menyediakan profil penerimaan sederhana. Saat penerimaan sumber daya diaktifkan, Queuety memeriksa durasi rata-rata dan penggunaan memori puncak terbaru untuk handler yang akan dijalankan. Jika worker saat ini hampir mencapai batas memorinya, antrean atau grup penyedia sudah mencapai anggaran berbobotnya, kontainer saat ini tampaknya tidak memiliki sisa memori yang cukup untuk profil handler yang teramati, atau eksekusi bergaya --once hampir mencapai batas waktunya, Queuety menunda job alih-alih memulai pekerjaan yang kemungkinan tidak dapat diselesaikan dengan baik.

Anda mengaktifkan kebijakan ini dari definisi pekerjaan, bukan dari perintah worker:

class CallProviderStep implements \Queuety\Step {
    public function handle( array $state ): array {
        // ...
    }

    public function config(): array {
        return [
            'concurrency_group' => 'providers',
            'concurrency_limit' => 3,
            'cost_units' => 5,
        ];
    }
}

Metadata yang sama dapat dilampirkan ke kelas job yang dapat di-dispatch melalui properti publik, atau ke dispatch individual melalui PendingJob::concurrency_group() dan PendingJob::cost_units().

Profil penerimaan di sisi worker dikendalikan oleh konstanta berikut:

Siklus hidup worker

Worker berhenti dengan aman setelah menyelesaikan job saat ini jika salah satu kondisi berikut terpenuhi:

Setelah berhenti, worker keluar dengan bersih. Di lingkungan produksi, pengelola proses memulainya kembali secara otomatis.

Penanganan sinyal belum dilakukan dengan aman (#4). Satu proses wp queuety work tidak memasang penangan sinyal, sehingga SIGTERM atau SIGINT (Ctrl+C) akan langsung menghentikannya, bahkan di tengah job. Job tersebut tetap berstatus processing sampai pemulihan job usang mengatur ulang statusnya setelah QUEUETY_STALE_TIMEOUT detik, jadi pastikan handler aman untuk dicoba ulang. Pool worker meneruskan SIGTERM ke proses turunannya, tetapi proses tersebut tidak menanganinya. Karena itu, pool terus berjalan sampai berhenti karena salah satu alasan di atas. Untuk menghentikan pool, atur batas waktu berhenti pada pengelola proses. Setelah batas waktu itu, pengelola akan mengirim SIGKILL.

Konkurensi beberapa worker

Flag --workers=N membuat fork untuk N proses turunan menggunakan pcntl_fork(). Setiap proses turunan membuat koneksi basis datanya sendiri dan memproses job secara terpisah. Proses induk memantau proses turunannya dan memulai ulang proses yang mengalami crash.

wp queuety work --queue=default --workers=4

Persyaratan: Ekstensi pcntl dan posix PHP harus terpasang (tanpa pcntl, perintah akan gagal dengan pesan "pcntl extension is required for --workers=N."), dan setiap proses turunan menggunakan koneksi basis datanya sendiri, jadi pengaturan MySQL max_connections Anda harus mengakomodasi pool. Pool adaptif (--min-workers/--max-workers) memiliki persyaratan yang sama.

Pool worker adaptif

Pool adaptif menjaga sejumlah worker minimum tetap aktif, lalu menambah atau mengurangi jumlahnya dalam rentang yang dikonfigurasi:

wp queuety work --queue=providers --min-workers=2 --max-workers=6

Proses induk memantau backlog yang dapat diklaim pada kumpulan antrean yang dipilih dan menyesuaikan jumlah proses turunan untuk mendekati backlog tersebut, hingga batas maksimum yang dikonfigurasi. Saat antrean mulai lengang, pool menunggu selama jendela waktu siaga sebelum mengurangi jumlah worker agar lonjakan singkat tidak langsung menyebabkan daftar proses berubah-ubah.

Peningkatan skala masih dilakukan secara konservatif. Saat Queuety dapat membaca telemetri memori kontainer atau host, alat ini menolak menambah proses anak jika sisa memori tidak cukup untuk satu alokasi penuh bagi pekerja tambahan.

Loop penskalaan di sisi induk dikendalikan oleh:

Pemulihan job usang

Jika worker mengalami crash saat memproses job, job tersebut akan berstatus processing. Worker yang berjalan lama memeriksa kira-kira sekali per menit apakah ada job yang telah berstatus processing lebih lama dari QUEUETY_STALE_TIMEOUT detik. Job usang yang masih memiliki sisa percobaan akan dicoba kembali setelah jeda backoff; job yang telah menggunakan semua percobaannya akan dikubur dan alur kerjanya ditandai gagal. Worker --once dan fallback WP-Cron tidak menjalankan pemeriksaan ini.

Pulihkan job usang secara manual:

wp queuety recover

Mode flush

Proses semua job tertunda lalu keluar. Berguna untuk pengujian atau pemrosesan batch satu kali:

wp queuety flush
wp queuety flush --queue=emails

Deployment produksi

Untuk produksi, gunakan pengelola proses agar worker tetap berjalan. Berikut dua pendekatan yang umum digunakan.

Supervisord

[program:queuety-default]
command=wp queuety work --queue=default
directory=/var/www/html
autostart=true
autorestart=true
numprocs=2
process_name=%(program_name)s_%(process_num)02d
stdout_logfile=/var/log/queuety-default.log
stderr_logfile=/var/log/queuety-default-error.log

[program:queuety-emails]
command=wp queuety work --queue=emails
directory=/var/www/html
autostart=true
autorestart=true
numprocs=1
stdout_logfile=/var/log/queuety-emails.log
stderr_logfile=/var/log/queuety-emails-error.log

Systemd

[Unit]
Description=Queuety Worker (default queue)
After=mysql.service

[Service]
Type=simple
User=www-data
WorkingDirectory=/var/www/html
ExecStart=/usr/local/bin/wp queuety work --queue=default
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

Aktifkan dan jalankan:

sudo systemctl enable queuety-worker
sudo systemctl start queuety-worker

Tips deployment

Jalankan worker terpisah untuk antrean yang berbeda jika Anda memerlukan isolasi lebih kuat. Gunakan --workers=N untuk beban kerja yang dibatasi CPU, dan pilih worker yang diawasi secara terpisah jika beban kerja lebih banyak melibatkan I/O. Tetapkan QUEUETY_WORKER_MAX_JOBS jika Anda ingin memulai ulang secara berkala untuk membebaskan memori, dan pantau jumlah buried melalui webhook atau metrik.