Если ваш сайт начал тормозить из-за отправки писем, генерации отчетов или обработки загрузок — скорее всего, пора задуматься об очередях. Очереди — это способ сказать тяжелой задаче: «Подожди, я сейчас занят, сделаю тебя позже». В этой статье мы разберем, как устроены очереди в PHP, когда они нужны, и напишем простую, но рабочую систему своими руками. Без магии и сложных терминов.

Digital-студия WNDER

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

Представьте, что вы — официант в ресторане. Если вы бежите на кухню за каждым блюдом сразу, как только его заказали, то к концу вечера у вас ноги отвалятся, а половина заказов остынет. Гораздо умнее записывать заказы в блокнот и передавать их повару пачками. Вот это и есть очередь.

В программировании очередь — это структура данных, куда вы складываете задачи, которые нужно выполнить, но не обязательно прямо сейчас. PHP-скрипт, который обрабатывает запрос пользователя, не должен ждать, пока отправится 100 писем. Он просто кладет их в очередь и отвечает пользователю: «Готово, я все сделаю». А отдельный воркер (worker) — это программа, которая запускается отдельно и разбирает очередь.

Когда очередь реально нужна?

  • Отправка email-рассылок или уведомлений.
  • Генерация PDF-отчетов, превью, обработка изображений.
  • Выполнение длительных запросов к внешним API.
  • Обновление поискового индекса или кэша.
  • Импорт больших файлов.

Если ваш скрипт делает что-то дольше 1-2 секунд и это не критично для ответа пользователю — смело отправляйте в очередь.

Виды очередей: от простого к сложному

Существует несколько способов организовать очередь в PHP. Разберем основные, от самых простых до продвинутых.

Тип очереди Инструменты Плюсы Минусы
Файловая Файлы на диске Простота, не требует внешних сервисов Медленно, небезопасно, нет масштабирования
База данных MySQL, PostgreSQL Надежно, транзакции, легко стартовать Нагрузка на БД, нужна очистка старых задач
Redis / Memcached Redis, Memcached Быстро, удобные структуры данных, TTL Данные в памяти, нужен отдельный сервер
RabbitMQ / Kafka RabbitMQ, Kafka Мощно, гибко, масштабируется, гарантии доставки Сложно в настройке, требует администрирования

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

Создаем простую очередь на базе MySQL

Начнем с самого простого и понятного варианта — таблицы в MySQL. Это идеально для старта, когда у вас нет инфраструктуры Redis, но хочется уже сегодня избавиться от тормозов.

Создадим таблицу для задач:

CREATE TABLE queue (
    id INT AUTO_INCREMENT PRIMARY KEY,
    task_type VARCHAR(100) NOT NULL, -- тип задачи: email, report, image
    payload TEXT NOT NULL, -- данные задачи в JSON
    status ENUM('pending', 'processing', 'done', 'failed') DEFAULT 'pending',
    attempts INT DEFAULT 0,
    available_at DATETIME NOT NULL, -- когда можно брать задачу
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) ENGINE=InnoDB;

Теперь напишем класс для работы с очередью. Будем использовать PDO для безопасности.

class Queue {
    private $pdo;

    public function __construct(PDO $pdo) {
        $this->pdo = $pdo;
    }

    /**
     * Добавить задачу в очередь
     */
    public function push(string $taskType, array $payload, int $delay = 0): void {
        $stmt = $this->pdo->prepare(
            'INSERT INTO queue (task_type, payload, available_at) VALUES (?, ?, ?)'
        );
        $availableAt = (new DateTime())->modify("+{$delay} seconds")->format('Y-m-d H:i:s');
        $stmt->execute([$taskType, json_encode($payload), $availableAt]);
    }

    /**
     * Получить следующую задачу для обработки (с блокировкой)
     */
    public function pull(): ?array {
        // Начинаем транзакцию, чтобы не обработать задачу дважды
        $this->pdo->beginTransaction();
        try {
            $stmt = $this->pdo->prepare(
                'SELECT * FROM queue WHERE status = "pending" AND available_at <= NOW() ORDER BY id ASC LIMIT 1 FOR UPDATE'
            );
            $stmt->execute();
            $task = $stmt->fetch(PDO::FETCH_ASSOC);

            if ($task) {
                $this->pdo->prepare('UPDATE queue SET status = "processing" WHERE id = ?')
                    ->execute([$task['id']]);
            }

            $this->pdo->commit();
            return $task ?: null;
        } catch (Exception $e) {
            $this->pdo->rollBack();
            throw $e;
        }
    }

    /**
     * Отметить задачу выполненной или проваленной
     */
    public function complete(int $id, bool $success = true): void {
        $status = $success ? 'done' : 'failed';
        $this->pdo->prepare('UPDATE queue SET status = ? WHERE id = ?')
            ->execute([$status, $id]);
    }

    /**
     * Вернуть задачу в очередь при ошибке (с увеличением попыток)
     */
    public function retry(int $id, int $maxAttempts = 3): void {
        $stmt = $this->pdo->prepare('SELECT attempts FROM queue WHERE id = ?');
        $stmt->execute([$id]);
        $attempts = (int)$stmt->fetchColumn();

        if ($attempts >= $maxAttempts) {
            $this->complete($id, false);
        } else {
            $this->pdo->prepare('UPDATE queue SET status = "pending", attempts = attempts + 1, available_at = DATE_ADD(NOW(), INTERVAL 1 MINUTE) WHERE id = ?')
                ->execute([$id]);
        }
    }
}

