ハートビート

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

長時間実行するジョブやストリーミングステップは、ハートビートを送信して、処理が継続中であることを示せます。ハートビートがないと、reserved_atタイムスタンプが更新されていないため、まだ処理中のジョブが古いジョブの検出機能によって再取得されることがあります。

仕組み

ワーカーはジョブの処理を開始すると、Heartbeat::init()を呼び出して現在のジョブコンテキストを設定します。ハンドラー内でHeartbeat::beat()を呼び出すと、ジョブのreserved_at列がNOW()に更新されます。古いジョブの検出機能はreserved_atを使ってジョブが停止しているか判断するため、これを更新すればジョブは安全に処理されます。

ジョブが終了すると、ワーカーはHeartbeat::clear()を呼び出してコンテキストをリセットします。

ハートビートの送信

任意のジョブハンドラーまたはストリーミングステップ内からHeartbeat::beat()を呼び出します:

PHP

use Queuety\Heartbeat;

class LongRunningStep implements \Queuety\Step {

    public function handle( array $state ): array {
        $items = $state['items'];

        foreach ( $items as $i => $item ) {
            $this->process( $item );
            Heartbeat::beat( [ 'processed' => $i + 1, 'total' => count( $items ) ] );
        }

        return [ 'done' => true ];
    }

    public function config(): array {
        return [];
    }

    private function process( $item ): void {
        // Expensive work here.
    }
}

進捗データ

$progressパラメーターは任意の連想配列です。指定するとJSONエンコードされ、ジョブレコードのheartbeat_data列に保存されます。これにより、ワーカーの外部から進捗を追跡できます:

PHP

Heartbeat::beat( [
    'processed' => 450,
    'total'     => 1000,
    'percent'   => 45,
] );

引数なしで呼び出すと、reserved_atのみが更新されます:

PHP

Heartbeat::beat();

ハートビートAPI

Queuety\Heartbeatクラスは完全に静的です。インスタンス化する必要はありません。

メソッド	説明
`Heartbeat::init( int $job_id, Connection $conn )`	ジョブコンテキストを設定します。処理の前にワーカーによって自動的に呼び出されます。
`Heartbeat::beat( array $progress = [] )`	ハートビートを送信します。`reserved_at`を更新し、必要に応じて進捗データを保存します。
`Heartbeat::clear()`	ジョブコンテキストをクリアします。処理の後にワーカーによって自動的に呼び出されます。
`Heartbeat::current_job_id()`	現在のジョブIDを取得します（テストに便利です）。

ストリーミングステップでの使用

ワーカーは、永続化する各チャンクの後に、heartbeat_dataを{"streaming_chunks": n}に設定してすでにハートビートを送信するため、定期的にデータを返すストリームはそのまま最新の状態に保たれます。データが返るまで長い間隔が空く場合（たとえば最初のトークンを待っている間）は、Heartbeat::beat()を呼び出してください。渡したデータは、ワーカーが次のチャンクごとのハートビートを送信すると上書きされます。

PHP

use Queuety\Contracts\StreamingStep;
use Queuety\Heartbeat;

class StreamAPIResponse implements StreamingStep {

    public function stream( array $state, array $existing_chunks = [] ): \Generator {
        $response = $this->client->stream( $state['prompt'] );

        foreach ( $response as $i => $chunk ) {
            Heartbeat::beat( [ 'chunks' => $i + 1 ] );
            yield $chunk;
        }
    }

    public function on_complete( array $chunks, array $state ): array {
        return [ 'response' => implode( '', $chunks ) ];
    }

    public function config(): array {
        return [];
    }
}

ハートビートを使う場面

1つのジョブまたはステップがQUEUETY_STALE_TIMEOUT秒（デフォルト: 600秒）を超えて実行される場合に、ハートビートを使います。よくあるケース:

LLMのAPI呼び出しでストリーミング応答を使う場合
大容量ファイルのダウンロードやアップロード
1つのステップ内でのバッチ処理の繰り返し
応答に数分かかることがある外部API呼び出し

ステップが1分未満で完了する場合、ハートビートは不要です。古いジョブの検出機能は、設定されたタイムアウトが経過してから動作します。