ミドルウェア

Markdownとして表示 GitHubで編集ドキュメント: 2cc7b2b

ミドルウェアは、玉ねぎ型のパイプラインでジョブの実行をラップします。各ミドルウェアは、ジョブの前後に処理を実行したり、実行を途中で打ち切ったり、例外を捕捉したりできます。Queuetyには組み込みミドルウェアが5つあり、Contracts\Middlewareインターフェースを使ってカスタムミドルウェアも追加できます。

ミドルウェアは、Contracts\Jobを実装し、Dispatchableトレイトを使うディスパッチ可能なジョブクラスで利用できます。

`Contracts\Middleware`インターフェース

PHP

use Queuety\Contracts\Middleware;

interface Middleware {
    public function handle( object $job, \Closure $next ): void;
}

$jobパラメーターは、処理中のジョブインスタンスです。$next( $job )を呼び出すと、パイプライン内の次のミドルウェアに実行を渡します（最後のミドルウェアの場合は、ジョブのhandle()メソッドに渡します）。

ジョブへのミドルウェアの追加

ジョブクラスにmiddleware()メソッドを定義し、ミドルウェアインスタンスの配列を返します。

PHP

use Queuety\Contracts\Job;
use Queuety\Dispatchable;
use Queuety\Middleware\RateLimited;
use Queuety\Middleware\Timeout;

class SyncInventoryJob implements Job {
    use Dispatchable;

    public function __construct(
        public readonly string $sku,
    ) {}

    public function handle(): void {
        // Sync inventory for this SKU...
    }

    public function middleware(): array {
        return [
            new Timeout( seconds: 30 ),
            new RateLimited( max: 100, window: 60 ),
        ];
    }
}

パイプラインの実行順序

ミドルウェアは配列の順に適用され、最初のミドルウェアが最外層になります。middleware()メソッドが[A, B, C]を返す場合、実行順序は次のとおりです。

コード

A::handle() -> B::handle() -> C::handle() -> $job->handle()

各ミドルウェアは次のミドルウェアをラップします。Aはジョブの前に最初に実行され、ジョブの後に最後に実行されるため、時間計測やtry/catchラッパーに適しています。

カスタムミドルウェアの作成

Contracts\Middlewareインターフェースを実装します。

PHP

use Queuety\Contracts\Middleware;

class LogExecution implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        $class = get_class( $job );
        error_log( "[Queuety] Starting: {$class}" );

        $start = microtime( true );

        try {
            $next( $job );
        } finally {
            $elapsed = round( microtime( true ) - $start, 3 );
            error_log( "[Queuety] Finished: {$class} in {$elapsed}s" );
        }
    }
}

ジョブで使用します。

PHP

public function middleware(): array {
    return [
        new LogExecution(),
    ];
}

実行の途中終了

$next()を呼び出さなければ、ミドルウェアでジョブ全体をスキップできます。

PHP

class SkipOnMaintenanceMode implements Middleware {
    public function handle( object $job, \Closure $next ): void {
        if ( wp_is_maintenance_mode() ) {
            return; // Job silently skipped.
        }

        $next( $job );
    }
}

組み込みミドルウェア

`ThrottlesExceptions`

ジョブハンドラーで発生する例外を抑制し、ジョブの大量発生を防ぎます。一定時間内にハンドラーが例外を何度もスローした場合、そのハンドラーの後続ジョブは処理されず、代わりにRuntimeExceptionが返されます。これにより、障害が発生した外部サービスが多数のジョブのリトライ回数を一度に使い切るのを防ぎます。

PHP

use Queuety\Middleware\ThrottlesExceptions;

public function middleware(): array {
    return [
        new ThrottlesExceptions( max_attempts: 10, decay_minutes: 10 ),
    ];
}

パラメーター	型	デフォルト	説明
`$max_attempts`	`int`	`10`	減衰ウィンドウ内で許可される例外の最大数
`$decay_minutes`	`int`	`10`	ウィンドウの時間（分）

