Redis Streams as a Message Queue Replacement: A Practical Guide with Examples

Introduction: Why Redis Streams Are More Than Just Pub/Sub

Most developers know Redis as a high-performance cache. Many use Redis Pub/Sub for simple notifications. But Redis Streams is a separate, significantly more powerful data structure introduced in Redis 5.0 that fundamentally changes how you approach message queuing in a microservices architecture.

What makes Redis Streams fundamentally different from Pub/Sub? In Pub/Sub, messages are not persisted: if a subscriber is unavailable at the time of publishing, it will miss the message forever. Redis Streams, on the other hand, store all messages in an ordered log with unique IDs, support consumer groups with acknowledgment (ACK), and allow replaying message history — much like Apache Kafka, but without the operational complexity.

This article is aimed at backend developers who already use Redis as a cache, and at microservices architects looking for a pragmatic alternative to RabbitMQ or Kafka for moderate-complexity tasks. We'll cover everything: from basic concepts to production-ready Go code and Laravel integration.

Core Concepts of Redis Streams

Stream — A Message Log

A Stream in Redis is an append-only log: a data structure to which you can only add new entries. Each entry has a unique ID in the format millisecondsTime-sequenceNumber (e.g., 1700000000000-0) and a set of arbitrary key-value fields.

# Add a message to the stream
XADD orders * user_id 42 product_id 101 action purchase

# Read the last 10 messages
XRANGE orders - + COUNT 10

# Get the stream length
XLEN orders

The * symbol in the XADD command means "auto-generate the ID." You can also pass an explicit ID, which is useful when replicating data from external systems.

Consumer Groups — Parallel Processing

A Consumer Group is a mechanism that allows multiple consumers to jointly process a single stream, where each message is delivered to only one consumer in the group. This is the key difference from a regular XREAD, where every reader receives all messages.

# Create a consumer group
XGROUP CREATE orders processing-group $ MKSTREAM

# Read new messages as consumer worker-1
XREADGROUP GROUP processing-group worker-1 COUNT 5 BLOCK 2000 STREAMS orders >

# Acknowledge message processing (ACK)
XACK orders processing-group 1700000000000-0 1700000000001-0

The > symbol means "give me only unprocessed messages." The $ symbol when creating a group means "start from new messages"; 0 means start from the very beginning of the stream.

Pending Entries List (PEL)

When a consumer fetches a message via XREADGROUP, it enters the Pending Entries List — a list of messages that have been delivered but not yet acknowledged. This is critical for delivery guarantees: if a worker crashes before sending an ACK, the message remains in the PEL and can be handed off to another worker.

# View pending messages
XPENDING orders processing-group - + 10

# Detailed info on specific pending entries
XPENDING orders processing-group - + 10 worker-1

Comparing Redis Streams with RabbitMQ and Kafka

Before writing any code, it's important to understand when Redis Streams is the right choice and when a dedicated broker is a better fit.

Use Redis Streams when:

• You already have Redis in your infrastructure and want to avoid adding a new component
• Your load is in the thousands, not millions, of messages per second
• You need a simple event-driven architecture for microservices
• Low latency is important
• You don't need complex routing

Stick with Kafka if:

• Data volumes are hundreds of gigabytes per day
• You need multi-level replication and exactly-once guarantees
• Long-term event storage is required (event sourcing at scale)

Stick with RabbitMQ if:

• You need complex routing via exchanges and bindings
• AMQP protocol support is important
• Your team knows RabbitMQ well and migration isn't justified

Hands-On: Producer/Consumer Implementation in Go

Let's look at a real-world example: an order processing system for an online store. The producer publishes order creation events, and multiple consumer workers process them in parallel.

Installing Dependencies

go mod init orders-processor
go get github.com/redis/go-redis/v9

Producer: Publishing Events

package main

import (
    "context"
    "fmt"
    "log"
    "time"

    "github.com/redis/go-redis/v9"
)

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    ctx := context.Background()

    // Create consumer group on startup (ignore error if it already exists)
    err := rdb.XGroupCreateMkStream(ctx, "orders", "processing-group", "$").Err()
    if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
        log.Fatalf("Failed to create consumer group: %v", err)
    }

    // Publish order events
    for i := 1; i <= 100; i++ {
        msgID, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: "orders",
            Values: map[string]interface{}{
                "order_id":   fmt.Sprintf("ORD-%04d", i),
                "user_id":    i * 10,
                "amount":     float64(i) * 99.9,
                "created_at": time.Now().Unix(),
            },
        }).Result()
        if err != nil {
            log.Printf("Failed to publish order: %v", err)
            continue
        }
        fmt.Printf("Published order ORD-%04d with ID: %s\n", i, msgID)
        time.Sleep(10 * time.Millisecond)
    }
}

