ジョブ
ジョブはQueuetyにおける作業の基本単位です。ハンドラー名とペイロードを指定してジョブをディスパッチすると、ワーカーが取得してバックグラウンドで実行します。
ジョブのディスパッチ
use Queuety\Queuety;
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] );
dispatch()はPendingJobビルダーを返します。自動ディスパッチされる前に、オプションをチェーンできます:
use Queuety\Enums\Priority;
Queuety::dispatch( 'send_email', [ 'to' => 'user@example.com' ] )
->on_queue( 'emails' )
->with_priority( Priority::High )
->delay( 300 )
->max_attempts( 5 );
ジョブIDをすぐに取得するには、->id()を呼び出します:
$job_id = Queuety::dispatch( 'send_email', $payload )->id();
ハンドラーの作成
ハンドラーとは、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',
];
}
}
handle()メソッドはジョブのペイロードを受け取り、処理を実行します。config()メソッドは省略可能なデフォルト値を返します:
| キー | 型 | 説明 |
|---|---|---|
queue |
string |
デフォルトのキュー名 |
max_attempts |
int |
ジョブが埋没状態になるまでの最大実行回数。最初の実行も含みます |
backoff |
string |
再試行のバックオフ戦略: exponential、linear、またはfixed |
rate_limit |
array |
[max_executions, window_seconds]として指定するレート制限 |
concurrency_group |
string |
コストの高いハンドラーで共有するワーカー受け入れグループ |
concurrency_limit |
int |
そのグループ内で同時に処理できるジョブの最大数 |
cost_units |
int |
重み付きキューやプロバイダー予算、ワークフロー予算、受け入れ制御の判定で使う相対的な実行コスト |
これらのリソースキーは、Queuetyのより広範なリソース管理モデルの一部であり、ジョブメタデータ、ワークフロー予算、ワーカーの受け入れ制御が連携します。
ハンドラーを登録します:
Queuety::register( 'send_email', SendEmailHandler::class );
または、PHP属性を使って自動登録できます。
優先度
ジョブには4段階の優先度があります。ワーカーは優先度の高いジョブから取得します。
use Queuety\Enums\Priority;
Queuety::dispatch( 'handler', $payload )->with_priority( Priority::Urgent );
| レベル | 値 | 説明 |
|---|---|---|
Priority::Low |
0 |
デフォルト。後回しにできるバックグラウンド処理。 |
Priority::Normal |
1 |
標準の優先度。 |
Priority::High |
2 |
低および通常より先に処理されます。 |
Priority::Urgent |
3 |
最初に処理されます。使用は控えめにしてください。 |
遅延
指定した秒数が経過した後にのみ実行可能になるよう、ジョブを遅延させます:
Queuety::dispatch( 'send_reminder', $payload )->delay( 3600 ); // 1 hour
ジョブはすぐに挿入されますが、遅延時間が経過するまでワーカーは取得しません。
一意なジョブ
同じハンドラーとペイロードによる重複ディスパッチを防ぎます。同じハンドラーとペイロードを持つ保留中または処理中のジョブがすでに存在する場合、新しいジョブは作成せず、既存のジョブIDを返します。
Queuety::dispatch( 'sync_inventory', [ 'sku' => 'ABC-123' ] )->unique();
ジョブの依存関係
ジョブは別のジョブに依存させることができます。依存先のジョブが完了するまで、ワーカーは取得しません。
$first_id = Queuety::dispatch( 'import_data', $payload )->id();
Queuety::dispatch( 'send_notification', [ 'import_id' => $first_id ] )
->after( $first_id );
バッチディスパッチ
スループットを最大化するため、1回の複数行INSERTで複数のジョブをディスパッチします:
$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' ] ],
] );
各項目には、queue、priority、delay、max_attemptsの省略可能なキーを含められます。
再試行とバックオフ
ジョブが例外をスローすると、Queuetyはpendingに戻し、遅延付きのavailable_atを設定します。合計でmax_attempts回実行されると終了します(最初の実行も数えるため、max_attempts( 3 )は最大2回の再試行を意味します)。各再試行までの遅延はバックオフ戦略で決まり、attemptは直前に失敗した実行の回数を表します:
| 戦略 | 遅延の計算式 | 例(実行1、2、3の後) |
|---|---|---|
exponential |
2^attempt * 30秒、上限3600秒 |
60s、120s、240s |
linear |
attempt * 60秒 |
60s、120s、180s |
fixed |
60秒 |
60s、60s、60s |
全体のデフォルト値はwp-config.phpで設定します:
define( 'QUEUETY_RETRY_BACKOFF', 'linear' );
またはconfig()でハンドラーごとに指定します:
public function config(): array {
return [ 'backoff' => 'exponential', 'max_attempts' => 5 ];
}
またはディスパッチごとに指定します:
Queuety::dispatch( 'handler', $payload )->max_attempts( 10 );
埋没ジョブ
すべての再試行を使い切ると、ジョブは埋没としてマークされます。埋没したジョブは確認用にデータベースに残りますが、処理されません。手動で再試行できます。手動再試行では、試行回数をリセットせずにジョブを再びpendingにするため、再試行された埋没ジョブはもう一度実行され、その実行に失敗すると再び埋没します:
// Retry a specific buried job
Queuety::retry( $job_id );
// Retry all buried jobs
$count = Queuety::retry_buried();
またはCLI経由で指定します:
wp queuety retry 42
wp queuety retry-buried
タイムアウト
Queuetyは最大実行時間に対するpcntl_alarm()を設定し、超過時にTimeoutExceptionで試行を失敗させることを意図しています。現在、このタイムアウトは適用されません(#10): ワーカーは非同期シグナルを有効にしないため、長時間実行されるハンドラーは最後まで実行され、sleep()でブロックされたハンドラーはアラーム発生時に中断され、その後完了済みとしてマークされます。タイムアウトをグローバルに設定します:
define( 'QUEUETY_MAX_EXECUTION_TIME', 300 ); // 5 minutes
processing状態のままQUEUETY_STALE_TIMEOUT秒を超えたジョブは、自動的に復旧されます:
define( 'QUEUETY_STALE_TIMEOUT', 600 ); // 10 minutes
古いジョブを手動で復旧するには:
wp queuety recover
キュー
ジョブは名前付きキューにディスパッチされます。各ワーカーは1つのキュー、または順序付きのキューリストを処理します(Workersを参照):
Queuety::dispatch( 'handler', $payload )->on_queue( 'emails' );
wp queuety work --queue=emails
デフォルトのキューはdefaultです。ワークロードを分離するために、キューごとに別々のワーカープロセスを実行できます。
レート制限
一定時間内にハンドラーを実行できる頻度を制限します:
Queuety::dispatch( 'call_api', $payload )
->rate_limit( 10, 60 ); // max 10 executions per 60 seconds
詳しくはRate Limitingを参照してください。
ディスパッチ可能なジョブクラス
Queuety v0.6.0では、クラスベースのジョブAPIが導入されました。ハンドラー名、ペイロード、ハンドラークラスを分ける代わりに、型付きのパブリックプロパティとしてのペイロード、実行ロジック、オプションのミドルウェアをすべて1つにまとめたクラスを定義します。
Contracts\Jobインターフェース
use Queuety\Contracts\Job;
interface Job {
public function handle(): void;
}
Jobクラスはペイロードをパブリックプロパティとして保持し、handle()で独自のロジックを実行します。別個のconfig()メソッドやハンドラー登録は必要ありません。
Dispatchableトレイト
Dispatchableトレイトを追加すると、ジョブクラスに静的なdispatch()メソッドを追加できます:
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 );
}
}
コンストラクター引数を渡してディスパッチします:
SendEmailJob::dispatch( 'user@example.com', 'Welcome!', 'Hello there.' )
->on_queue( 'emails' )
->with_priority( Priority::High );
Dispatchableトレイトは内部でQueuety::dispatch()を呼び出すため、すべてのPendingJobオプション(queue、priority、delay、max_attempts、unique、after)は同じように機能します。
シリアライズの仕組み
Jobインスタンスをディスパッチすると、QueuetyのJobSerializerが完全修飾クラス名をハンドラーとして取り出し、すべてのパブリックプロパティをペイロードとして取得します。ワーカー側では、シリアライザーがペイロードのキーとコンストラクターパラメーター名を照合してインスタンスを再構築します。バックドEnumは基になる値にシリアライズされ、自動的に復元されます。
ジョブのプロパティ
ジョブクラスでは、再試行やタイムアウトの動作を制御するパブリックプロパティを宣言できます。ワーカーはリフレクションでこれらを読み取り、自動的に適用します。
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...
}
}
| プロパティ | 型 | 説明 |
|---|---|---|
$tries |
int |
ジョブが埋没するまでの最大試行回数。ディスパッチ時のmax_attempts値を上書きします。 |
$timeout |
int |
最大実行時間(秒)。グローバルなQUEUETY_MAX_EXECUTION_TIME定数を上書きします。pcntlが必要で、現在は適用されません(Timeoutを参照)。 |
$backoff |
array |
各再試行に適用する遅延時間(秒)の配列。最後の値は、それ以降の再試行でも再利用されます。 |
$concurrency_group |
string |
コストの高いジョブに共通するワーカー受け入れグループ。 |
$concurrency_limit |
int |
グループ内で同時実行できるprocessingジョブの最大数。$concurrency_groupがない場合は無視されます。 |
$cost_units |
int |
ワークフロー予算とワーカー受け入れで使用される相対的な実行コスト。 |
$backoff配列を使うと、再試行の遅延を段階的に増やせます。たとえば[10, 30, 60, 120]の場合、1回目の再試行は10秒後、2回目は30秒後、3回目は60秒後に行われ、それ以降の再試行は120秒後になります。
ジョブクラスが異なるコスト特性を表す場合、リソース関連のプロパティが役立ちます。高速なキャッシュ更新も大規模なプロバイダー呼び出しも、どちらも「1つのジョブ」ですが、ワークフロー予算やワーカー受け入れの判断でプロセスを安全な範囲内に保つ際、同じコストとして数えるべきとは限りません。
ワークフロー予算やワーカー側の受け入れを含む全体モデルについては、Resource Managementを参照してください。
失敗時フック
すべての再試行を使い切ってジョブが埋没すると、Queuetyはジョブインスタンスにfailed()メソッドが定義されている場合、それを呼び出します。クリーンアップ、アラート、ログ記録に使用できます。
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()}" );
}
}
failed()メソッドには、最終的な失敗の原因となった例外が渡されます。ジョブインスタンスは元のペイロードでデシリアライズされるため、すべてのパブリックプロパティを利用できます。
条件付きディスパッチ
Dispatchableトレイトは、条件付きディスパッチ用にdispatch_if()とdispatch_unless()を提供します。どちらも最初の引数に真偽値の条件を取り、その後にジョブのコンストラクター引数を受け取ります。
// 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 );
条件がtrueの場合、dispatch_if()はディスパッチを行い、PendingJobを返します。条件がfalseの場合はnullを返します。
条件がfalseの場合、dispatch_unless()はディスパッチを行い、PendingJobを返します。条件がtrueの場合はnullを返します。
ディスパッチ時には、どちらのメソッドでもPendingJobオプションをチェーンできます:
SendEmailJob::dispatch_if( $should_send, $to, $subject, $body )
?->on_queue( 'emails' )
?->with_priority( Priority::High );
同期ディスパッチ
Queuety::dispatch_sync()を使うと、キューに追加せずに現在のプロセス内でジョブをすぐに実行できます。テストや、インラインで実行する必要があるジョブに便利です。
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' ] );
ジョブのhandle()メソッドを直接呼び出します。データベース行は作成されず、ワーカーも使用されず、ミドルウェアも適用されません。
ジョブチェーン
ジョブチェーンは、一連のジョブを順番に実行します。チェーン内の各ジョブは、直前のジョブが正常に完了することを条件とします。いずれかのジョブが失敗すると、後続のジョブは保留状態のままになります。
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()は、キューと失敗ハンドラーを設定する流れるようなメソッドを備えたChainBuilderを返します。dispatch()メソッドは、連続するdepends_on関係を持つすべてのジョブを作成し、最初のジョブのIDを返します。
with_chain()を使う
Dispatchableトレイトはwith_chain()メソッドを提供します。現在、これはQueuety::chain( $jobs )と同等です。呼び出し元のクラスはチェーンに追加されないため(#13)、配列に最初のジョブも含めてください:
FetchDataJob::with_chain( [
new FetchDataJob(),
new ProcessDataJob(),
new NotifyCompleteJob(),
] )
->on_queue( 'pipeline' )
->dispatch();
チェーン失敗の処理
catch()を使って失敗ハンドラークラスを登録します。ハンドラークラスにはhandle()メソッドが必要です。チェーン内のいずれかのジョブが恒久的に埋没したときに呼び出されます。ワーカーはhandle()、handle( Job $job )、handle( Job $job, \Throwable $exception )をサポートします。
class ChainFailedHandler {
public function handle(): void {
// Alert, clean up, etc.
}
}
ミドルウェアの追加
ジョブクラスでは、実行をラップするミドルウェアを返すmiddleware()メソッドを定義できます:
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 ),
];
}
}
組み込みミドルウェアやカスタムミドルウェアの作成については、Middlewareを参照してください。
ハンドラーベースのジョブとの比較
| ハンドラーベース(v0.1+) | クラスベース(v0.6+) | |
|---|---|---|
| ディスパッチ | Queuety::dispatch( 'send_email', $payload ) |
SendEmailJob::dispatch( $to, $subject ) |
| ハンドラー | Handlerを実装する別クラス |
同じクラスがContracts\Jobを実装 |
| ペイロード | 型指定なしのarray $payload |
型指定されたパブリックプロパティ |
| 登録 | Queuety::register()または#[QueuetyHandler] |
不要(FQCNを使用) |
| ミドルウェア | 利用不可 | オプションのmiddleware()メソッド |
| 設定 | config()メソッド |
ジョブプロパティ + PendingJobチェーン |
どちらの方式も完全にサポートされています。既存のハンドラーベースのジョブは変更せずに引き続き動作します。同じプロジェクト内で両方のスタイルを組み合わせることもできます。
ジョブのライフサイクル
dispatch() -> pending -> processing -> completed
| |
| +-> exception -> pending (retry after backoff)
| |
| +-> buried (max attempts reached)
|
+-> delayed (available_at in the future)
| 状態 | 説明 |
|---|---|
pending |
ワーカーによる取得待ち |
processing |
実行中 |
completed |
正常に完了 |
failed |
予約済み。現在、ワーカーはこの状態を設定しません。再試行される失敗試行はpendingに戻ります |
buried |
すべての再試行を使い切った状態 |