Signály aktivity

Zobrazit jako Markdown Upravit na GitHub Dokumentace na 2cc7b2b

Dlouho běžící úlohy a streamovací kroky mohou odesílat signály aktivity, které potvrzují, že stále běží. Bez nich může detektor zastaralých úloh znovu zařadit úlohu, která se stále zpracovává, protože její časové razítko reserved_at nebylo aktualizováno.

Jak to funguje

Když worker začne zpracovávat úlohu, zavolá Heartbeat::init() a nastaví kontext aktuální úlohy. Zavoláním Heartbeat::beat() uvnitř obslužné rutiny se sloupec úlohy reserved_at aktualizuje na NOW(). Detektor zastaralých úloh pomocí reserved_at zjišťuje, zda se úloha zasekla, takže její obnovení ji udržuje v bezpečném stavu.

Po dokončení úlohy worker zavolá Heartbeat::clear() a resetuje kontext.

Odesílání signálů aktivity

Zavolejte Heartbeat::beat() uvnitř libovolné obslužné rutiny úlohy nebo streamovacího kroku:

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

Údaje o průběhu

Parametr $progress je volitelné asociativní pole. Pokud ho zadáte, zakóduje se ve formátu JSON a uloží do sloupce heartbeat_data v řádku úlohy. Díky tomu můžete sledovat průběh i mimo worker:

PHP

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

Při zavolání bez argumentů se aktualizuje pouze reserved_at:

PHP

Heartbeat::beat();

API signálu aktivity

Třída Queuety\Heartbeat obsahuje pouze statické metody. Nevytvářejte její instanci.

Metoda	Popis
`Heartbeat::init( int $job_id, Connection $conn )`	Nastaví kontext úlohy. Worker ji automaticky zavolá před zpracováním.
`Heartbeat::beat( array $progress = [] )`	Odešle signál aktivity. Aktualizuje `reserved_at` a případně uloží údaje o průběhu.
`Heartbeat::clear()`	Vymaže kontext úlohy. Worker ji automaticky zavolá po zpracování.
`Heartbeat::current_job_id()`	Vrátí aktuální ID úlohy (užitečné při testování).

Použití ve streamovacích krocích

Worker po uložení každého bloku už odesílá signál aktivity s hodnotou {"streaming_chunks": n} v heartbeat_data, takže stream, který pravidelně vrací data, zůstává automaticky aktuální. Pokud mezi jednotlivými výstupy mohou být dlouhé prodlevy, například při čekání na první token, zavolejte Heartbeat::beat() sami. Data, která předáte, worker při příštím signálu po zpracování bloku přepíše:

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

Kdy používat signály aktivity

Signály aktivity používejte, když jeden úkol nebo krok běží déle než QUEUETY_STALE_TIMEOUT sekund (výchozí hodnota: 600). Typické případy:

Volání API LLM se streamovanými odpověďmi
Stahování nebo nahrávání velkých souborů
Opakované dávkové zpracování v rámci jednoho kroku
Volání externího rozhraní API, jehož odpověď může trvat několik minut

Pokud se krok dokončí za méně než minutu, signály aktivity nejsou potřeba. Detektor zastaralých úloh se aktivuje až po uplynutí nastaveného časového limitu.