Worker

I worker sono processi a lunga esecuzione che interrogano il database alla ricerca di job in attesa, li eseguono e fanno avanzare i workflow. Il comando wp queuety work viene eseguito all'interno di WP-CLI, mentre la coda stessa viene acquisita direttamente da MySQL tramite il driver DB Queuety migliore disponibile.

Il plugin installa anche un fallback WordPress cron predefinito, eseguito una volta al minuto, per gli ambienti senza accesso alla shell. Questa pagina illustra i worker WP-CLI dedicati, la modalità di distribuzione consigliata per una maggiore capacità di elaborazione.

Avviare un worker

wp queuety work

Avvia un singolo worker sulla coda default. Viene eseguito in un ciclo: acquisisce un job, lo esegue e ripete l'operazione. Quando la coda è vuota, attende QUEUETY_WORKER_SLEEP secondi (valore predefinito: 1) prima di interrogarla di nuovo.

Opzioni

# 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

Priorità tra più code

Un singolo worker può elaborare più code in ordine di priorità. Passate un elenco di nomi di code separati da virgole a --queue:

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

Il worker tenta di acquisire un job da ciascuna coda nell'ordine indicato. Controlla prima critical. Se critical contiene un job, lo elabora. Se critical è vuota, passa a default e poi a low. In questo modo le code con priorità più alta vengono svuotate prima di quelle con priorità inferiore.

Funzionamento dell'ordine di priorità

A ogni iterazione, il worker tenta di acquisire un job dalla prima coda dell'elenco. Se la coda è vuota o in pausa, passa alla successiva e continua finché non trova un job o esaurisce l'elenco. Se non trova job in nessuna coda, attende e riprova. Il modello applica priorità rigorose: un flusso costante di job critical impedirà l'elaborazione di default e low. Se le code con priorità inferiore devono garantire una certa capacità di elaborazione, usate worker dedicati.

Usare più code in PHP

I metodi Worker::run() e Worker::flush() accettano una stringa o un 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' );

Strategie di distribuzione

I worker dedicati per coda offrono il pieno controllo sulla concorrenza e sull'isolamento. Ogni coda ha il proprio process(es) worker e non può essere ostacolata dalle altre:

[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

L'ordine di priorità con un solo worker è più semplice da gestire e funziona bene quando le code con priorità inferiore possono tollerare ritardi:

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

Potete anche combinare entrambi gli approcci: worker dedicati per critical per garantire la capacità di elaborazione e un worker con code ordinate per priorità per tutto il resto:

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

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

Ammissione in base alle risorse

I worker non si limitano ad acquisire la prossima riga in attesa. Prima di avviare un job acquisito, Queuety può applicare controlli di ammissione generali in base al gruppo di concorrenza del job e al profilo di esecuzione recente per quel gestore.

Per il modello più ampio, incluso il modo in cui si integra con i metadati dei job e i budget dei workflow, consultate Gestione delle risorse.

Un gruppo di concorrenza assegna un limite condiviso a più gestori o classi di job. Se assegnate lo stesso gruppo e limite a più provider dispendiosi, quando il gruppo raggiunge il limite un worker rilascia il nuovo lavoro acquisito invece di avviarlo. In questo modo le code indipendenti non sovraccaricano un singolo sistema downstream solo perché ci sono più worker inattivi.

Anche i log di esecuzione recenti alimentano un semplice profilo di ammissione. Quando l'ammissione basata sulle risorse è attiva, Queuety esamina la durata media recente e il picco di memoria del gestore che sta per eseguire. Se il worker corrente è già vicino al limite di memoria configurato, se una coda o un gruppo di provider ha già raggiunto il budget ponderato, se il container corrente non sembra avere memoria residua sufficiente per il profilo osservato del gestore oppure se un'esecuzione in stile --once è vicina al proprio limite di tempo, Queuety posticipa il job invece di avviare un'attività che probabilmente non terminerà correttamente.

Potete attivare la policy dalla definizione del lavoro, non dal comando del 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,
        ];
    }
}

