Events and queues: Valkey, NATS and RabbitMQ.

When a site hands work to a worker — send an email, resize an image, charge an order — or one service tells the others that something happened, the message goes through a queue or a message broker. A project can have three of them, each added as a service: Overview → Services → Add a service.

If you are not sure, start with Valkey for background jobs: most frameworks support it out of the box. Choose RabbitMQ when messages must be routed between many queues and none may be lost, and NATS when services exchange many small messages or ask each other for answers.

Valkey: job queues and Streams

Valkey is a Redis-compatible store — the same managed service as in Databases. The first Valkey of the application gives every code service REDIS_URL, REDIS_HOST, REDIS_PORT and REDIS_PASSWORD; every Valkey also gives its own RIVERVO_<NAME>_URL.

Three ways to send messages through it:

• Job queues — Laravel queues with the redis driver and Horizon, BullMQ, Celery, RQ. The library keeps the jobs in Valkey; a worker takes them one by one and retries the ones that fail.
• Streams with consumer groups — XADD appends an event to a stream, consumers of a group read it with XREADGROUP and confirm it with XACK. Events are stored: a consumer that was stopped continues where it left off, an event that was not confirmed can be taken again, and a new group can read the stream from the start.
• Pub/Sub — PUBLISH and SUBSCRIBE deliver a message only to the subscribers connected at that moment. The message is not stored: a subscriber that is restarting or not yet connected misses it. Use Pub/Sub only where a lost message does no harm (for example a «refresh» signal), and Streams or a queue for everything else.

Valkey is backed up every day like the other databases; it can also be reached from outside when you open access. See Databases.

NATS

NATS is a lightweight, fast messaging system, with JetStream turned on:

• Publish and subscribe — a service publishes to a subject such as orders.created, every subscriber of the subject gets the message. Without JetStream this is fire-and-forget, like Pub/Sub above.
• Request and reply — a service asks, another answers. When several instances answer as a queue group, NATS spreads the requests among them.
• JetStream — durable streams: messages are stored on the service's disk, a consumer remembers what it has read, and every message is delivered at least once until it is acknowledged.
• Key-value and object store built on JetStream — for small shared settings, locks or files between services.

Variables. The first NATS of the application gives every code service:

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

Every NATS service also gives RIVERVO_<NAME>_URL with an address of the same form — NAME is the service's name in capital letters, with dashes turned into _: a service called events gives RIVERVO_EVENTS_URL.

JetStream keeps its streams on the service's disk; streams kept in memory may use about a quarter of the service's memory.

RabbitMQ

RabbitMQ 4.1 is a classic message broker: producers publish to exchanges, exchanges route the messages to queues by routing keys, consumers acknowledge each message, and a message that is rejected or expires can go to a dead-letter exchange. For queues whose messages must survive a restart, use quorum queues (the argument x-queue-type: quorum) and publish messages as persistent.

Variables. The first RabbitMQ of the application gives every code service:

%2f at the end of AMQP_URL is the default virtual host /, written the way connection addresses require. Every RabbitMQ service also gives its own RIVERVO_<NAME>_URL with an address of the same form.

Management API. The management HTTP API listens on port 15672 inside the application's private network: the application's services reach it at http://<service name>:15672 with the same user and password, for example:

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

There is no management page on the internet. The service's page in the panel shows a short read-only list of its queues: messages ready, unacknowledged and consumers.

Internal only

NATS and RabbitMQ have no public address and cannot be reached from outside the application — from your computer, another server or another application. Unlike PostgreSQL, MySQL and Valkey, access from outside is not offered for them. The services that publish and consume messages run as services of the same application: web services, background workers and scheduled jobs.

Second service of the same kind

Only the first Valkey, NATS and RabbitMQ of the application get the usual names (REDIS_*, NATS_URL, AMQP_URL and RABBITMQ_*). The next one of the same kind adds only its own RIVERVO_<NAME>_URL; read that variable in your code, or set the usual names to its values on the Variables tab. Variables reach a service with its next deploy — see Environment variables.

Laravel

Valkey. Set on the Variables tab:

QUEUE_CONNECTION=redis

Run php artisan queue:work — or php artisan horizon with Horizon — as a background worker. See Laravel.

RabbitMQ. Laravel has no RabbitMQ driver built in, and Rivervo does not change QUEUE_CONNECTION by itself. To use RabbitMQ for Laravel queues:

