ワーカー

ワーカーは長時間稼働するプロセスで、保留中のジョブをデータベースから取得して実行し、ワークフローを次へ進めます。wp queuety workコマンドはWP-CLI内で実行され、キュー自体は利用可能な最適なQueuety DBドライバーを通じてMySQLから直接確保されます。

このプラグインは、シェルアクセスのない環境向けに、1分ごとに実行するデフォルトのWordPress cronフォールバックもインストールします。このページでは、より高いスループットを実現する推奨デプロイ方式である、専用WP-CLIワーカーを説明します。

ワーカーの起動

wp queuety work

これにより、defaultキューで単一のワーカーが起動します。ジョブを確保して実行する処理を繰り返すループで動作します。キューが空の場合はQUEUETY_WORKER_SLEEP秒（デフォルトは1秒）待ってから、再び確認します。

オプション

# 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

複数キューの優先順位

1つのワーカーで複数のキューを優先順位順に処理できます。キュー名をコンマ区切りで--queueに渡します。

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

ワーカーは、指定された順に各キューからジョブの確保を試みます。まずcriticalを確認します。criticalにジョブがあれば、そのジョブを処理します。criticalが空の場合はdefaultに進み、その後lowへ進みます。つまり、優先度の高いキューが常に先に処理されます。

優先順位の仕組み

各反復で、ワーカーはリストの先頭のキューからジョブの確保を試みます。そのキューが空または一時停止中の場合は次のキューへ進み、ジョブを見つけるかリストを最後まで確認するまで続けます。どのキューにもジョブがなければ、ワーカーは待機して再試行します。これは厳密な優先順位モデルです。そのため、criticalジョブが継続的に流入すると、defaultとlowが処理されない状態になることがあります。優先度の低いキューにも処理量を保証する必要がある場合は、専用ワーカーを使ってください。

PHPで複数キューを使う

Worker::run()メソッドとWorker::flush()メソッドは、文字列または配列を受け取ります。

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

デプロイ戦略

キューごとの専用ワーカーを使うと、並行処理数と分離を完全に制御できます。各キューに専用のワーカーprocess(es)が割り当てられるため、他のキューに処理を奪われません。

