События и очереди: Valkey, NATS и RabbitMQ.

Когда сайт передаёт работу фоновому процессу — отправить письмо, уменьшить картинку, провести оплату заказа — или один сервис сообщает остальным, что что-то произошло, сообщение идёт через очередь или брокер сообщений. В проекте их может быть три, каждый добавляется как сервис: Обзор → Сервисы → Добавить сервис.

Если не уверены, начните с Valkey для фоновых задач: большинство фреймворков поддерживают его сразу. Выбирайте RabbitMQ, когда сообщения нужно распределять между многими очередями и ни одно не должно потеряться, и NATS, когда сервисы обмениваются множеством небольших сообщений или запрашивают друг у друга ответы.

Valkey: очереди задач и Streams

Valkey — Redis-совместимое хранилище, тот же управляемый сервис, что и в статье Базы данных. Первый Valkey приложения даёт каждому сервису с кодом REDIS_URL, REDIS_HOST, REDIS_PORT и REDIS_PASSWORD; каждый Valkey также даёт свою RIVERVO_<NAME>_URL.

Три способа передавать через него сообщения:

• Очереди задач — очереди Laravel с драйвером redis и Horizon, BullMQ, Celery, RQ. Библиотека хранит задачи в Valkey; фоновый процесс берёт их по одной и повторяет те, что завершились ошибкой.
• Streams с группами потребителей — XADD добавляет событие в поток, потребители группы читают его через XREADGROUP и подтверждают через XACK. События хранятся: остановленный потребитель продолжает с того места, где остановился, неподтверждённое событие можно забрать снова, а новая группа может прочитать поток с начала.
• Pub/Sub — PUBLISH и SUBSCRIBE доставляют сообщение только тем подписчикам, которые подключены в этот момент. Сообщение не сохраняется: подписчик, который перезапускается или ещё не подключился, его пропустит. Используйте Pub/Sub только там, где потеря сообщения ничему не вредит (например, сигнал «обновить»), а для всего остального — Streams или очередь.

Резервные копии Valkey создаются каждый день, как и у других баз данных; к нему также можно подключиться извне, если открыть доступ. См. Базы данных.

NATS

NATS — лёгкая и быстрая система обмена сообщениями, с включённым JetStream:

• Публикация и подписка — сервис публикует сообщение в тему (subject), например orders.created, и его получает каждый подписчик этой темы. Без JetStream сообщение отправляется без гарантии доставки, как в Pub/Sub выше.
• Запрос и ответ — один сервис спрашивает, другой отвечает. Когда отвечают несколько экземпляров в одной группе (queue group), NATS распределяет запросы между ними.
• JetStream — надёжные потоки: сообщения хранятся на диске сервиса, потребитель (consumer) помнит, что он уже прочитал, и каждое сообщение доставляется как минимум один раз, пока его не подтвердят.
• Хранилище «ключ — значение» и объектное хранилище на основе JetStream — для небольших общих настроек, блокировок или файлов между сервисами.

Переменные. Первый NATS приложения даёт каждому сервису с кодом:

NATS_URL=nats://<user>:<password>@<service name>:4222

Каждый сервис NATS также даёт RIVERVO_<NAME>_URL с адресом того же вида — NAME это имя сервиса заглавными буквами, дефисы заменяются на _: сервис с именем events даёт RIVERVO_EVENTS_URL.

JetStream хранит потоки на диске сервиса; потоки, которые хранятся в памяти, могут занимать примерно четверть памяти сервиса.

RabbitMQ

RabbitMQ 4.1 — классический брокер сообщений: отправители публикуют сообщения в exchange, exchange по ключам маршрутизации направляет их в очереди, потребители подтверждают каждое сообщение, а отклонённое или просроченное сообщение может уйти в dead-letter exchange. Для очередей, сообщения которых должны пережить перезапуск, используйте quorum-очереди (аргумент x-queue-type: quorum) и публикуйте сообщения как постоянные (persistent).

Переменные. Первый RabbitMQ приложения даёт каждому сервису с кодом:

%2f в конце AMQP_URL — это виртуальный хост по умолчанию /, записанный так, как того требуют адреса подключения. Каждый сервис RabbitMQ также даёт свою RIVERVO_<NAME>_URL с адресом того же вида.

API управления. HTTP API управления слушает порт 15672 внутри частной сети приложения: сервисы приложения обращаются к нему по адресу http://<service name>:15672 с теми же пользователем и паролем, например:

curl -u "$RABBITMQ_USER:$RABBITMQ_PASSWORD" "http://$RABBITMQ_HOST:15672/api/queues"

Страницы управления в интернете нет. На странице сервиса в панели показан короткий список его очередей только для просмотра: сообщения, готовые к доставке, неподтверждённые сообщения и потребители.

Только внутренний доступ