Теперь как это использовать? Создаем скрипт-воркер, который будет запускаться, например, каждую минуту через cron.

// worker.php
require_once 'Queue.php';

$pdo = new PDO('mysql:host=localhost;dbname=test', 'user', 'password');
$queue = new Queue($pdo);

while (true) {
    $task = $queue->pull();
    if (!$task) {
        // Нет задач — спим и ждем
        sleep(5);
        continue;
    }

    try {
        // Обработка задачи в зависимости от типа
        switch ($task['task_type']) {
            case 'send_email':
                $data = json_decode($task['payload'], true);
                sendEmail($data['to'], $data['subject'], $data['body']);
                break;
            case 'generate_report':
                generateReport($task['id']);
                break;
            default:
                throw new Exception('Unknown task type');
        }

        $queue->complete($task['id']);
    } catch (Exception $e) {
        $queue->retry($task['id']);
    }
}

Не забудьте настроить cron, чтобы воркер запускался постоянно или по расписанию. Но лучше использовать супервизор (например, Supervisor), который будет держать процесс живым.

Лайфхак: Не запускайте воркер через cron каждую минуту, если задачи приходят нечасто. Это лишняя нагрузка. Лучше использовать бесконечный цикл, как в примере, и управлять им через Supervisor. Так вы сэкономите ресурсы сервера.

Улучшаем очередь с помощью Redis

MySQL-очередь — это хорошо, но есть нюансы: каждая операция требует запроса к БД, а это медленно. Redis — это хранилище в памяти, которое работает в разы быстрее. Плюс у Redis есть удобные структуры данных: списки, множества, сортированные множества.

Для очереди мы будем использовать список (list). Задача добавляется в конец списка через команду LPUSH, а воркер забирает с начала через RPOP. Но есть проблема: если воркер упал, задача потеряется. Поэтому используем команду BRPOPLPUSH — она атомарно забирает задачу и кладет ее в отдельный список «в обработке». Если задача выполнена, удаляем из «в обработке», если нет — возвращаем обратно.

Вот пример класса на PHP с использованием phpredis:

class RedisQueue {
    private $redis;
    private $queueName;
    private $processingName;

    public function __construct(\Redis $redis, string $queueName = 'queue') {
        $this->redis = $redis;
        $this->queueName = $queueName;
        $this->processingName = $queueName . ':processing';
    }

    public function push(string $taskType, array $payload): void {
        $task = json_encode([
            'type' => $taskType,
            'payload' => $payload,
            'created_at' => time()
        ]);
        $this->redis->rPush($this->queueName, $task);
    }

    public function pull(): ?array {
        // Блокирующе достаем задачу и перемещаем в processing
        $taskJson = $this->redis->brPopLPush($this->queueName, $this->processingName, 5);
        if (!$taskJson) {
            return null;
        }
        $task = json_decode($taskJson, true);
        $task['id'] = md5($taskJson); // простой идентификатор для отслеживания
        return $task;
    }

    public function complete(string $id): void {
        // Удаляем задачу из processing (поиск по содержимому неэффективен, поэтому храним id)
        // Для простоты пропустим детали, в реальности используют отдельный список с ID.
        // Здесь мы просто удалим все задачи с таким же содержимым? Нет, лучше хранить в задаче уникальный ключ.
        // Поэтому в push добавим 'uid' => uniqid(), а здесь будем удалять по значению.
        // Для примера опустим реализацию.
    }

    public function retry(string $id, int $delay = 5): void {
        // Возвращаем задачу обратно в очередь с задержкой
        // В Redis для задержек используют сортированные множества (ZSet)
        // Но это уже сложнее, упростим.
    }
}

На самом деле, для полноценной работы с Redis лучше использовать библиотеку Enqueue или Laravel Queue, которые уже все реализуют. Но если вы хотите понять суть — то вот она.

Как правильно обрабатывать ошибки и повторы

Очередь — это не просто «складирование задач». Нужно продумать, что делать, если задача упала. Основные принципы:

  • Каждая задача должна иметь лимит попыток. Обычно 3-5.
  • Между попытками нужна задержка, чтобы внешний сервис успел восстановиться.
  • Если задача так и не выполнилась — она должна попасть в отдельный список «failed» для ручного разбора.
  • Всегда логируйте ошибки с контекстом задачи.
Правило: Никогда не теряйте данные. Если задача не выполнилась после всех попыток, сохраните ее в отдельную таблицу или файл, чтобы можно было разобраться в причине. Иначе вы рискуете потерять заказы или письма.

Заключение

Мы разобрали, как сделать простую очередь на MySQL и чуть-чуть заглянули в Redis. Для большинства проектов этого достаточно. Главное — понять принцип: тяжелые задачи уходят в сторону, а пользователь получает ответ мгновенно.

Если вам нужно быстрое решение без написания своего велосипеда — используйте готовые библиотеки: Laravel Queue, Symfony Messenger, RabbitMQ. Но если хочется разобраться, как все устроено под капотом — наш пример на MySQL будет отличной отправной точкой.

Помните: очередь — это не панацея. Если у вас нет узких мест, не нужно усложнять. Но когда они появятся — вы уже будете знать, что делать.

Студия WNDER