Gli stessi metadati possono essere associati a classi di job inviabili tramite proprietà pubbliche oppure a un singolo invio tramite PendingJob::concurrency_group() e PendingJob::cost_units().

Il profilo di ammissione lato worker è controllato da queste costanti:

Ciclo di vita del worker

Un worker si arresta in modo regolare, dopo aver completato il job corrente, quando si verifica una delle seguenti condizioni:

Dopo l'arresto, il worker termina regolarmente. In produzione, un gestore di processi lo riavvia automaticamente.

La gestione dei segnali non è ancora corretta (#4). Un singolo processo wp queuety work non installa gestori di segnali, quindi SIGTERM o SIGINT (Ctrl+C) lo terminano immediatamente, anche durante un job. Il job resta in stato processing finché il ripristino dei job obsoleti non lo reimposta dopo QUEUETY_STALE_TIMEOUT secondi; assicuratevi quindi che i gestori possano essere eseguiti di nuovo in sicurezza. Un pool di worker inoltra SIGTERM ai processi figli, ma questi non lo gestiscono, quindi il pool continua a funzionare finché non si arrestano per uno dei motivi sopra indicati. Per arrestare un pool, impostate nel gestore di processi un timeout di arresto, al termine del quale verrà inviato SIGKILL.

Concorrenza con più worker

Il flag --workers=N avvia N processi figli usando pcntl_fork(). Ogni processo figlio crea la propria connessione al database ed elabora i job in modo indipendente. Il processo padre monitora i processi figli e riavvia quelli che si arrestano in modo anomalo.

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

Requisiti: Devono essere installate le estensioni pcntl e posix PHP (senza pcntl, il comando non riesce e restituisce "pcntl extension is required for --workers=N."); ogni processo figlio usa la propria connessione al database, quindi l'impostazione MySQL max_connections deve poter gestire il pool. I pool adattivi (--min-workers/--max-workers) hanno gli stessi requisiti.

Pool di worker adattivi

I pool adattivi mantengono attivo un numero minimo di worker, quindi aumentano o riducono il numero entro l'intervallo configurato:

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

Il processo padre controlla il numero di job acquisibili nell'insieme di code selezionato e adegua il numero di processi figli a tale quantità, senza superare il massimo configurato. Quando la coda si calma, il pool attende il periodo di tolleranza per l'inattività prima di ridurre il numero di worker, evitando così che brevi picchi facciano variare continuamente l'elenco dei processi.

L'aumento del pool resta prudente. Quando Queuety può leggere i dati telemetrici sulla memoria del container o dell'host, non aggiunge altri processi figli se la memoria residua non è sufficiente per l'intero fabbisogno di un altro worker.

Il ciclo di scalabilità lato processo padre è controllato da:

Ripristino dei job obsoleti

Se un worker si arresta in modo anomalo mentre elabora un job, il job resta nello stato processing. Un worker a lunga esecuzione controlla circa una volta al minuto se ci sono job rimasti in processing per più di QUEUETY_STALE_TIMEOUT secondi. Un job obsoleto con tentativi residui viene riprovato dopo il ritardo previsto dal backoff; un job che ha esaurito i tentativi viene messo in quarantena e il relativo workflow viene contrassegnato come non riuscito. I worker --once e il fallback WP-Cron non eseguono questo controllo.

Ripristinate manualmente i job obsoleti:

wp queuety recover

Modalità flush

Elabora tutti i job in attesa e termina. Utile per i test o per elaborazioni batch una tantum:

wp queuety flush
wp queuety flush --queue=emails

Distribuzione in produzione

In produzione, usate un gestore di processi per mantenere attivi i worker. Ecco due approcci comuni.

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

Abilitate e avviate:

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

Suggerimenti per la distribuzione

Esegui worker separati per code diverse quando serve un isolamento maggiore. Usa --workers=N per carichi di lavoro limitati dalla CPU e preferisci worker supervisionati singolarmente quando il carico è prevalentemente I/O. Imposta QUEUETY_WORKER_MAX_JOBS se vuoi riavvii periodici per liberare memoria e tieni d’occhio il conteggio di buried tramite webhook o metriche.