心跳

以 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 [];
    }
}

何时使用心跳

当单个任务或步骤运行时间超过 QUEUETY_STALE_TIMEOUT 秒时（默认值：600），请使用心跳。常见场景包括：

LLM API 调用（支持流式响应）
大文件下载或上传
单个步骤内的批量处理重复操作
任何可能需要几分钟才能响应的外部 API 调用

如果你的步骤能在一分钟内完成，就不需要心跳。只有配置的超时时间已过后，过期检测器才会启动。