例外数は、Queuetyのロックテーブルでハンドラークラスごとに追跡され、すべてのワーカープロセスで共有されます。しきい値に達すると、ミドルウェアはジョブを呼び出さずにRuntimeExceptionをスローします。ワーカーはこれを通常の失敗と同様に扱います。試行回数はmax_attemptsに加算され、通常のバックオフで再試行され、試行回数を使い切ると埋没状態になります。

例：外部APIが停止した場合、最初の10回の失敗は通常どおり処理され（例外カウンターが毎回増加）、その後はAPIを呼び出さずに即座に失敗します。新たな例外が発生せずに10分経過すると、再び試行できるようになります。記録された例外ごとにウィンドウがリセットされます。

`RateLimited`

時間枠内の実行回数に上限を設けます。ミドルウェア内で上限を超えるとRateLimitExceededExceptionがスローされ、ワーカーは通常の失敗試行として扱い、通常のバックオフで再試行し、max_attemptsに加算します。ミドルウェアがワーカーのレートリミッターに上限を登録すると、上限に達している間は、そのハンドラーの後続ジョブも実行前にpendingへ戻されます（レート制限を参照）。

PHP

use Queuety\Middleware\RateLimited;

public function middleware(): array {
    return [
        new RateLimited( max: 60, window: 60 ), // 60 per minute
    ];
}

パラメーター	型	説明
`$max`	`int`	ウィンドウ内で許可される実行の最大数
`$window`	`int`	ウィンドウの時間（秒）

レートリミッターはQueuetyの内部RateLimiterインスタンスを使うため、上限はすべてのワーカープロセスに適用されます。

`Timeout`

pcntl_alarmでジョブごとの実行タイムアウトを設定し、ジョブの実行時間が長すぎる場合にTimeoutExceptionをスローすることを意図しています。現在、タイムアウトは適用されません（#10）。Queuetyでは非同期シグナルが有効にならないため、ジョブの実行中にアラームハンドラーが動作しません。また、このミドルウェアは終了時にアラームをリセットし、ワーカー独自のジョブタイムアウトアラームもキャンセルします。

PHP

use Queuety\Middleware\Timeout;

public function middleware(): array {
    return [
        new Timeout( seconds: 30 ),
    ];
}

パラメーター	型	説明
`$seconds`	`int`	最大実行時間（秒）

pcntlPHP拡張機能が必要です。pcntlが利用できない場合、ミドルウェアは何もせず処理を続行します（タイムアウトは適用されません）。

`UniqueJob`

同じキーを持つジョブの同時実行を防ぎます。データベースのロックテーブルを使い、一度に1つのインスタンスだけが実行されるようにします。すでにロックされている場合、ジョブは通知なしにスキップされ、再試行されません。

PHP

use Queuety\Middleware\UniqueJob;

public function middleware(): array {
    return [
        new UniqueJob(),                         // Lock key defaults to the job class name
        new UniqueJob( key: 'sync_inventory' ),   // Custom lock key
    ];
}

パラメーター	型	デフォルト	説明
`$key`	`string\|null`	`null`（クラス名を使用）	一意性を保つためのロックキー

`WithoutOverlapping`

論理キーを共有するジョブの重複実行を防ぎ、クラッシュしたワーカーによるデッドロックを防ぐため、ロックを自動的に期限切れにします。UniqueJobと同様ですが、タイムアウト後にロックを自動解除する必要がある長時間処理向けに設計されています。

PHP

use Queuety\Middleware\WithoutOverlapping;

public function middleware(): array {
    return [
        new WithoutOverlapping( key: 'import_products', release_after: 600 ),
    ];
}

パラメーター	型	デフォルト	説明
`$key`	`string`	（必須）	論理操作キー
`$release_after`	`int`	`300`	自動解除までのロック最大時間（秒）

すでにロックされている場合、ジョブは通知なしにスキップされます。期限切れのロックは、取得を試みるたびに自動でクリーンアップされます。