Как использовать списки Redis (LPUSH, RPOP) в качестве очередей сообщений

Списки Redis могут стать полезной небольшой очередью сообщений, когда вам нужно что-то более легкое, чем RabbitMQ, Kafka или полноценный фреймворк для фоновых задач. Обычный шаблон прост: продюсеры добавляют задачи с помощью LPUSH, а воркеры забирают задачи с помощью RPOP или BRPOP.

Именно эта простота заставляет людей обращаться к этому решению. Веб-запрос может поместить задачу по отправке email в Redis и быстро вернуться. Воркер может подхватить эту задачу мгновение спустя. Вам не нужна топология брокера, обменники, топики или новый операционный стек. Однако вам нужно честно признать компромисс: обычный RPOP удаляет сообщение до того, как воркер завершит работу. Если воркер упадет в неподходящий момент, эта задача будет потеряна, если вы не построите вокруг нее шаблон подтверждения.

Понимание списков Redis как очередей

Список Redis — это упорядоченная коллекция строк. Его можно рассматривать как последовательность элементов, и Redis предоставляет команды для добавления или удаления элементов как с начала, так и с конца списка. Эта двунаправленная природа делает списки идеально подходящими для реализации очередей.

Постановка в очередь (добавление сообщений): Мы можем добавлять новые сообщения в очередь, помещая их в один конец списка. Команда LPUSH помещает элементы в начало (левую сторону) списка.
Извлечение из очереди (обработка сообщений): Мы можем извлекать и удалять сообщения из очереди, выталкивая их с другого конца списка. Команда RPOP извлекает элементы из конца (правой стороны) списка.

Эта конкретная комбинация (LPUSH для постановки и RPOP для извлечения) создает очередь «первым пришел — первым обслужен» (FIFO), что является наиболее распространенным и ожидаемым поведением для очереди сообщений.

Основные команды: LPUSH и RPOP

Давайте углубимся в две основные команды, которые составляют основу нашей очереди сообщений Redis.

`LPUSH key value [value ...]`

Команда LPUSH вставляет одно или несколько строковых значений в начало (левую сторону) списка, хранящегося по ключу key. Если ключ key не существует, создается новый список, и значения вставляются в него.

Пример:

Представьте, что у вас есть задача, которую нужно обработать, например, отправить email. Вы можете поместить эту задачу как сообщение в список Redis с именем email_tasks.

# Помещаем одну задачу по отправке email
LPUSH email_tasks "{'to': '[email protected]', 'subject': 'Welcome!', 'body': 'Thanks for signing up!'}"

# Помещаем другую задачу, она будет размещена перед предыдущей
LPUSH email_tasks "{'to': '[email protected]', 'subject': 'New User Registration', 'body': 'A new user has registered.'}"

После этих команд список email_tasks будет выглядеть так (от начала к концу):

1) "{'to': '[email protected]', 'subject': 'New User Registration', 'body': 'A new user has registered.'}"
2) "{'to': '[email protected]', 'subject': 'Welcome!', 'body': 'Thanks for signing up!'}"

`RPOP key`

Команда RPOP удаляет и возвращает последний элемент (из конца, правой стороны) списка, хранящегося по ключу key. Если список пуст, она возвращает nil.

Пример:

Процесс-воркер может периодически опрашивать список email_tasks на наличие новых задач с помощью RPOP.

# Воркер пытается получить задачу
RPOP email_tasks

Если список не пуст, RPOP вернет последний помещенный элемент (который является первым элементом с конца). В нашем примере выше первый вызов RPOP вернет:

"{'to': '[email protected]', 'subject': 'Welcome!', 'body': 'Thanks for signing up!'}"

Последующие вызовы затем извлекут следующую доступную задачу с конца.

Создание базовой системы очереди сообщений

Давайте опишем типичный поток простой очереди сообщений с использованием LPUSH и RPOP.

1. Продюсер (постановка задач в очередь)

Любая часть вашего приложения, которой нужно разгрузить работу, может выступать в роли продюсера. Он формирует сообщение (часто строку JSON, представляющую детали задачи) и помещает его в список Redis с помощью LPUSH.

Логика продюсера (концептуальный пример на Python):

import redis
import json

r = redis.Redis(host='localhost', port=6379, db=0)

def send_email_task(to_email, subject, body):
    task_message = {
        'type': 'send_email',
        'payload': {
            'to': to_email,
            'subject': subject,
            'body': body
        }
    }
    # LPUSH добавляет в начало списка 'email_queue'
    r.lpush('email_queue', json.dumps(task_message))
    print(f"Pushed email task to queue: {to_email}")

