Сигналы активности

Просмотреть как Markdown Редактировать на GitHub Документация по 2cc7b2b

Длительные задания и потоковые этапы могут отправлять сигналы активности, подтверждая, что они всё ещё выполняются. Без таких сигналов детектор зависших заданий может повторно захватить задание, которое всё ещё обрабатывается, поскольку его метка времени reserved_at не обновлялась.

Как это работает

Когда рабочий процесс начинает обработку задания, он вызывает Heartbeat::init(), чтобы задать контекст текущего задания. В обработчике вызов Heartbeat::beat() обновляет столбец reserved_at задания, устанавливая значение NOW(). Детектор зависших заданий проверяет reserved_at, чтобы определить, не зависло ли задание; обновление этого значения защищает задание от повторного захвата.

После завершения задания рабочий процесс вызывает Heartbeat::clear(), чтобы сбросить контекст.

Отправка сигналов активности

Вызывайте 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.
    }
}

Данные о ходе выполнения

Параметр $progress — необязательный ассоциативный массив. Если он указан, массив кодируется в формате JSON и сохраняется в столбце heartbeat_data строки задания. Так вы сможете отслеживать ход выполнения вне рабочего процесса:

PHP

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

При вызове без аргументов обновляется только reserved_at:

PHP

Heartbeat::beat();

API сигналов активности

Класс Queuety\Heartbeat полностью состоит из статических методов. Создавать его экземпляр не нужно.

Метод	Описание
`Heartbeat::init( int $job_id, Connection $conn )`	Задаёт контекст задания. Автоматически вызывается рабочим процессом перед обработкой.
`Heartbeat::beat( array $progress = [] )`	Отправляет сигнал активности. Обновляет `reserved_at` и, при необходимости, сохраняет данные о ходе выполнения.
`Heartbeat::clear()`	Очищает контекст задания. Автоматически вызывается рабочим процессом после обработки.
`Heartbeat::current_job_id()`	Возвращает текущий идентификатор задания (полезно для тестирования).

Использование в потоковых этапах

Рабочий процесс уже отправляет сигнал активности после каждого сохранённого фрагмента, устанавливая heartbeat_data в значение {"streaming_chunks": n}, поэтому поток, регулярно выдающий данные, остаётся актуальным и без дополнительных действий. Вызывайте Heartbeat::beat() самостоятельно, если между выдачами возможны длительные паузы, например при ожидании первого токена. Переданные вами данные будут перезаписаны следующим сигналом рабочего процесса после очередного фрагмента:

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

Когда использовать сигналы активности

Используйте сигналы активности, если выполнение отдельного задания или этапа длится дольше QUEUETY_STALE_TIMEOUT секунд (по умолчанию: 600). Типичные случаи:

Вызовы API с потоковыми ответами
Загрузка больших файлов
Повторная пакетная обработка внутри одного этапа
Вызовы внешнего API, ответ от которого может занять несколько минут

Если этап завершается менее чем за минуту, сигналы активности не нужны. Детектор зависших заданий срабатывает только после истечения заданного тайм-аута.