Skip to content

RabbitMQ Streams#

RabbitMQ streams are persistent, replicated append-only logs. Unlike regular queues, consuming a message does not remove it, so different consumers can read the same messages or replay an earlier part of the log. Streams are useful for large fan-outs, replay, high throughput, and large backlogs. See the RabbitMQ streams guide for the full feature and operational details.

Declare and consume a stream#

Set queue_type=QueueType.STREAM when declaring a RabbitQueue. Streams are always durable, and RabbitMQ requires a non-zero consumer prefetch. The example sets that prefetch on the broker's default channel.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
from faststream import FastStream, Logger
from faststream.rabbit import RabbitBroker, RabbitQueue, QueueType, Channel

broker = RabbitBroker(default_channel=Channel(prefetch_count=10))
app = FastStream(broker)

queue = RabbitQueue(
    name="test-stream",
    durable=True,
    queue_type=QueueType.STREAM,
    arguments={
        "x-max-age": "7D",
        "x-max-length-bytes": 20_000_000_000,
    },
)


@broker.subscriber(
    queue,
    consume_args={"x-stream-offset": "first"},
)
async def handle(msg, logger: Logger) -> None:
    logger.info(msg)


@app.after_startup
async def test() -> None:
    await broker.publish("Hi!", queue)

Choose the starting offset#

Pass the RabbitMQ x-stream-offset consumer argument through consume_args to choose where a subscriber starts reading. The example uses first, which replays the stream from its first available message.

Supported values include:

  • first to start at the first available message.
  • last to start at the last stored chunk of messages, not only the final message.
  • next, or no offset argument, to wait for messages published after the subscriber starts.
  • An integer offset, a timestamp, or a relative time interval to start at a specific position.

Configure retention#

Because consumption does not delete messages, configure retention so a stream does not grow without limit. The example keeps no more than seven days or 20 GB of data by passing x-max-age and x-max-length-bytes in the queue's arguments dictionary. You can also configure these limits with a RabbitMQ policy. RabbitMQ removes the oldest stream segments when a retention limit is reached.