Настройка приоритезации задач в очереди (Priority Queue)

Настройка приоритезации задач в очереди (Priority Queue)

Все задачи в одной очереди обрабатываются в порядке поступления. Это нормально, пока задачи однородные. Как только появляются задачи с разными требованиями — письма о восстановлении пароля не должны ждать в очереди за отчётами, которые генерируются час, — нужна приоритезация.

Модель приоритетов

Типичное разделение на три уровня:

Реализация в Laravel (Redis)

Laravel поддерживает приоритеты очередей через порядок перечисления в --queue:

php artisan queue:work --queue=critical,default,low

Воркер сначала проверяет critical, если пусто — переходит к default, затем к low. Это soft-приоритет: при наличии задач в critical задачи в low не обрабатываются.

Диспатч в конкретную очередь:

// Критичная
SendPasswordResetEmail::dispatch($user)->onQueue('critical');

// Обычный приоритет
SendWelcomeEmail::dispatch($user)->onQueue('default');

// Низкий приоритет
GenerateMonthlyReport::dispatch($reportId)->onQueue('low');

Или определить очередь внутри самого Job:

class GenerateMonthlyReport implements ShouldQueue
{
    public string $queue = 'low';
    // ...
}

Horizon: разные пулы воркеров для разных приоритетов

В Horizon создаём отдельные supervisor-пулы:

// config/horizon.php
'environments' => [
    'production' => [
        // Пул для критичных задач — всегда минимум 2 воркера
        'critical-supervisor' => [
            'connection'     => 'redis',
            'queue'          => ['critical'],
            'balance'        => 'simple',
            'minProcesses'   => 2,
            'maxProcesses'   => 8,
            'timeout'        => 30,
        ],
        // Пул для стандартных задач — автоскейлинг
        'default-supervisor' => [
            'connection'     => 'redis',
            'queue'          => ['default'],
            'balance'        => 'auto',
            'minProcesses'   => 1,
            'maxProcesses'   => 5,
            'timeout'        => 60,
        ],
        // Пул для тяжёлых задач — ограниченно
        'low-supervisor' => [
            'connection'     => 'redis',
            'queue'          => ['low'],
            'balance'        => 'simple',
            'processes'      => 2,
            'timeout'        => 3600,
        ],
    ],
],

При такой конфигурации критичные задачи обрабатываются на выделенных воркерах независимо от загрузки других очередей.

Динамический приоритет на основе данных

Иногда приоритет нужно определять не статически, а на основе данных — например, платящие клиенты должны иметь более высокий приоритет обработки:

class ProcessUserExportJob implements ShouldQueue
{
    public function __construct(private int $userId) {}

    public function queue(): string
    {
        $user = User::find($this->userId);
        return match(true) {
            $user?->isPremium() => 'default',
            $user?->isEnterprise() => 'critical',
            default => 'low',
        };
    }
}

Но метод queue() как динамический в Laravel не поддерживается напрямую — приоритет нужно определять при диспатче:

$queue = match(true) {
    $user->isEnterprise() => 'critical',
    $user->isPremium()    => 'default',
    default               => 'low',
};

ProcessUserExportJob::dispatch($user->id)->onQueue($queue);

Приоритет в BullMQ (Node.js)

BullMQ поддерживает числовой приоритет задач:

import { Queue } from 'bullmq';

const queue = new Queue('tasks', {
    connection: { host: 'localhost', port: 6379 }
});

// Меньшее число = выше приоритет (1 = самый высокий)
await queue.add('send-password-reset', { userId: 123 }, { priority: 1 });
await queue.add('send-welcome-email',  { userId: 456 }, { priority: 5 });
await queue.add('generate-report',     { reportId: 789}, { priority: 10 });

BullMQ использует Redis Sorted Set для хранения задач с приоритетом — гарантированная сортировка по приоритету внутри одной очереди.

Воркер просто забирает задачи в порядке приоритета:

import { Worker } from 'bullmq';

const worker = new Worker('tasks', async (job) => {
    console.log(`Processing ${job.name} with priority ${job.opts.priority}`);
    // ...
}, { connection: { host: 'localhost', port: 6379 } });

Защита от голодания низкоприоритетных задач

При большом потоке критичных задач задачи из low могут никогда не обрабатываться (starvation). Два подхода:

Aging — задача увеличивает свой приоритет со временем ожидания. Реализуется через scheduled job, который периодически пересматривает очередь:

// Каждые 15 минут повышаем приоритет давно ждущих задач
Schedule::call(function () {
    $staleJobs = DB::table('jobs')
        ->where('queue', 'low')
        ->where('created_at', '<', now()->subMinutes(30))
        ->get();

    foreach ($staleJobs as $job) {
        DB::table('jobs')
            ->where('id', $job->id)
            ->update(['queue' => 'default']); // повышаем до default
    }
})->everyFifteenMinutes();

Выделенный воркер для low — один dedicated воркер всегда работает только с low-очередью, не отвлекаясь на critical. Гарантирует прогресс даже при загрузке:

[program:low-dedicated-worker]
command=php artisan queue:work --queue=low --sleep=5 --timeout=3600
numprocs=1
autostart=true
autorestart=true

Мониторинг глубины очередей

Важно следить за длиной каждой очереди — если critical накапливается, это симптом нехватки воркеров:

use Illuminate\Support\Facades\Redis;

$depths = [
    'critical' => Redis::llen('queues:critical'),
    'default'  => Redis::llen('queues:default'),
    'low'      => Redis::llen('queues:low'),
];

// Алерт если critical > 50 задач
if ($depths['critical'] > 50) {
    // уведомление
}

Horizon показывает это в дашборде без дополнительного кода.

Сроки

Настройка трёх очередей, разделение существующих задач по приоритетам, Horizon-конфиг пулов — 3–5 часов. Логика антиголодания, мониторинг глубины, алертинг — ещё 2–4 часа.