Workers

Workers zijn langlopende processen die de database controleren op wachtende taken, deze uitvoeren en workflows verderzetten. De opdracht wp queuety work wordt uitgevoerd binnen WP-CLI, terwijl de wachtrij zelf rechtstreeks wordt opgehaald uit MySQL via de best beschikbare DB-driver Queuety.

De plugin installeert ook een standaardfallback WordPress cron die eenmaal per minuut wordt uitgevoerd voor omgevingen zonder shelltoegang. Op deze pagina worden speciale WP-CLI-workers besproken, de aanbevolen implementatiemodus voor een hogere doorvoer.

Een worker starten

wp queuety work

Hiermee start u één worker op de wachtrij default. Deze draait in een lus: een taak ophalen, uitvoeren en herhalen. Als de wachtrij leeg is, wacht de worker QUEUETY_WORKER_SLEEP seconden (standaard: 1) voordat deze opnieuw controleert.

Opties

# 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

Prioriteiten voor meerdere wachtrijen

Eén worker kan meerdere wachtrijen in prioriteitsvolgorde verwerken. Geef een kommagescheiden lijst met wachtrijnamen door aan --queue:

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

De worker probeert een taak op te halen uit elke wachtrij in de opgegeven volgorde. Eerst controleert deze critical. Als critical een taak bevat, verwerkt de worker die taak. Als critical leeg is, gaat de worker verder met default en daarna met low. Dit betekent dat wachtrijen met een hogere prioriteit altijd worden afgehandeld voordat wachtrijen met een lagere prioriteit aan de beurt komen.

Hoe de prioriteitsvolgorde werkt

Bij elke iteratie probeert de worker een taak op te halen uit de eerste wachtrij in de lijst. Als die wachtrij leeg of gepauzeerd is, gaat de worker verder met de volgende wachtrij en zo door totdat er een taak is gevonden of de lijst is uitgeput. Als nergens een taak wordt gevonden, wacht de worker en probeert het opnieuw. Dit is een strikt prioriteitsmodel: een constante stroom taken critical laat default en low zonder werk achter. Als wachtrijen met een lagere prioriteit gegarandeerde doorvoer nodig hebben, gebruik dan speciale workers.

Meerdere wachtrijen gebruiken in PHP

De methoden Worker::run() en Worker::flush() accepteren een string of een 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' );

Implementatiestrategieën

Speciale workers per wachtrij geven u volledige controle over gelijktijdigheid en isolatie. Elke wachtrij krijgt een eigen worker-process(es) en kan niet worden uitgehongerd door andere wachtrijen:

