نبضات الحياة

يمكن للمهام طويلة التشغيل والخطوات المتدفقة إرسال نبضات حياة لإثبات أنها لا تزال تعمل. ومن دونها، قد يستعيد كاشف المهام القديمة مهمة لا تزال قيد المعالجة لأن طابعها الزمني reserved_at لم يُحدّث.

طريقة العمل

عندما يبدأ عامل معالجة مهمة، يستدعي Heartbeat::init() لتعيين سياق المهمة الحالية. داخل المعالج، يؤدي استدعاء Heartbeat::beat() إلى تحديث عمود reserved_at للمهمة إلى NOW(). ويستخدم الكاشف القديم reserved_at لتحديد ما إذا كانت المهمة عالقة، لذا فإن تحديثه يحافظ على سلامة المهمة.

عند انتهاء المهمة، يستدعي العامل Heartbeat::clear() لإعادة تعيين السياق.

إرسال نبضات الحياة

استدعِ Heartbeat::beat() من داخل أي معالج مهام أو خطوة متدفقة:

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

بيانات التقدم

المعلمة $progress مصفوفة ترابطية اختيارية. عند توفيرها، تُرمّز بصيغة JSON وتُخزّن في عمود heartbeat_data في سجل المهمة. يتيح لك ذلك تتبع التقدم من خارج العامل:

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

عند الاستدعاء من دون معاملات، لا يُحدّث سوى reserved_at:

Heartbeat::beat();

API لنبضات الحياة

الفئة Queuety\Heartbeat ثابتة بالكامل. لا تنشئ مثيلًا منها.

الاستخدام في الخطوات المتدفقة

يرسل العامل بالفعل نبضة حياة بعد كل جزء يحفظه، مع تعيين heartbeat_data إلى {"streaming_chunks": n}، لذا يظل التدفق الذي ينتج أجزاء بانتظام محدثًا تلقائيًا. استدعِ Heartbeat::beat() بنفسك عند احتمال وجود فترات طويلة بين الأجزاء، مثل انتظار الرمز الأول. وستُستبدل البيانات التي تمررها بنبضة العامل التالية لكل جزء:

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

متى تستخدم نبضات الحياة

استخدم نبضات الحياة عندما تستغرق مهمة أو خطوة واحدة أكثر من QUEUETY_STALE_TIMEOUT ثانية (الافتراضي: 600). ومن الحالات الشائعة:

• استدعاءات API لنماذج اللغة الكبيرة مع استجابات متدفقة
• تنزيل ملفات كبيرة أو تحميلها
• تكرارات المعالجة المجمّعة داخل خطوة واحدة
• أي استدعاء خارجي لـAPI قد يستغرق دقائق للرد

إذا اكتملت خطوتك في أقل من دقيقة، فلا حاجة إلى نبضات الحياة. لا يعمل كاشف المهام القديمة إلا بعد انقضاء المهلة المحددة.