Налаштування пріоритизації завдань у черзі (Priority Queue)

Налаштування пріоритизації завдань у черзі (Priority Queue)

Всі завдання в одній черзі обробляються в порядку надходження. Це нормально, поки завдання однорідні. Як тільки з'являються завдання з різними вимогами — листи про відновлення пароля не мають чекати в черзі за звітами, які генеруються годину, — потрібна пріоритизація.

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

Типовий поділ на три рівні:

Реалізація у Laravel (Redis)

Laravel підтримує пріоритети черг через порядок перелічення в --queue:

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

Worker спочатку перевіряє 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 для зберігання завдань з пріоритетом — гарантована сортування за пріоритетом всередину однієї черги.

Worker просто забирає завдання в порядку пріоритету:

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 години.