# Пример использования:
send_email_task('[email protected]', 'Hello from Producer', 'This is a test message.')
send_email_task('[email protected]', 'Important Update', 'New features available.')

2. Потребитель (извлечение и обработка задач)

Процессы-воркеры, работающие независимо, будут непрерывно отслеживать список Redis на наличие новых сообщений. Они используют RPOP для извлечения и удаления сообщения из очереди.

Логика потребителя (концептуальный пример на Python):

import redis
import json
import time

r = redis.Redis(host='localhost', port=6379, db=0)

def process_tasks():
    while True:
        # RPOP пытается получить сообщение из конца списка 'email_queue'
        message_bytes = r.rpop('email_queue')
        if message_bytes:
            message_str = message_bytes.decode('utf-8')
            try:
                task = json.loads(message_str)
                print(f"Processing task: {task}")
                # Имитация обработки задачи
                if task.get('type') == 'send_email':
                    print(f"  -> Sending email to {task['payload']['to']}...")
                    # Замените на реальную логику отправки email
                    time.sleep(1) # Имитация работы
                    print(f"  -> Email sent to {task['payload']['to']}.")
                else:
                    print(f"  -> Unknown task type: {task.get('type')}")
            except json.JSONDecodeError:
                print(f"Error decoding JSON: {message_str}")
            except Exception as e:
                print(f"Error processing task {message_str}: {e}")
        else:
            # Нет сообщения, подождать немного перед следующим опросом
            # print("No tasks available, waiting...")
            time.sleep(0.5)

if __name__ == "__main__":
    print("Worker started. Waiting for tasks...")
    process_tasks()

Когда вы запускаете продюсера, он помещает сообщения. Когда вы запускаете потребителя, он начинает их забирать и обрабатывать. Порядок обработки будет соответствовать порядку, в котором они были помещены (FIFO), потому что LPUSH добавляет в начало, а RPOP удаляет из конца.

Вопросы надежности

Хотя LPUSH и RPOP обеспечивают базовый механизм очереди, создание производственной очереди означает принятие решений о том, что должно происходить, когда воркеры падают, задания терпят неудачу или продюсеры отправляют некорректные полезные нагрузки.

1. Потеря сообщений во время обработки

Если процесс-воркер падает после того, как RPOP удалил сообщение, но до того, как он завершил обработку, это сообщение теряется. Чтобы предотвратить это:

Используйте BRPOP вместо активного опроса: BRPOP блокируется до тех пор, пока в списке не появится элемент или не истечет тайм-аут. Само по себе это не делает обработку надежной, но это предотвращает пробуждение воркеров каждые несколько миллисекунд только для того, чтобы обнаружить пустую очередь.
```
# Блокирующее извлечение справа с тайм-аутом 0 (блокировка на неопределенный срок)
BRPOP email_queue 0
```
Используйте список обработки для подтверждения: Распространенный шаблон — атомарно переместить сообщение из email_queue в email_processing, обработать его и удалить из email_processing только после успешного выполнения работы. Если воркер умирает, отдельный процесс-«жнец» может искать устаревшие элементы в списке обработки и перемещать их обратно в основную очередь. RPOPLPUSH — это классическая команда для этого шаблона, а в более новых версиях Redis также доступны LMOVE/BLMOVE.

2. Обработка неудачных задач

Что произойдет, если задача не будет выполнена во время обработки (например, из-за временной проблемы с сетью или неверных данных)?

Механизмы повторных попыток: Реализуйте логику повторных попыток внутри воркера. После нескольких неудач переместите задачу в список 'failed_tasks' для ручной проверки.
Очередь недоставленных сообщений (DLQ): Выделенный список Redis (или другое хранилище), куда отправляются сообщения, которые неоднократно не удалось обработать. Это важно для отладки и восстановления.

3. Несколько потребителей

Если у вас есть несколько экземпляров воркеров, потребляющих из одной очереди, RPOP (и BRPOP) гарантирует, что каждое сообщение будет обработано только одним воркером. Это связано с тем, что RPOP атомарно удаляет элемент.

4. Порядок сообщений

Хотя LPUSH и RPOP создают очередь FIFO, эта гарантия настолько же сильна, насколько сильна ваша логика обработки. Если потребители повторно ставят в очередь неудачные сообщения без надлежащей обработки или если вы вводите другие операции, строгий порядок FIFO может быть нарушен.

