Életjelek

Megtekintés Markdownként Szerkesztés itt: GitHub Dokumentáció itt: 2cc7b2b

A hosszan futó feladatok és adatfolyam-lépések életjeleket küldhetnek annak igazolására, hogy még futnak. Életjelek nélkül az elavult feladatokat kereső észlelő újra sorba állíthat egy még aktívan feldolgozott feladatot, mert annak reserved_at időbélyege nem frissült.

Működés

Amikor egy munkafolyamat elkezd feldolgozni egy feladatot, meghívja a Heartbeat::init() függvényt az aktuális feladatkontextus beállításához. A kezelőben a Heartbeat::beat() meghívása frissíti a feladat reserved_at oszlopát erre: NOW(). Az elavult feladatokat észlelő a reserved_at alapján állapítja meg, hogy egy feladat elakadt-e, így ennek frissítése biztonságban tartja a feladatot.

A feladat befejezésekor a munkafolyamat meghívja a Heartbeat::clear() függvényt a kontextus visszaállításához.

Életjelek küldése

Hívja meg a Heartbeat::beat() függvényt bármely feladatkezelőből vagy adatfolyam-lépésből:

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

Folyamatadatok

A $progress paraméter opcionális asszociatív tömb. Megadása esetén JSON formátumra kódolja, majd a feladatsor heartbeat_data oszlopában tárolja. Így a munkafolyamaton kívülről is nyomon követheti az előrehaladást:

PHP

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

Argumentumok nélkül hívva csak a reserved_at frissül:

PHP

Heartbeat::beat();

Életjel API

A Queuety\Heartbeat osztály teljesen statikus. Nem kell példányosítani.

Metódus	Leírás
`Heartbeat::init( int $job_id, Connection $conn )`	Beállítja a feladatkontextust. A munkafolyamat automatikusan meghívja feldolgozás előtt.
`Heartbeat::beat( array $progress = [] )`	Életjelet küld. Frissíti a `reserved_at` értékét, és opcionálisan folyamatadatokat tárol.
`Heartbeat::clear()`	Törli a feladatkontextust. A munkafolyamat automatikusan meghívja feldolgozás után.
`Heartbeat::current_job_id()`	Lekéri az aktuális feladatazonosítót (tesztelésnél hasznos).

Használat adatfolyam-lépésekben

A munkafolyamat minden mentett adatfolyam-rész után automatikusan életjelet küld, és a heartbeat_data értékét {"streaming_chunks": n}-re állítja, így a rendszeresen adatot szolgáltató adatfolyam magától is friss marad. Hívja meg a Heartbeat::beat() függvényt, ha két adatközlés között hosszú szünet lehet, például az első tokenre várakozás közben. Az átadott adatokat felülírja a munkafolyamat következő, adatfolyam-részenként küldött életjele:

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

Mikor használjon életjeleket

Akkor használjon életjeleket, ha egyetlen feladat vagy lépés több mint QUEUETY_STALE_TIMEOUT másodpercig fut (alapértelmezett: 600). Gyakori esetek:

LLM API hívások adatfolyamos válaszokkal
Nagy fájlok letöltése vagy feltöltése
Kötegelt feldolgozási ismétlések egyetlen lépésen belül
Bármely külső API hívás, amelynek válasza percekig tarthat

Ha a lépés egy percen belül befejeződik, nincs szükség életjelekre. Az elavult feladatokat észlelő csak a beállított időkorlát letelte után lép működésbe.