Sykesignaalit

Näytä Markdown-muodossa Muokkaa palvelussa GitHub Dokumentaatio: 2cc7b2b

Pitkään suoritettavat työt ja suoratoistovaiheet voivat lähettää sykesignaaleja osoittaakseen olevansa yhä käynnissä. Ilman sykesignaaleja vanhentuneiden töiden tunnistin voi ottaa työn uudelleen käsiteltäväksi, vaikka sitä yhä käsitellään aktiivisesti, koska sen reserved_at-aikaleimaa ei ole päivitetty.

Näin se toimii

Kun työntekijä aloittaa työn käsittelyn, se kutsuu toimintoa Heartbeat::init() nykyisen työyhteyden asettamiseksi. Käsittelijässäsi kutsu Heartbeat::beat() päivittää työn reserved_at-sarakkeen arvoon NOW(). Vanhentuneiden töiden tunnistin käyttää arvoa reserved_at jumiutuneiden töiden tunnistamiseen, joten sen päivittäminen pitää työn turvassa.

Kun työ päättyy, työntekijä kutsuu toimintoa Heartbeat::clear() yhteyden nollaamiseksi.

Sykesignaalien lähettäminen

Kutsu toimintoa Heartbeat::beat() minkä tahansa työn käsittelijän tai suoratoistovaiheen sisältä:

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

Edistymistiedot

Parametri $progress on valinnainen assosiatiivinen taulukko. Kun se annetaan, se koodataan JSON-muodossa ja tallennetaan työrivin sarakkeeseen heartbeat_data. Näin voit seurata edistymistä työntekijäprosessin ulkopuolelta:

PHP

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

Ilman argumentteja kutsuttaessa päivitetään vain reserved_at:

PHP

Heartbeat::beat();

Sykesignaalien API

Luokka Queuety\Heartbeat on kokonaan staattinen. Siitä ei luoda instanssia.

Metodi	Kuvaus
`Heartbeat::init( int $job_id, Connection $conn )`	Asettaa työyhteyden. Työntekijä kutsuu sitä automaattisesti ennen käsittelyä.
`Heartbeat::beat( array $progress = [] )`	Lähettää sykesignaalin. Päivittää arvon `reserved_at` ja tallentaa haluttaessa edistymistietoja.
`Heartbeat::clear()`	Tyhjentää työyhteyden. Työntekijä kutsuu sitä automaattisesti käsittelyn jälkeen.
`Heartbeat::current_job_id()`	Hakee nykyisen työn tunnuksen (hyödyllinen testauksessa).

Käyttö suoratoistovaiheissa

Työntekijä lähettää jo sykesignaalin jokaisen tallentamansa osan jälkeen, ja heartbeat_data on asetettu arvoon {"streaming_chunks": n}, joten säännöllisesti tietoja tuottava virta pysyy ajan tasalla automaattisesti. Kutsu toimintoa Heartbeat::beat() itse, jos tuottojen välillä voi olla pitkiä taukoja, esimerkiksi ensimmäistä tunnistetta odotettaessa. Antamasi tiedot korvautuvat työntekijän seuraavan osakohtaisen sykesignaalin yhteydessä:

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

Milloin sykesignaaleja kannattaa käyttää

Käytä sykesignaaleja, kun yksittäinen työ tai vaihe kestää yli QUEUETY_STALE_TIMEOUT sekuntia (oletus: 600). Tavallisia tapauksia:

LLM API -kutsut, jotka palauttavat vastauksia suoratoistona
Suurten tiedostojen lataukset tai lähetykset
Erän toistojen käsittely yhden vaiheen sisällä
Ulkoiset API-kutsut, joiden vastaamiseen voi kulua minuutteja

Jos vaihe valmistuu alle minuutissa, sykesignaaleja ei tarvita. Vanhentuneiden töiden tunnistin aktivoituu vasta määritetyn aikakatkaisun kuluttua.