5. Формат полезной нагрузки и идемпотентность

Относитесь к телу сообщения как к небольшому контракту. JSON распространен, потому что его легко просматривать в redis-cli, но используйте валидный JSON, а не словари в стиле Python с одинарными кавычками:

{"type":"send_email","id":"email-1842","payload":{"to":"[email protected]","template":"welcome"}}

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

6. Длина очереди и обратное давление

Следите за длиной очереди с помощью LLEN email_queue. Растущая очередь не обязательно плоха; это может просто означать, что воркеры догоняют после всплеска трафика. Очередь, которая растет часами, обычно означает, что продюсеры быстрее потребителей, воркеры выходят из строя или одна медленная зависимость сдерживает все.

На практике я предпочитаю оповещать как о возрасте, так и о длине. Списки Redis не хранят время постановки в очередь отдельно, поэтому поместите временную метку в полезную нагрузку, если возраст задания имеет значение:

{"type":"resize_image","id":"img-991","created_at":"2026-05-24T08:15:00Z","payload":{"image_id":991}}

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

Продвинутые методы (кратко)

RPOPLPUSH: Атомарно извлекает сообщение из одного списка и помещает его в другой (например, в список 'processing'). Это ключевая команда для реализации надежной обработки с подтверждениями.
BLPOP / BRPOP с несколькими ключами: Блокируется и извлекает из первого списка, который стал непустым. Полезно для потребления из нескольких очередей.
Скрипты Lua: Для сложных атомарных операций, которые не покрывает RPOPLPUSH, можно использовать скрипты Lua, чтобы гарантировать выполнение критических последовательностей команд без прерывания.

Более надежная форма воркера

Для чего-то более важного, чем уведомление с максимальными усилиями, избегайте простого воркера «извлеки и надейся». Более безопасная форма выглядит так:

RPOPLPUSH email_queue email_processing

Воркер получает сообщение, и Redis уже переместил его в email_processing. После того, как email отправлен и приложение зафиксировало успех, воркер удаляет эту точную полезную нагрузку из списка обработки:

LREM email_processing 1 '{"type":"send_email","id":"email-1842"}'

Это все еще не идеальная корпоративная очередь. LREM должен соответствовать полезной нагрузке, большие списки обработки могут стать неудобными, и вам нужен процесс-«жнец», который знает, когда сообщение достаточно старое для повторной попытки. Но это полезным образом меняет режим отказа. Сбой воркера больше не удаляет единственную копию задания.

Если вы используете этот подход, поместите метаданные повторных попыток в сообщение или сохраните их рядом с идентификатором сообщения в другом ключе. Например, «жнец» может переместить устаревшее сообщение обратно в email_queue первые несколько раз, а затем переместить его в email_failed после превышения лимита повторных попыток. Это дает вам место для проверки «ядовитых» сообщений вместо того, чтобы наблюдать, как одна и та же плохая полезная нагрузка терпит неудачу вечно.

Когда списки Redis — неправильная очередь

Списки Redis легко понять, но они не всегда являются правильным инструментом. Если вам нужны отложенные задания, приоритеты заданий, запланированные повторные попытки, видимость рабочего процесса или долгосрочная история аудита, библиотека заданий или выделенный брокер могут в конечном итоге потребовать меньше работы. Также стоит рассмотреть Redis Streams, поскольку они имеют группы потребителей и семантику подтверждения, встроенные в тип данных.

Мне все еще нравятся списки для небольших внутренних очередей: генерация миниатюр, прогрев кэша, разветвление вебхуков, где исходная система может повторить попытку, или простые фоновые задания, принадлежащие одному приложению. Как только несколько команд начинают зависеть от контракта очереди, запишите ожидания по доставке. «Как минимум один раз», «не более одного раза» и «максимальные усилия» — это не академические термины во время инцидента. Они определяют, приемлемы ли дубликаты, терпимы ли потерянные сообщения и сколько механизмов восстановления вам нужно.

Списки Redis — хороший выбор, когда вам нужна небольшая, понятная очередь и у вас уже есть Redis. Начните с LPUSH и BRPOP для простой фоновой работы. Добавьте список обработки, счетчик повторных попыток и список недоставленных сообщений, как только потеря задания будет иметь значение. Если вам нужно отложенное планирование, приоритеты, разветвление, длительное хранение или строгие гарантии доставки между многими сервисами, обычно это тот момент, когда специализированная очередь становится проще в использовании, чем растущая куча соглашений Redis.