1. Install the package: composer require vladimir-yuldashev/laravel-queue-rabbitmq.
2. Add the rabbitmq connection to config/queue.php as the package's README shows for your version. It reads RABBITMQ_HOST, RABBITMQ_PORT, RABBITMQ_USER, RABBITMQ_PASSWORD and RABBITMQ_VHOST, which the application already has:

'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. Set QUEUE_CONNECTION=rabbitmq on the Variables tab and deploy. The worker stays php artisan queue:work.

NATS. There is no first-party Laravel queue driver for NATS. Use a NATS client for PHP in your own code — for example basis-company/nats — and give it the host, port, user and password from NATS_URL. Check that the client supports the features you need (JetStream, key-value) before you build on it.

Node.js

BullMQ (Valkey)

import { Queue, Worker } from 'bullmq'
import IORedis from 'ioredis'
 
const connection = new IORedis(process.env.REDIS_URL, { maxRetriesPerRequest: null })
 
// in the web service
const emails = new Queue('emails', { connection })
await emails.add('welcome', { userId: 42 }, { attempts: 3 })
 
// in the background worker
new Worker('emails', async (job) => {
  await sendWelcome(job.data.userId)
}, { connection })

BullMQ needs maxRetriesPerRequest: null on the connection its workers use.

Valkey Streams

import IORedis from 'ioredis'
 
const redis = new IORedis(process.env.REDIS_URL)
 
// producer
await redis.xadd('orders', '*', 'id', '42')
 
// consumer of the group «billing»
await redis.xgroup('CREATE', 'orders', 'billing', '0', 'MKSTREAM').catch((err) => {
  if (!String(err).includes('BUSYGROUP')) throw err // the group already exists
})
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)
    }
  }
}

An entry that was read but not acknowledged stays pending in the group; XAUTOCLAIM hands it to another consumer.

NATS (nats.js)

import { connect, JSONCodec, AckPolicy } from 'nats'
 
const nc = await connect({ servers: process.env.NATS_URL })
const jc = JSONCodec()
 
// once: a stream for the subjects orders.* and a durable consumer
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 })
 
// producer
const js = nc.jetstream()
await js.publish('orders.created', jc.encode({ id: 42 }))
 
// consumer
const consumer = await js.consumers.get('ORDERS', 'billing')
for await (const m of await consumer.consume()) {
  await handleOrder(jc.decode(m.data))
  m.ack()
}

Request and reply: one service answers with nc.subscribe('prices.get', { queue: 'prices', callback: (err, msg) => msg.respond(jc.encode(price)) }), another asks with 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' } })
 
// producer
ch.sendToQueue('emails', Buffer.from(JSON.stringify({ userId: 42 })), { persistent: true })
 
// consumer: at most 10 unacknowledged messages at a time
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) // to the dead-letter exchange, if the queue has one
  }
})

Every service that declares the queue must declare it with the same arguments.

Python

Celery (Valkey or RabbitMQ)

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

Run celery -A tasks worker --loglevel=info as a background worker. Celery accepts amqp:// addresses as they are. With RabbitMQ, recent Celery versions can declare quorum queues (task_default_queue_type = "quorum") — see Celery's documentation for your version.

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)

The worker: 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"})
 
# producer
ch.basic_publish(exchange="", routing_key="emails",
                 body=json.dumps({"user_id": 42}),
                 properties=pika.BasicProperties(delivery_mode=2))
 
# consumer
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()

Delivered more than once

Queues and brokers deliver a message again when a worker stops before it confirms it — during a deploy, a restart or a restore. Write consumers so that handling the same message twice does no harm: keep the ID of every processed message, or make the action itself repeatable (for example «set the status to paid» instead of «add a payment»).

Resources and price

Each is priced by its memory, CPU and disk like any service — see Services of a project. Resources can be changed later in the service's Settings.

Backups and restoring

NATS and RabbitMQ are backed up every day; we keep the last 7 days and 4 weeks, like the other services.

• The backup copies the service's data files while it runs. For RabbitMQ the same backup also holds an export of its definitions: users, virtual hosts, queues, exchanges, bindings and policies.
• There is no restoring to an exact moment — that is only for PostgreSQL and MySQL. You restore one of the daily copies.
• A restore stops the service for a short time, puts the files back and starts it again. Messages that arrived or were acknowledged around the time of the backup may be missing or delivered again — another reason to make consumers safe to repeat.
• For RabbitMQ, if the copied files cannot be started, the service starts with the definitions only: the queues, exchanges and users are there, but the messages from the backup are lost.

Maintenance and moves

NATS and RabbitMQ have no maintenance time to set. They are not moved to another server automatically; if that is ever needed, support arranges it with you.