Aktyvumo signalai

Peržiūrėti kaip Markdown Redaguoti per GitHub Dokumentacija: 2cc7b2b

Ilgai trunkančios užduotys ir srautiniai veiksmai gali siųsti aktyvumo signalus, patvirtinančius, kad jie vis dar vykdomi. Be šių signalų pasenusių užduočių aptikimo funkcija gali perimti vis dar apdorojamą užduotį, nes jos reserved_at laiko žyma nebuvo atnaujinta.

Kaip tai veikia

Pradėjęs apdoroti užduotį, vykdytojas iškviečia Heartbeat::init(), kad nustatytų dabartinį užduoties kontekstą. Apdorojimo funkcijoje iškvietus Heartbeat::beat(), užduoties reserved_at stulpelis atnaujinamas į NOW(). Pasenusių užduočių aptikimo funkcija naudoja reserved_at, kad nustatytų, ar užduotis užstrigo, todėl atnaujinant šią reikšmę užduotis lieka apsaugota.

Užduočiai baigus veikti, vykdytojas iškviečia Heartbeat::clear(), kad iš naujo nustatytų kontekstą.

Aktyvumo signalų siuntimas

Iškvieskite Heartbeat::beat() bet kurios užduoties apdorojimo funkcijos ar srautinio veiksmo viduje:

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

Eigos duomenys

Parametras $progress yra pasirenkamas asociatyvusis masyvas. Jei jis pateikiamas, jis užkoduojamas naudojant JSON ir įrašomas į užduoties eilutės stulpelį heartbeat_data. Taip galite stebėti eigą ne vykdytojo procese:

PHP

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

Iškvietus be argumentų, atnaujinamas tik reserved_at:

PHP

Heartbeat::beat();

Aktyvumo signalų API

Klasė Queuety\Heartbeat yra visiškai statinė. Jos egzemplioriaus kurti nereikia.

Metodas	Aprašymas
`Heartbeat::init( int $job_id, Connection $conn )`	Nustato užduoties kontekstą. Vykdytojas tai atlieka automatiškai prieš apdorojimą.
`Heartbeat::beat( array $progress = [] )`	Siunčia aktyvumo signalą. Atnaujina `reserved_at` ir pasirinktinai išsaugo eigos duomenis.
`Heartbeat::clear()`	Išvalo užduoties kontekstą. Vykdytojas tai atlieka automatiškai po apdorojimo.
`Heartbeat::current_job_id()`	Grąžina dabartinės užduoties ID (naudinga testuojant).

Naudojimas srautiniuose veiksmuose

Vykdytojas jau siunčia aktyvumo signalą po kiekvienos išsaugotos dalies, nustatydamas heartbeat_data į {"streaming_chunks": n}, todėl reguliariai duomenis pateikiantis srautas savaime išlieka aktyvus. Patys iškvieskite Heartbeat::beat(), kai tarp duomenų pateikimų gali būti ilgos pertraukos, pavyzdžiui, laukiant pirmojo žetono. Jūsų perduotus duomenis perrašys kitas vykdytojo signalas po apdorotos dalies:

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

Kada naudoti aktyvumo signalus

Naudokite aktyvumo signalus, kai viena užduotis ar veiksmas trunka ilgiau nei QUEUETY_STALE_TIMEOUT sekundžių (numatytoji reikšmė: 600). Dažniausi atvejai:

LLM API iškvietimai su srautiniais atsakymais
Didelių failų atsisiuntimas ar įkėlimas
Pakartotinis paketinis apdorojimas vieno veiksmo metu
Bet kuris išorinis API iškvietimas, kurio atsakymo gali tekti laukti kelias minutes

Jei veiksmas baigiamas greičiau nei per minutę, aktyvumo signalų nereikia. Pasenusių užduočių aptikimo funkcija įsijungia tik praėjus sukonfigūruotam skirtajam laikui.