У NATS и RabbitMQ нет публичного адреса, и к ним нельзя подключиться снаружи приложения — с вашего компьютера, другого сервера или из другого приложения. В отличие от PostgreSQL, MySQL и Valkey, доступ извне для них не предусмотрен. Сервисы, которые публикуют и получают сообщения, работают как сервисы того же приложения: веб-сервисы, фоновые процессы и задачи по расписанию.

Второй сервис того же вида

Привычные имена (REDIS_*, NATS_URL, AMQP_URL и RABBITMQ_*) получают только первые Valkey, NATS и RabbitMQ приложения. Следующий сервис того же вида добавляет только свою RIVERVO_<NAME>_URL; читайте эту переменную в коде или задайте привычным именам её значения на вкладке Переменные. Переменные доходят до сервиса со следующим развёртыванием — см. Переменные окружения.

Laravel

Valkey. Задайте на вкладке Переменные:

QUEUE_CONNECTION=redis

Запускайте php artisan queue:work — или php artisan horizon с Horizon — как фоновый процесс. См. Laravel.

RabbitMQ. Встроенного драйвера RabbitMQ в Laravel нет, и Rivervo не меняет QUEUE_CONNECTION сам. Чтобы использовать RabbitMQ для очередей Laravel:

1. Установите пакет: composer require vladimir-yuldashev/laravel-queue-rabbitmq.
2. Добавьте подключение rabbitmq в config/queue.php так, как показано в README пакета для вашей версии. Оно читает RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USER, RABBITMQ_PASSWORD и RABBITMQ_VHOST, которые у приложения уже есть:

'rabbitmq' => [
    'driver' => 'rabbitmq',
    'hosts' => [
        [
            'host' => env('RABBITMQ_HOST', '127.0.0.1'),
            'port' => env('RABBITMQ_PORT', 5672),
            'user' => env('RABBITMQ_USER', 'guest'),
            'password' => env('RABBITMQ_PASSWORD', 'guest'),
            'vhost' => env('RABBITMQ_VHOST', '/'),
        ],
    ],
],

3. Задайте QUEUE_CONNECTION=rabbitmq на вкладке Переменные и разверните приложение. Фоновый процесс остаётся прежним — php artisan queue:work.

NATS. Официального драйвера очередей Laravel для NATS нет. Используйте в своём коде NATS-клиент для PHP — например, basis-company/nats — и передайте ему хост, порт, пользователя и пароль из NATS_URL. Прежде чем строить на нём решение, проверьте, что клиент поддерживает нужные вам возможности (JetStream, «ключ — значение»).

Node.js

BullMQ (Valkey)

import { Queue, Worker } from 'bullmq'
import IORedis from 'ioredis'
 
const connection = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null })
 
// в веб-сервисе
const emails = new Queue('emails', { connection })
await emails.add('welcome', { userId: 42 }, { attempts: 3 })
 
// в фоновом процессе
new Worker('emails', async (job) => {
  await sendWelcome(job.data.userId)
}, { connection })

BullMQ требует maxRetriesPerRequest: null у подключения, которое используют его обработчики.

Valkey Streams

import IORedis from 'ioredis'
 
const redis = new IORedis(process.env.REDIS_URL)
 
// отправитель
await redis.xadd('orders', '*', 'id', '42')
 
// потребитель группы «billing»
await redis.xgroup('CREATE', 'orders', 'billing', '0', 'MKSTREAM').catch((err) => {
  if (!String(err).includes('BUSYGROUP')) throw err // группа уже есть
})
const me = process.env.HOSTNAME ?? 'worker'
for (;;) {
  const res = await redis.xreadgroup('GROUP', 'billing', me, 'COUNT', 10, 'BLOCK', 5000, 'STREAMS', 'orders', '>')
  for (const [, entries] of res ?? []) {
    for (const [id, fields] of entries) {
      await handleOrder(fields)
      await redis.xack('orders', 'billing', id)
    }
  }
}

Прочитанная, но не подтверждённая запись остаётся в группе как ожидающая; XAUTOCLAIM передаёт её другому потребителю.

NATS (nats.js)

import { connect, JSONCodec, AckPolicy } from 'nats'
 
const nc = await connect({ servers: process.env.NATS_URL })
const jc = JSONCodec()
 
// один раз: поток для тем orders.* и постоянный потребитель
const jsm = await nc.jetstreamManager()
await jsm.streams.add({ name: 'ORDERS', subjects: ['orders.>'] })
await jsm.consumers.add('ORDERS', { durable_name: 'billing', ack_policy: AckPolicy.Explicit })
 
// отправитель
const js = nc.jetstream()
await js.publish('orders.created', jc.encode({ id: 42 }))
 
// потребитель
const consumer = await js.consumers.get('ORDERS', 'billing')
for await (const m of await consumer.consume()) {
  await handleOrder(jc.decode(m.data))
  m.ack()
}

Запрос и ответ: один сервис отвечает через nc.subscribe('prices.get', { queue: 'prices', callback: (err, msg) => msg.respond(jc.encode(price)) }), другой спрашивает через await nc.request('prices.get', jc.encode({ sku: 'A1' }), { timeout: 2000 }).