[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

Prioriteitsvolgorde met één worker is eenvoudiger te beheren en werkt goed wanneer wachtrijen met een lagere prioriteit vertraging kunnen verdragen:

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

U kunt beide aanpakken ook combineren: speciale workers voor critical om de doorvoer te garanderen en een worker met prioriteitsvolgorde voor al het overige:

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

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

Toelating op basis van resources

Workers doen meer dan alleen de volgende wachtende rij ophalen. Voordat een opgehaalde taak start, kan Queuety globale toelatingscontroles uitvoeren op basis van de gelijktijdigheidsgroep van de taak en het recente uitvoeringsprofiel van die handler.

Zie Resourcebeheer voor het bredere model en hoe dit samenhangt met taakmetagegevens en workflowbudgetten.

Een gelijktijdigheidsgroep geeft meerdere handlers of taakklassen een gedeelde limiet. Als u verschillende zware providers dezelfde groep en limiet geeft, geeft een worker nieuw opgehaald werk vrij in plaats van het te starten zodra de groep verzadigd is. Zo voorkomen onafhankelijke wachtrijen dat ze tegelijk hetzelfde downstreamsysteem overbelasten, alleen omdat er meerdere workers niets te doen hebben.

Recente uitvoeringslogs leveren ook gegevens voor een eenvoudig toelatingsprofiel. Wanneer toelating op basis van resources is ingeschakeld, bekijkt Queuety de recente gemiddelde duur en het piekgeheugen voor de handler die wordt uitgevoerd. Als de huidige worker al dicht bij de ingestelde geheugenlimiet zit, een wachtrij- of providergroep het gewogen budget al heeft bereikt, de huidige container onvoldoende resterend geheugen lijkt te hebben voor het waargenomen profiel van die handler, of een uitvoering in de stijl van --once dicht bij de tijdslimiet zit, stelt Queuety de taak uit in plaats van werk te starten dat waarschijnlijk niet netjes wordt afgerond.

U schakelt dit beleid in vanuit de werkdefinitie, niet vanuit de workeropdracht:

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

Dezelfde metagegevens kunnen worden toegevoegd aan dispatchbare taakklassen met publieke eigenschappen of aan een afzonderlijke dispatch via PendingJob::concurrency_group() en PendingJob::cost_units().

Het toelatingsprofiel aan de workerzijde wordt geregeld met deze constanten:

Levenscyclus van workers

Een worker stopt op een nette manier nadat de huidige taak is voltooid, zodra aan een van deze voorwaarden wordt voldaan:

Na het stoppen wordt de worker netjes afgesloten. In productie start een procesbeheerder deze automatisch opnieuw.

Signalen worden nog niet netjes afgehandeld (#4). Eén wp queuety work-proces installeert geen signaalhandler, dus SIGTERM of SIGINT (Ctrl+C) beëindigt het onmiddellijk, zelfs midden in een taak. Die taak blijft processing totdat herstel van verouderde taken deze na QUEUETY_STALE_TIMEOUT seconden opnieuw instelt. Zorg er dus voor dat handlers veilig opnieuw kunnen worden uitgevoerd. Een workerpool stuurt SIGTERM door naar de onderliggende processen, maar die reageren er niet op. Daarom blijft de pool draaien totdat ze stoppen om een van de bovenstaande redenen. Als u een pool wilt stoppen, stelt u een stoptime-out in bij uw procesbeheerder, waarna deze SIGKILL verstuurt.

Gelijktijdigheid met meerdere workers

De vlag --workers=N start N onderliggende processen met pcntl_fork(). Elk onderliggend proces maakt een eigen databaseverbinding aan en verwerkt taken zelfstandig. Het bovenliggende proces houdt de onderliggende processen in de gaten en start processen die crashen opnieuw.

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

Vereisten: De extensies pcntl en posix PHP moeten zijn geïnstalleerd (zonder pcntl mislukt de opdracht met "pcntl extension is required for --workers=N."), en elk onderliggend proces gebruikt een eigen databaseverbinding. Uw instelling MySQL max_connections moet daarom geschikt zijn voor de pool. Voor adaptieve pools (--min-workers/--max-workers) gelden dezelfde vereisten.

Adaptieve workerpools

Adaptieve pools houden een minimumaantal workers paraat en schalen vervolgens binnen een ingesteld bereik op en af:

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

Het bovenliggende proces controleert de opvraagbare achterstand in de geselecteerde wachtrijen en past het aantal onderliggende processen aan die achterstand aan, tot aan het ingestelde maximum. Als de wachtrij rustig wordt, wacht de pool de ingestelde respijtperiode voor inactiviteit af voordat deze weer wordt verkleind. Zo worden de processen bij korte pieken niet meteen voortdurend gestart en gestopt.

Opschalen gebeurt nog steeds voorzichtig. Wanneer Queuety gegevens over container- of hostgeheugen kan uitlezen, worden er geen onderliggende processen toegevoegd als er niet genoeg geheugen over is voor een volledige extra worker.

De schaalcyclus aan de bovenliggende zijde wordt geregeld met:

Herstel van verouderde taken

Als een worker crasht tijdens het verwerken van een taak, blijft die taak in de status processing staan. Een langlopende worker controleert ongeveer eenmaal per minuut op taken die langer dan QUEUETY_STALE_TIMEOUT seconden in de status processing staan. Een verouderde taak met resterende pogingen wordt opnieuw geprobeerd na de wachttijd; een taak waarvan alle pogingen zijn verbruikt, wordt in de wacht gezet en de bijbehorende workflow wordt als mislukt gemarkeerd. --once-workers en de fallback WP-Cron voeren deze controle niet uit.

Herstel verouderde taken handmatig:

wp queuety recover

Flushmodus

Verwerk alle wachtende taken en sluit af. Handig voor tests of eenmalige batchverwerking:

wp queuety flush
wp queuety flush --queue=emails

Implementatie in productie

Gebruik in productie een procesbeheerder om workers actief te houden. Dit zijn twee veelgebruikte opties.

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

Inschakelen en starten:

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

Implementatietips

Voer aparte workers uit voor aparte wachtrijen wanneer u sterkere isolatie nodig hebt. Gebruik --workers=N voor CPU-intensieve workloads en geef de voorkeur aan afzonderlijk bewaakte workers wanneer de workload vooral uit I/O bestaat. Stel QUEUETY_WORKER_MAX_JOBS in als u periodiek opnieuw wilt opstarten om geheugen vrij te maken, en houd het aantal buried in de gaten via webhooks of statistieken.