[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

1つのワーカーによる優先順位付けは管理が簡単で、優先度の低いキューで遅延が許容される場合に適しています。

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

両方の方法を組み合わせることもできます。criticalにはスループットを保証する専用ワーカーを使い、それ以外には優先順位を付けたワーカーを使います。

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

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

リソースを考慮した受付

ワーカーは、次の保留中の行を取得するだけではありません。取得したジョブを開始する前に、Queuety はジョブの同時実行グループと、そのハンドラーの最近の実行プロファイルに基づいて、大まかな受付チェックを適用できます。

ジョブのメタデータやワークフローの予算との関係を含む全体像については、リソース管理を参照してください。

同時実行グループを使うと、複数のハンドラーやジョブクラスに共通の上限を設定できます。負荷の高いプロバイダーに同じグループと上限を設定すると、グループが上限に達したとき、ワーカーは新たに取得したジョブを開始せずに解放します。これにより、アイドル状態のワーカーが複数いても、独立したキューが同じ下流システムに一斉に殺到するのを防げます。

最近の実行ログは、簡易的な受付プロファイルにも反映されます。リソース受付が有効な場合、Queuety は実行しようとしているハンドラーの最近の平均所要時間とピークメモリを確認します。現在のワーカーが設定済みメモリ上限にすでに近い場合、キューまたはプロバイダーグループが重み付け予算の上限に達している場合、現在のコンテナーにそのハンドラーの実測プロファイルに必要な空きメモリがないと思われる場合、または --once 形式の実行が時間枠の上限に近い場合、Queuety は正常に完了しそうにないジョブを開始せず、実行を延期します。

このポリシーは、ワーカーコマンドではなく、作業定義から有効にします。

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

同じメタデータは、公開プロパティを持つディスパッチ可能なジョブクラス、または PendingJob::concurrency_group() と PendingJob::cost_units() を使った個別のディスパッチにも設定できます。

ワーカー側の受付プロファイルは、次の定数で制御します。

ワーカーのライフサイクル

次のいずれかの条件を満たすと、ワーカーは現在のジョブを完了してから正常に停止します。

停止後、ワーカーは正常に終了します。本番環境では、プロセスマネージャーが自動的に再起動します。

シグナルはまだ正常に処理されません（#4）。単一の wp queuety work プロセスにはシグナルハンドラーが登録されていないため、ジョブの実行中であっても SIGTERM または SIGINT（Ctrl+C）を受けると即座に終了します。そのジョブは processing のまま残り、QUEUETY_STALE_TIMEOUT 秒後に古いジョブの復旧処理でリセットされるため、ハンドラーは安全に再試行できるようにしてください。ワーカープールは SIGTERM を子プロセスに転送しますが、子プロセスはそれに対応しないため、上記のいずれかの理由で子プロセスが停止するまでプールは動作し続けます。プールを停止するには、プロセスマネージャーに停止タイムアウトを設定し、その後に SIGKILL を送信させてください。

マルチワーカーの同時実行

--workers=N フラグは pcntl_fork() を使って N 個の子プロセスをフォークします。各子プロセスは独自のデータベース接続を作成し、個別にジョブを処理します。親プロセスは子プロセスを監視し、クラッシュした子プロセスを再起動します。

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

要件: pcntl と posix PHP 拡張機能がインストールされている必要があります（pcntl がない場合、コマンドは「pcntl extension is required for --workers=N.」というエラーで失敗します）。各子プロセスは独自のデータベース接続を使用するため、MySQL の max_connections 設定はプールに対応できる値にしてください。アダプティブプール（--min-workers/--max-workers）にも同じ要件が適用されます。

アダプティブワーカープール

アダプティブプールは、最小数のワーカーを常時稼働させ、設定した範囲内で数を増減します。

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

親プロセスは、選択したキュー群で処理可能なバックログを監視し、設定済みの最大数を上限として、バックログに合わせて子プロセス数を調整します。キューが落ち着くと、プールはアイドル猶予期間が過ぎるまで待ってから縮小するため、短い負荷の急増でプロセス一覧がすぐに頻繁に変わることはありません。

スケールアップは引き続き慎重に行われます。Queuety がコンテナーまたはホストのメモリ情報を取得できる場合、次のワーカーに必要なメモリを確保できなければ、子プロセスの追加を見送ります。

親プロセス側のスケーリングループは、次の項目で制御します。

古いジョブの復旧

ジョブの処理中にワーカーがクラッシュすると、そのジョブは processing 状態のまま残ります。長時間稼働するワーカーは、processing 状態が QUEUETY_STALE_TIMEOUT 秒を超えたジョブがないか、約1分ごとに確認します。試行回数が残っている古いジョブはバックオフ時間の経過後に再試行され、すべての試行を使い切ったジョブは隔離され、ワークフローは失敗としてマークされます。--once ワーカーと WP-Cron のフォールバックでは、この確認は行われません。

古いジョブを手動で復旧するには:

wp queuety recover

フラッシュモード

保留中のジョブをすべて処理して終了します。テストや一度限りのバッチ処理に便利です。

wp queuety flush
wp queuety flush --queue=emails

本番環境へのデプロイ

本番環境では、プロセスマネージャーを使ってワーカーを稼働させ続けます。一般的な方法を2つ紹介します。

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

有効化して起動します:

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

デプロイのヒント

より強い分離が必要な場合は、キューごとにワーカーを分けて実行します。CPU負荷が高いワークロードには --workers=N を使い、主にI/Oを行うワークロードには個別に監視するワーカーを推奨します。メモリを解放するため定期的に再起動する場合は QUEUETY_WORKER_MAX_JOBS を設定し、webhooks または metrics で buried の数を確認してください。