RabbitMQ (amqplib)

import amqp from 'amqplib'
 
const conn = await amqp.connect(process.env.AMQP_URL)
const ch = await conn.createChannel()
await ch.assertQueue('emails', { durable: true, arguments: { 'x-queue-type': 'quorum' } })
 
// отправитель
ch.sendToQueue('emails', Buffer.from(JSON.stringify({ userId: 42 })), { persistent: true })
 
// потребитель: не больше 10 неподтверждённых сообщений одновременно
await ch.prefetch(10)
await ch.consume('emails', async (msg) => {
  try {
    await sendWelcome(JSON.parse(msg.content.toString()).userId)
    ch.ack(msg)
  } catch {
    ch.nack(msg, false, false) // в dead-letter exchange, если он есть у очереди
  }
})

Каждый сервис, который объявляет очередь, должен объявлять её с одинаковыми аргументами.

Python

Celery (Valkey или RabbitMQ)

import os
from celery import Celery
 
app = Celery("tasks", broker=os.environ["REDIS_URL"])  # или os.environ["AMQP_URL"]
 
@app.task(acks_late=True)
def send_welcome(user_id):
    ...

Запускайте celery -A tasks worker --loglevel=info как фоновый процесс. Celery принимает адреса amqp:// как есть. С RabbitMQ последние версии Celery умеют объявлять quorum-очереди (task_default_queue_type = "quorum") — см. документацию Celery для вашей версии.

RQ (Valkey)

import os
from redis import Redis
from rq import Queue
 
queue = Queue("emails", connection=Redis.from_url(os.environ["REDIS_URL"]))
queue.enqueue(send_welcome, 42)

Обработчик: rq worker emails --url "$REDIS_URL".

NATS (nats-py)

import asyncio, json, os
import nats
from nats.errors import TimeoutError
 
async def main():
    nc = await nats.connect(os.environ["NATS_URL"])
    js = nc.jetstream()
    await js.add_stream(name="ORDERS", subjects=["orders.>"])
 
    await js.publish("orders.created", json.dumps({"id": 42}).encode())
 
    sub = await js.pull_subscribe("orders.>", durable="billing")
    while True:
        try:
            msgs = await sub.fetch(10, timeout=5)
        except TimeoutError:
            continue
        for msg in msgs:
            await handle_order(json.loads(msg.data))
            await msg.ack()
 
asyncio.run(main())

RabbitMQ (pika)

import json, os
import pika
 
conn = pika.BlockingConnection(pika.URLParameters(os.environ["AMQP_URL"]))
ch = conn.channel()
ch.queue_declare(queue="emails", durable=True, arguments={"x-queue-type": "quorum"})
 
# отправитель
ch.basic_publish(exchange="", routing_key="emails",
                 body=json.dumps({"user_id": 42}),
                 properties=pika.BasicProperties(delivery_mode=2))
 
# потребитель
def on_message(ch, method, properties, body):
    send_welcome(json.loads(body)["user_id"])
    ch.basic_ack(delivery_tag=method.delivery_tag)
 
ch.basic_qos(prefetch_count=10)
ch.basic_consume(queue="emails", on_message_callback=on_message)
ch.start_consuming()

Повторная доставка

Очереди и брокеры доставляют сообщение повторно, если обработчик остановился до того, как подтвердил его, — во время развёртывания, перезапуска или восстановления. Пишите потребителей так, чтобы повторная обработка того же сообщения ничему не вредила: сохраняйте идентификатор каждого обработанного сообщения или делайте само действие повторяемым (например, «установить статус „оплачен“» вместо «добавить платёж»).

Ресурсы и цена

Цена каждого зависит от его памяти, CPU и диска, как у любого сервиса, — см. Сервисы проекта. Ресурсы можно изменить позже в Настройки сервиса.

Резервные копии и восстановление

Резервные копии NATS и RabbitMQ создаются каждый день; мы храним копии за последние 7 дней и 4 недели, как и у других сервисов.

• Резервная копия копирует файлы данных сервиса во время его работы. Для RabbitMQ в той же копии сохраняется и экспорт его определений: пользователи, виртуальные хосты, очереди, exchange, привязки и политики.
• Восстановления на точный момент нет — оно есть только у PostgreSQL и MySQL. Восстанавливается одна из ежедневных копий.
• При восстановлении сервис ненадолго останавливается, файлы возвращаются на место, и сервис запускается снова. Сообщения, которые пришли или были подтверждены около времени создания копии, могут отсутствовать или прийти повторно — ещё одна причина делать потребителей безопасными для повторов.
• Для RabbitMQ: если скопированные файлы не удаётся запустить, сервис запускается только с определениями — очереди, exchange и пользователи на месте, но сообщения из копии потеряны.

Обслуживание и перенос

У NATS и RabbitMQ нет времени обслуживания, которое нужно задавать. Автоматически на другой сервер они не переносятся; если это когда-нибудь понадобится, поддержка согласует перенос с вами.