하트비트

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를 가져옵니다(테스트에 유용함).

스트리밍 단계에서 사용

워커는 저장하는 각 청크가 끝난 후 이미 하트비트를 보내며, {"streaming_chunks": n}를 heartbeat_data로 설정합니다. 따라서 스트림이 규칙적으로 생성되면 자동으로 최신 상태를 유지합니다. 첫 토큰을 기다리는 동안처럼 생성 사이에 긴 간격이 있을 수 있다면 직접 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 호출

단계가 1분 이내에 완료된다면 하트비트는 필요하지 않습니다. 오래된 작업 감지기는 설정된 시간 제한이 지난 후에만 작동합니다.