Очереди задач в PHP: руководство по реализации и оптимизации

Работа с очередями задач в PHP: от простого к сложному

В современных веб-приложениях часто возникает необходимость выполнять ресурсоёмкие или длительные операции, не заставляя пользователя ждать. Отправка email-уведомлений, генерация отчётов, обработка изображений, интеграция с внешними API — все эти задачи идеально подходят для асинхронного выполнения. Именно здесь на помощь приходят очереди задач (job queues). В этой статье мы подробно рассмотрим, как реализовать и оптимизировать работу с очередями в PHP, начиная с простейших решений и заканчивая промышленными инструментами.

Что такое очередь задач и зачем она нужна?

Очередь задач (job queue) — это механизм, который позволяет отложить выполнение некоторой работы на потом. Вместо того чтобы выполнять задачу немедленно в рамках HTTP-запроса, вы помещаете её в очередь. Специальный воркер (worker) или группа воркеров извлекают задачи из очереди и обрабатывают их в фоновом режиме.

Основные преимущества использования очередей:

• Ускорение ответа пользователю: Пользователь получает ответ сразу, а тяжёлая работа выполняется асинхронно.
• Масштабирование: Вы можете запустить несколько воркеров для параллельной обработки задач.
• Надёжность: Если задача не выполнилась (например, из-за временной ошибки), её можно повторить позже.
• Разгрузка основного сервера: Фоновые процессы не блокируют обработку новых HTTP-запросов.

Простая очередь на основе базы данных

Самый простой способ организовать очередь в PHP — использовать таблицу в базе данных (например, MySQL). Этот подход не требует установки дополнительного программного обеспечения и отлично подходит для небольших проектов.

Шаг 1: Создание таблицы jobs

CREATE TABLE jobs (
    id INT AUTO_INCREMENT PRIMARY KEY,
    queue_name VARCHAR(255) DEFAULT 'default',
    payload TEXT NOT NULL,
    status ENUM('pending', 'processing', 'completed', 'failed') DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    available_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    attempts INT DEFAULT 0
);

Шаг 2: Добавление задачи в очередь

function addJob(string $queueName, array $data): void {
    $pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'password');
    $stmt = $pdo->prepare(
        "INSERT INTO jobs (queue_name, payload, available_at) 
         VALUES (?, ?, NOW())"
    );
    $stmt->execute([$queueName, json_encode($data)]);
}

// Пример использования
addJob('emails', ['to' => 'user@example.com', 'subject' => 'Привет!']);

Шаг 3: Создание воркера (worker.php)

// Бесконечный цикл для обработки задач
while (true) {
    $pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'password');
    
    // Захват задачи с блокировкой строки
    $stmt = $pdo->prepare(
        "SELECT * FROM jobs 
         WHERE status = 'pending' AND available_at <= NOW() 
         ORDER BY id ASC 
         LIMIT 1 FOR UPDATE"
    );
    $stmt->execute();
    $job = $stmt->fetch(PDO::FETCH_ASSOC);
    
    if ($job) {
        // Помечаем как "в обработке"
        $updateStmt = $pdo->prepare("UPDATE jobs SET status = 'processing' WHERE id = ?");
        $updateStmt->execute([$job['id']]);
        
        try {
            // Выполнение задачи
            $data = json_decode($job['payload'], true);
            processJob($data);
            
            // Отмечаем как выполненную
            $pdo->prepare("UPDATE jobs SET status = 'completed' WHERE id = ?")
                ->execute([$job['id']]);
        } catch (Exception $e) {
            // Обработка ошибки
            $attempts = $job['attempts'] + 1;
            if ($attempts < 3) {
                // Повтор через 5 секунд
                $pdo->prepare("UPDATE jobs SET status = 'pending', attempts = ?, available_at = DATE_ADD(NOW(), INTERVAL 5 SECOND) WHERE id = ?")
                    ->execute([$attempts, $job['id']]);
            } else {
                $pdo->prepare("UPDATE jobs SET status = 'failed' WHERE id = ?")
                    ->execute([$job['id']]);
            }
        }
    } else {
        // Если задач нет — ждём 1 секунду
        sleep(1);
    }
}

Недостатки такого подхода:

• Низкая производительность при большом количестве задач.
• Риск блокировок таблицы при параллельных воркерах.
• Отсутствие встроенных механизмов задержки и повторных попыток (всё нужно реализовывать вручную).

Продвинутая очередь с использованием Redis

Redis — это отличный выбор для организации очередей благодаря своей скорости и встроенным структурам данных, таким как списки (lists) и сортированные множества (sorted sets).

Установка: Убедитесь, что у вас установлен Redis и PHP-расширение phpredis или predis.

Пример с использованием списков (Lists):

// Подключение к Redis
$redis = new Redis();
$redis->connect('127.0.0.1', 6379);

// Добавление задачи в очередь
$redis->rPush('queue:emails', json_encode([
    'to' => 'user@example.com',
    'subject' => 'Привет из Redis!'
]));

// Воркер — извлечение задачи (блокирующая операция)
while (true) {
    // BLPOP блокирует выполнение, пока не появится задача
    $jobData = $redis->blPop('queue:emails', 0);
    
    if ($jobData) {
        $data = json_decode($jobData[1], true);
        try {
            processJob($data);
        } catch (Exception $e) {
            // Логирование ошибки
            error_log("Job failed: " . $e->getMessage());
        }
    }
}

Очередь с задержкой (Delayed Jobs):

Для задач, которые нужно выполнить не сразу, используйте сортированные множества:

// Добавление задачи с задержкой в 10 секунд
$delay = time() + 10;
$redis->zAdd('delayed:emails', $delay, json_encode([
    'to' => 'user@example.com',
    'subject' => 'Отложенное письмо'
]));

// Воркер, проверяющий отложенные задачи
while (true) {
    // Получаем задачи, время выполнения которых наступило
    $jobs = $redis->zRangeByScore('delayed:emails', 0, time());
    
    foreach ($