Kalp Atışları

Markdown olarak görüntüle GitHub üzerinde düzenle Dokümantasyon: 2cc7b2b

Uzun süren işler ve akış adımları, hâlâ etkin olduklarını göstermek için kalp atışları gönderebilir. Kalp atışları olmadan, reserved_at zaman damgası güncellenmediği için eski iş algılayıcısı hâlâ etkin biçimde işlenen bir işi yeniden alabilir.

Nasıl çalışır

Bir çalışan bir işi işlemeye başladığında geçerli iş bağlamını ayarlamak için Heartbeat::init() çağrısı yapar. İşleyiciniz içinde Heartbeat::beat() çağrısı, iş satırındaki reserved_at sütununu NOW() olarak günceller. Eski iş algılayıcısı, bir işin takılıp kalıp kalmadığını belirlemek için reserved_at kullanır; bu nedenle değeri yenilemek işi güvende tutar.

İş tamamlandığında çalışan, bağlamı sıfırlamak için Heartbeat::clear() çağrısı yapar.

Kalp atışları gönderme

Herhangi bir iş işleyicisinin veya akış adımının içinden Heartbeat::beat() çağrısı yapın:

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

İlerleme verileri

$progress parametresi isteğe bağlı bir ilişkisel dizidir. Sağlandığında JSON biçiminde kodlanır ve iş satırındaki heartbeat_data sütununda saklanır. Böylece ilerlemeyi çalışan dışından takip edebilirsiniz:

PHP

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

Bağımsız değişken olmadan çağrıldığında yalnızca reserved_at güncellenir:

PHP

Heartbeat::beat();

Kalp atışı API

Queuety\Heartbeat sınıfı tamamen statiktir. Bir örneğini oluşturmazsınız.

Yöntem	Açıklama
`Heartbeat::init( int $job_id, Connection $conn )`	İş bağlamını ayarlar. İşleme başlamadan önce çalışan tarafından otomatik olarak çağrılır.
`Heartbeat::beat( array $progress = [] )`	Kalp atışı gönderir. `reserved_at` değerini günceller ve isteğe bağlı olarak ilerleme verilerini saklar.
`Heartbeat::clear()`	İş bağlamını temizler. İşleme tamamlandıktan sonra çalışan tarafından otomatik olarak çağrılır.
`Heartbeat::current_job_id()`	Geçerli iş kimliğini alır (test için kullanışlıdır).

Akış adımlarında kullanım

Çalışan, sakladığı her parçadan sonra heartbeat_data değerini {"streaming_chunks": n} olarak ayarlayarak zaten bir kalp atışı gönderir; bu nedenle düzenli olarak çıktı üreten bir akış güncel kalır. İlk belirteci beklerken olduğu gibi, çıktılar arasında uzun aralar olabilecek durumlarda Heartbeat::beat() çağrısını kendiniz yapın. İlettiğiniz verilerin üzerine çalışanın bir sonraki parça başına kalp atışı sırasında yazılır:

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

Kalp atışları ne zaman kullanılır

Tek bir iş veya adım QUEUETY_STALE_TIMEOUT saniyeden uzun sürüyorsa kalp atışlarını kullanın (varsayılan: 600). Yaygın durumlar:

Akış yanıtlarıyla LLM API çağrıları
Büyük dosyaların indirilmesi veya yüklenmesi
Tek bir adım içindeki toplu işlem tekrarları
Yanıt vermesi dakikalar sürebilecek harici API çağrıları

Adımınız bir dakikadan kısa sürede tamamlanıyorsa heartbeat'lere gerek yoktur. Eski kayıt algılayıcısı yalnızca yapılandırılan zaman aşımı dolduktan sonra devreye girer.