Consumer: Processing Messages with ACK

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "syscall"
    "time"

    "github.com/redis/go-redis/v9"
)

const (
    streamName    = "orders"
    groupName     = "processing-group"
    blockDuration = 2 * time.Second
    maxRetries    = 3
)

func main() {
    consumerName := os.Getenv("CONSUMER_NAME")
    if consumerName == "" {
        consumerName = "worker-1"
    }

    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
    })
    ctx, cancel := context.WithCancel(context.Background())

    // Graceful shutdown
    sigCh := make(chan os.Signal, 1)
    signal.Notify(sigCh, syscall.SIGTERM, syscall.SIGINT)
    go func() {
        <-sigCh
        fmt.Println("Shutting down...")
        cancel()
    }()

    // Process pending messages first (after restart)
    processPendingMessages(ctx, rdb, consumerName)

    // Main loop for processing new messages
    for {
        select {
        case <-ctx.Done():
            return
        default:
            messages, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
                Group:    groupName,
                Consumer: consumerName,
                Streams:  []string{streamName, ">"},
                Count:    10,
                Block:    blockDuration,
            }).Result()
            if err != nil {
                if err == redis.Nil {
                    continue // timeout, no new messages
                }
                log.Printf("Error reading from stream: %v", err)
                time.Sleep(time.Second)
                continue
            }

            for _, stream := range messages {
                for _, msg := range stream.Messages {
                    if err := processOrder(msg); err != nil {
                        log.Printf("Failed to process message %s: %v", msg.ID, err)
                        continue
                    }
                    // Acknowledge successful processing
                    if err := rdb.XAck(ctx, streamName, groupName, msg.ID).Err(); err != nil {
                        log.Printf("Failed to ACK message %s: %v", msg.ID, err)
                    }
                }
            }
        }
    }
}

func processOrder(msg redis.XMessage) error {
    orderID := msg.Values["order_id"]
    amount := msg.Values["amount"]
    fmt.Printf("Processing order %s, amount: %s\n", orderID, amount)
    // Your business logic here: saving to DB, sending email, etc.
    time.Sleep(50 * time.Millisecond) // simulate work
    return nil
}

func processPendingMessages(ctx context.Context, rdb *redis.Client, consumerName string) {
    pending, err := rdb.XPendingExt(ctx, &redis.XPendingExtArgs{
        Stream:   streamName,
        Group:    groupName,
        Start:    "-",
        Stop:     "+",
        Count:    100,
        Consumer: consumerName,
    }).Result()
    if err != nil {
        return
    }
    for _, p := range pending {
        if p.RetryCount >= maxRetries {
            log.Printf("Message %s exceeded retry limit, moving to DLQ", p.ID)
            // Dead Letter Queue logic
            continue
        }
        // Reclaim and process
        msgs, err := rdb.XClaim(ctx, &redis.XClaimArgs{
            Stream:   streamName,
            Group:    groupName,
            Consumer: consumerName,
            MinIdle:  30 * time.Second,
            Messages: []string{p.ID},
        }).Result()
        if err != nil {
            continue
        }
        for _, msg := range msgs {
            if err := processOrder(msg); err == nil {
                rdb.XAck(ctx, streamName, groupName, msg.ID)
            }
        }
    }
}

Error Handling and Delivery Guarantees

ACK and At-Least-Once Guarantee

Redis Streams provides an at-least-once delivery guarantee: a message is considered processed only after an explicit XACK. If a worker crashes before sending an ACK, the message remains in the PEL and can be reclaimed (XCLAIM) by another worker.

XPENDING: Inspecting Unacknowledged Messages

# Summary of pending messages
XPENDING orders processing-group - + 100

# Result:
# 1) "1700000001234-0"
# 2) "worker-1"
# 3) (integer) 85000     <-- time since last delivery in ms
# 4) (integer) 2         <-- number of delivery attempts

XCLAIM: Reclaiming Stuck Messages

XCLAIM allows one worker to take over a message that another worker has been holding for too long. This is critical for ensuring reliability when workers fail.

# Reclaim messages idle for more than 60 seconds
XCLAIM orders processing-group worker-2 60000 1700000001234-0

