RabbitMQ Streams

Advanced

Advanced Queue Types

RabbitMQ Streams (3.9+) provide a persistent, append-only log akin to Kafka topics; multiple consumers can read from any offset without deleting messages on acknowledgement.

Overview

RabbitMQ Streams, introduced in 3.9, add a Kafka-like persistent append-only log data structure to RabbitMQ. Unlike classic queues where acknowledged messages are deleted, streams retain all messages and allow any number of consumers to read from any position — first offset, last offset, a specific offset, or a timestamp. Messages are stored in fixed-size segment files and retention is configured by size or age. The stream protocol uses a dedicated binary protocol (port 5552) and Java/Python/Go client libraries for high-throughput scenarios. For Java and Spring, the rabbitmq-stream-java-client and spring-rabbitmq-stream starter provide a native API. Streams are ideal for event replay, fan-out to many independent consumers, and audit log scenarios where classic queues would be wasteful.

Declaring and publishing to a stream

Streams are declared as x-queue-type=stream. Publishing uses the standard AMQP protocol; consuming uses the dedicated Stream protocol client for offset-based access.

Java — declare stream + publish via stream client

<!-- pom.xml -->
<dependency>
  <groupId>com.rabbitmq</groupId>
  <artifactId>stream-client</artifactId>
  <version>0.15.0</version>
</dependency>

// Declare stream via AMQP (standard RabbitTemplate)
@Bean
public Queue orderEventStream() {
    return QueueBuilder.durable("order-events-stream")
        .withArgument("x-queue-type", "stream")
        .withArgument("x-max-length-bytes", 10_000_000_000L) // 10 GB max
        .withArgument("x-max-age", "7D")                     // 7-day retention
        .build();
}

// Publishing via standard AMQP (same as classic queue)
rabbitTemplate.convertAndSend("", "order-events-stream", event);

// OR use dedicated stream Environment for high throughput
Environment env = Environment.builder()
    .host("localhost")
    .port(5552)
    .build();

Producer producer = env.producerBuilder()
    .stream("order-events-stream")
    .build();

producer.send(
    producer.messageBuilder()
        .addData(serialize(event))
        .properties().messageId(UUID.randomUUID().toString())
        .messageBuilder().build(),
    confirmationStatus -> {
        if (!confirmationStatus.isConfirmed()) {
            log.warn("Message not confirmed");
        }
    }
);

Consuming from a specific offset

The key differentiator: each consumer independently chooses its starting offset. Multiple consumers can read the full stream from the beginning without interfering with each other.

Java — offset-based stream consumers

Environment env = Environment.builder().host("localhost").port(5552).build();

// Consumer 1: audit service — reads ALL historical messages from beginning
Consumer auditConsumer = env.consumerBuilder()
    .stream("order-events-stream")
    .offset(OffsetSpecification.first())   // replay everything
    .messageHandler((context, message) -> {
        auditService.record(deserialize(message.getBodyAsBinary()));
        // No ack needed — streams use offset tracking, not message ack
        context.storeOffset();             // persist consumer offset
    })
    .build();

// Consumer 2: analytics — reads only new messages from now
Consumer analyticsConsumer = env.consumerBuilder()
    .stream("order-events-stream")
    .offset(OffsetSpecification.next())    // only new messages
    .name("analytics-consumer")            // named consumer for offset persistence
    .autoTrackingStrategy()                // auto-commit offset every N messages
    .messageHandler((context, message) -> {
        analyticsService.process(deserialize(message.getBodyAsBinary()));
    })
    .build();

// Consumer 3: start from a specific timestamp
Consumer recoveryConsumer = env.consumerBuilder()
    .stream("order-events-stream")
    .offset(OffsetSpecification.timestamp(
        Instant.now().minus(Duration.ofHours(2)).toEpochMilli()))
    .build();

Streams vs classic queues — when to use which

Streams outperform classic queues for fan-out and replay scenarios; classic queues are better for work queue patterns where each message should be processed by exactly one consumer.

Java — streams vs classic queues decision guide

// Use RabbitMQ Streams when:
// ✓ Multiple independent consumers need the same event (fan-out)
// ✓ New consumers need to replay historical events
// ✓ Audit log — all events must be retained for N days
// ✓ Very high throughput (streams use zero-copy I/O)

// Use Classic Queues when:
// ✓ Work queue pattern — each message processed by exactly one consumer
// ✓ Task fan-out to competing consumers sharing load
// ✓ Priority queues (x-max-priority)
// ✓ Already using Spring AMQP @RabbitListener (stream client is separate)

// Retention comparison
// Classic queue:  message deleted on ack or DLX
// Stream:         retained for x-max-age or x-max-length-bytes (like Kafka)

// Throughput comparison (single node, local)
// Classic queue:  ~50K msg/s publish, ~50K msg/s consume
// Stream:         ~1M+ msg/s publish with batching, zero-copy segment reads

Key Points to Remember

1Streams are append-only logs — acknowledged messages are NOT deleted; retention is by age or total size.
2Each consumer independently tracks its own offset — replay, fan-out, and concurrent reads all work without interference.
3The dedicated stream protocol (port 5552) outperforms AMQP for high-throughput stream consumers.
4Named consumers with autoTrackingStrategy persist offsets server-side — safe for consumer restarts.
5Streams require the stream_queue feature flag (RabbitMQ 3.9+) and rabbitmq_stream plugin enabled.
6Use streams for fan-out to many consumers; use classic queues for competing-consumers (work queue) patterns.

Interview Questions

How do RabbitMQ Streams differ from classic queues in message lifecycle?

EasyVMware

How would you replay all events from the last 24 hours for a new service using RabbitMQ Streams?

MediumAmazon

Compare RabbitMQ Streams to Apache Kafka — what can each do that the other cannot?

HardConfluent

How does offset tracking work in RabbitMQ Streams and what happens if a consumer crashes?

HardNetflix

When would you choose RabbitMQ Streams over a classic fanout exchange with multiple queues?

MediumZalando

Ask Aria about RabbitMQ Streams

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Lazy Queues

Delayed Message Exchange