Signali delovanja

Prikaži kot Markdown Uredi na GitHub Dokumentacija za 2cc7b2b

Dolgotrajna opravila in koraki pretakanja lahko pošiljajo signale delovanja, da potrdijo, da še vedno potekajo. Brez teh signalov lahko zaznavalnik zastarelih opravil znova prevzame opravilo, ki se še aktivno obdeluje, ker njegov časovni žig reserved_at ni bil posodobljen.

Kako deluje

Ko delavec začne obdelovati opravilo, pokliče Heartbeat::init(), da nastavi kontekst trenutnega opravila. Znotraj obravnavalnika klic Heartbeat::beat() posodobi stolpec opravila reserved_at na NOW(). Zaznavalnik zastarelih opravil uporablja reserved_at za ugotavljanje, ali je opravilo obtičalo, zato ga osvežitev zaščiti.

Ko se opravilo konča, delavec pokliče Heartbeat::clear(), da ponastavi kontekst.

Pošiljanje signalov delovanja

Pokličite Heartbeat::beat() znotraj katerega koli obravnavalnika opravil ali koraka pretakanja:

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

Podatki o napredku

Parameter $progress je izbirno asociativno polje. Če ga podate, se kodira v JSON in shrani v stolpec heartbeat_data v vrstici opravila. Tako lahko spremljate napredek zunaj delavca:

PHP

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

Če funkcijo pokličete brez argumentov, se posodobi samo reserved_at:

PHP

Heartbeat::beat();

Razred API za signale delovanja

Razred Queuety\Heartbeat je povsem statičen. Ne ustvarjate njegovih primerkov.

Metoda	Opis
`Heartbeat::init( int $job_id, Connection $conn )`	Nastavi kontekst opravila. Delavec to samodejno pokliče pred obdelavo.
`Heartbeat::beat( array $progress = [] )`	Pošlje signal delovanja. Posodobi `reserved_at` in po želji shrani podatke o napredku.
`Heartbeat::clear()`	Počisti kontekst opravila. Delavec to samodejno pokliče po obdelavi.
`Heartbeat::current_job_id()`	Pridobi ID trenutnega opravila (uporabno za testiranje).

Uporaba v korakih pretakanja

Delavec po vsakem shranjenem kosu že pošlje signal delovanja z nastavitvijo heartbeat_data na {"streaming_chunks": n}, zato se tok, ki redno oddaja podatke, sam ohranja kot svež. Sami pokličite Heartbeat::beat(), kadar so lahko med oddajami dolgi premori, na primer med čakanjem na prvi žeton. Podatke, ki jih posredujete, prepiše naslednji signal delavca po oddaji posameznega kosa:

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

Kdaj uporabiti signale delovanja

Uporabite signale delovanja, kadar posamezno opravilo ali korak traja dlje kot QUEUETY_STALE_TIMEOUT sekund (privzeto: 600). Pogosti primeri:

Klici LLM API s pretočnimi odzivi
Prenosi velikih datotek
Ponavljajoča se paketna obdelava znotraj enega koraka
Kateri koli zunanji klic API, ki lahko traja več minut

Če se vaš korak zaključi v manj kot minuti, utripi niso potrebni. Zaznavanje zastarelih opravil se sproži šele po preteku nastavljenega časovnega zamika.