# Automatically reclaim a batch of stuck messages (Redis 6.2+)
XAUTOCLAIM orders processing-group worker-2 60000 0-0 COUNT 10

Dead Letter Queue (DLQ)

For messages that cannot be processed after N attempts, it is recommended to implement a Dead Letter Queue — a separate stream for "poisoned" messages:

# Move a problematic message to the DLQ
XADD orders-dlq * original_id 1700000001234-0 order_id ORD-0042 reason "processing_failed" attempts 3
XACK orders processing-group 1700000001234-0
XDEL orders 1700000001234-0

Limiting Stream Size

Redis stores all messages in memory (when operating without disk offloading). To prevent overflow, use the MAXLEN parameter:

# Limit stream to 100,000 messages (approximate)
XADD orders MAXLEN ~ 100000 * order_id ORD-0001 ...

# Or trim manually on a schedule
XTRIM orders MAXLEN ~ 100000

Integrating Redis Streams into a Laravel Application

Laravel has a built-in Redis driver for queues, but it works via RPUSH/LPOP (List), not Streams. Let's see how to create a custom Streams driver for the Laravel Queue.

Creating the Driver

<?php
// app/Queue/RedisStreamConnector.php
namespace App\Queue;

use Illuminate\Queue\Connectors\ConnectorInterface;

class RedisStreamConnector implements ConnectorInterface
{
    public function connect(array $config): RedisStreamQueue
    {
        return new RedisStreamQueue(
            app('redis')->connection($config['connection'] ?? 'default'),
            $config['queue'] ?? 'default',
            $config['group'] ?? 'laravel-workers',
            $config['consumer'] ?? gethostname(),
        );
    }
}

<?php
// app/Queue/RedisStreamQueue.php
namespace App\Queue;

use Illuminate\Contracts\Queue\Queue;
use Illuminate\Queue\Queue as BaseQueue;
use Illuminate\Redis\Connections\Connection;

class RedisStreamQueue extends BaseQueue implements Queue
{
    public function __construct(
        protected Connection $redis,
        protected string $default,
        protected string $group,
        protected string $consumer,
    ) {}

    public function push($job, $data = '', $queue = null): mixed
    {
        $queue = $this->getQueue($queue);
        $payload = $this->createPayload($job, $queue, $data);

        return $this->redis->command('xadd', [
            $queue, '*',
            'payload', $payload,
            'attempts', 0,
        ]);
    }

    public function pop($queue = null): ?\Illuminate\Contracts\Queue\Job
    {
        $queue = $this->getQueue($queue);
        $this->ensureGroupExists($queue);

        $results = $this->redis->command('xreadgroup', [
            'GROUP', $this->group, $this->consumer,
            'COUNT', 1,
            'BLOCK', 2000,
            'STREAMS', $queue, '>',
        ]);

        if (empty($results[$queue])) {
            return null;
        }

        [$messageId, $values] = $results[$queue][0];
        $payload = json_decode($values['payload'], true);

        return new RedisStreamJob(
            $this->container, $this, $this->redis,
            $queue, $this->group, $messageId, $payload,
        );
    }

    public function ack(string $queue, string $messageId): void
    {
        $this->redis->command('xack', [$queue, $this->group, $messageId]);
    }

    protected function ensureGroupExists(string $queue): void
    {
        try {
            $this->redis->command('xgroup', ['CREATE', $queue, $this->group, '$', 'MKSTREAM']);
        } catch (\Exception $e) {
            // Group already exists — that's fine
        }
    }

    protected function getQueue(?string $queue): string
    {
        return 'stream:' . ($queue ?? $this->default);
    }

    // Other required Queue interface methods...
    public function size($queue = null): int { return 0; }
    public function later($delay, $job, $data = '', $queue = null): mixed { return null; }
    public function bulk($jobs, $data = '', $queue = null): void {}
}

Registering the Driver in AppServiceProvider

<?php
// app/Providers/AppServiceProvider.php
public function boot(): void
{
    Queue::extend('redis-stream', function () {
        return new \App\Queue\RedisStreamConnector();
    });
}

Configuration in config/queue.php

'connections' => [
    'redis-stream' => [
        'driver'     => 'redis-stream',
        'connection' => 'default',
        'queue'      => 'default',
        'group'      => 'laravel-workers',
        'consumer'   => env('QUEUE_CONSUMER_NAME', gethostname()),
        'retry_after' => 90,
    ],
],

