Hjärtslag

Visa som Markdown Redigera på GitHub Dokumentation på 2cc7b2b

Långvariga jobb och strömmande steg kan skicka hjärtslag för att visa att de fortfarande körs. Utan hjärtslag kan detektorn för inaktuella jobb återta ett jobb som fortfarande bearbetas, eftersom tidsstämpeln reserved_at inte har uppdaterats.

Så fungerar det

När en arbetare börjar bearbeta ett jobb anropar den Heartbeat::init() för att ange den aktuella jobbkontexten. Om du anropar Heartbeat::beat() inuti din hanterare uppdateras jobbets kolumn reserved_at till NOW(). Detektorn för inaktuella jobb använder reserved_at för att avgöra om ett jobb har fastnat, så genom att uppdatera den hålls jobbet säkert.

När jobbet är klart anropar arbetaren Heartbeat::clear() för att återställa kontexten.

Skicka hjärtslag

Anropa Heartbeat::beat() inifrån valfri jobhanterare eller strömmande steg:

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.
    }
}

Förloppsdata

Parametern $progress är en valfri associativ array. Om den anges kodas den med JSON och lagras i kolumnen heartbeat_data i jobbraden. Då kan du följa förloppet utanför arbetaren:

PHP

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

När funktionen anropas utan argument uppdateras bara reserved_at:

PHP

Heartbeat::beat();

Hjärtslagsklassen API

Klassen Queuety\Heartbeat är helt statisk. Du skapar ingen instans av den.

Metod	Beskrivning
`Heartbeat::init( int $job_id, Connection $conn )`	Anger jobbkontexten. Anropas automatiskt av arbetaren före bearbetning.
`Heartbeat::beat( array $progress = [] )`	Skickar ett hjärtslag. Uppdaterar `reserved_at` och lagrar valfritt förloppsdata.
`Heartbeat::clear()`	Rensar jobbkontexten. Anropas automatiskt av arbetaren efter bearbetning.
`Heartbeat::current_job_id()`	Hämtar det aktuella jobb-ID:t (användbart vid testning).

Användning i strömmande steg

Arbetaren skickar redan ett hjärtslag efter varje del den sparar, med heartbeat_data inställt på {"streaming_chunks": n}, så en ström som ger ifrån sig data regelbundet hålls automatiskt aktuell. Anropa själv Heartbeat::beat() när det kan gå lång tid mellan datautskick, till exempel medan den väntar på den första tokenen. Data du skickar med skrivs över vid arbetarens nästa hjärtslag efter en del:

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

När du bör använda hjärtslag

Använd hjärtslag när ett enskilt jobb eller steg körs längre än QUEUETY_STALE_TIMEOUT sekunder (standard: 600). Vanliga fall:

LLM API-anrop med strömmande svar
Nedladdning eller uppladdning av stora filer
Upprepade batchbearbetningar i ett enda steg
Alla externa API-anrop som kan ta flera minuter att besvara

Om steget slutförs på mindre än en minut behövs inga pulssignaler. Den inaktuella detektorn aktiveras först när den konfigurerade tidsgränsen har passerats.