Südamelöögid

Vaata Markdownina Muuda teenuses GitHub Dokumentatsioon: 2cc7b2b

Pikka aega kestvad tööd ja voogedastuse sammud saavad saata südamelööke, mis tõendavad, et need on endiselt aktiivsed. Ilma südamelöökideta võib aegunud töö tuvastaja võtta tagasi töö, mis alles töötleb, sest selle ajatemplit reserved_at pole uuendatud.

Kuidas see töötab

Kui töötaja alustab töö töötlemist, kutsub see välja Heartbeat::init(), et määrata praegune töökontekst. Töötleja sees uuendab Heartbeat::beat() väljakutsumine töö veeru reserved_at väärtuseks NOW(). Aegunud töö tuvastaja kasutab väärtust reserved_at, et otsustada, kas töö on takerdunud, seega selle värskendamine hoiab töö turvaliselt töös.

Kui töö lõpeb, kutsub töötaja välja Heartbeat::clear(), et kontekst lähtestada.

Südamelöökide saatmine

Kutsu mis tahes töötleja või voogedastuse sammu seest välja 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.
    }
}

Edenemisandmed

Parameeter $progress on valikuline assotsiatiivne massiiv. Kui see edastatakse, kodeeritakse see vormingus JSON ja salvestatakse töö rea veergu heartbeat_data. Nii saad jälgida edenemist ka väljastpoolt töötajat:

PHP

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

Ilma argumentideta väljakutsumisel uuendatakse ainult väärtust reserved_at:

PHP

Heartbeat::beat();

Südamelöögiklass API

Klass Queuety\Heartbeat koosneb ainult staatilistest liikmetest. Selle eksemplari ei looda.

Meetod	Kirjeldus
`Heartbeat::init( int $job_id, Connection $conn )`	Määra töökontekst. Töötaja kutsub selle enne töötlemist automaatselt välja.
`Heartbeat::beat( array $progress = [] )`	Saada südamelöök. Uuendab väärtust `reserved_at` ja salvestab soovi korral edenemisandmed.
`Heartbeat::clear()`	Tühjenda töökontekst. Töötaja kutsub selle pärast töötlemist automaatselt välja.
`Heartbeat::current_job_id()`	Hangi praeguse töö ID (kasulik testimisel).

Kasutamine voogedastuse sammudes

Töötaja saadab juba pärast iga salvestatud tükki südamelöögi, mille heartbeat_data väärtuseks on {"streaming_chunks": n}, nii et regulaarselt andmeid väljastav voog püsib ise värskena. Kui väljastuste vahele võivad jääda pikad pausid, näiteks esimese märgi ootamisel, kutsu ise välja Heartbeat::beat(). Edastatud andmed kirjutatakse üle töötaja järgmise tüki kaupa saadetava südamelöögiga:

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

Millal südamelööke kasutada

Kasuta südamelööke, kui üks töö või samm kestab üle QUEUETY_STALE_TIMEOUT sekundi (vaikimisi: 600). Levinud juhtumid:

LLM API-kutsed voogedastusega
Suurte failide allalaadimine või üleslaadimine
Pakktöötluse kordused ühe sammu sees
Iga väline API-kutse, millele vastamine võib võtta minuteid

Kui samm lõpeb vähem kui minutiga, pole südamelööke vaja. Aegunud töö tuvastamine käivitub alles pärast seadistatud ajalõpu möödumist.