After this, all standard Laravel jobs (dispatch, Queue::push) will work through Redis Streams, and you get all the benefits: ACK, PEL, replay, and consumer groups.

Scaling: Multiple Consumer Groups and Partitioning

Multiple Consumer Groups for Different Purposes

One of the most powerful features of Redis Streams: a single stream can be read by multiple independent consumer groups. This enables a fan-out pattern without duplicating data:

# Three different services read the same order events
XGROUP CREATE orders notification-service $ MKSTREAM
XGROUP CREATE orders analytics-service $ MKSTREAM  
XGROUP CREATE orders inventory-service $ MKSTREAM

Each group receives all messages and processes them independently. This is fundamentally different from round-robin within a single group, where each message is received by only one worker.

Partitioning via Multiple Streams

Redis is single-threaded by default, so for horizontal scaling, multiple named streams are used (analogous to Kafka partitions):

# Producer determines the partition by user_id hash
func getPartition(userID int, numPartitions int) string {
    return fmt.Sprintf("orders:partition:%d", userID % numPartitions)
}

// Publish to a partition
partition := getPartition(userID, 8) // 8 partitions
rdb.XAdd(ctx, &redis.XAddArgs{
    Stream: partition,
    Values: orderData,
})

When using Redis Cluster, partitions are automatically distributed across different nodes, providing linear throughput scaling.

Automatic Worker Scaling

In Kubernetes, you can configure an HPA (Horizontal Pod Autoscaler) based on PEL length or consumer group lag. Metrics are exported via redis_exporter to Prometheus:

# Example HPA based on custom metrics
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: order-worker-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: order-worker
  minReplicas: 2
  maxReplicas: 20
  metrics:
  - type: External
    external:
      metric:
        name: redis_stream_pending_entries
        selector:
          matchLabels:
            stream: orders
      target:
        type: AverageValue
        averageValue: "100"

Monitoring Redis Streams: XINFO, Metrics, and Alerts

XINFO: Built-In Diagnostics

# General stream information
XINFO STREAM orders

# Consumer group information
XINFO GROUPS orders

# Consumer information within a group
XINFO CONSUMERS orders processing-group

# Full details (Redis 7.0+)
XINFO STREAM orders FULL COUNT 10

Key metrics to monitor:

• pending-messages — number of messages in the PEL (should be close to 0 under normal conditions)
• lag — difference between the last ID in the stream and the last ID read by the group
• consumers count — number of active workers
• idle time — time a consumer has been idle

Monitoring via redis_exporter + Prometheus

Use redis_exporter to export metrics to Prometheus. Add stream metric monitoring to your configuration:

# prometheus.yml
scrape_configs:
  - job_name: 'redis'
    static_configs:
      - targets: ['redis-exporter:9121']
    params:
      stream-groups:
        - orders
        - payments

Alerts in Alertmanager

# Alert for high lag
- alert: RedisStreamHighLag
  expr: redis_stream_group_lag{stream="orders"} > 1000
  for: 5m
  labels:
    severity: warning
  annotations:
    summary: "Redis Stream lag is too high"
    description: "Consumer group lag: {{ $value }} messages"

# Alert for stale pending messages
- alert: RedisStreamStalePending  
  expr: redis_stream_group_pending{stream="orders"} > 100
  for: 10m
  labels:
    severity: critical

Grafana Dashboard

Recommended panels for a Redis Streams dashboard:

1. Stream length (XLEN) over time
2. Consumer group lag per group
3. Pending entries count
4. ACK rate (messages/sec)
5. Oldest pending message age
6. Consumer count per group

Conclusion

In 2026, Redis Streams is a mature, high-performance, and operationally simple tool for message queuing in microservices architecture. If you already have Redis in your stack, adopting Streams requires no new infrastructure component, and the learning curve is significantly lower than Kafka's.

Key takeaways from this article:

• Redis Streams differs from Pub/Sub through persistence and delivery guarantees via ACK/PEL
• Consumer Groups enable horizontal scaling of message processing
• XCLAIM and XAUTOCLAIM resolve the problem of stuck messages when workers fail
• Go integration takes a few hours; in Laravel, you can implement a custom queue driver
• Monitoring lag and pending entries is the key to reliable production operation

Redis Streams is not a replacement for Kafka in petabyte-scale systems, but it's an excellent alternative to RabbitMQ for most microservices use cases. Start with one stream, one consumer group, and two workers — and you'll be surprised how far that takes you.