In-Memory Message Queue with Redis

In the previous articles, [Kafka and the Producer-Consumer Model](https://xx/Kafka and the Producer-Consumer Model) and [RabbitMQ and the Producer-Consumer Model](https://xx/RabbitMQ and the Producer-Consumer Model), we explored two classic implementations of the producer-consumer model: Kafka and RabbitMQ. Kafka excels in high-throughput big data scenarios with its distributed architecture, while RabbitMQ provides reliable messaging and flexible routing in microservice environments. This article focuses on a lighter, faster solution: Redis.

What is Redis?

Redis (REmote DIctionary Server) is an open-source, in-memory key-value data store often used as a cache or fast-response NoSQL database.

It supports various data types, including strings, hashes, lists, sets, and sorted sets, and provides rich operations like push/pop, add/remove, and set intersections, all of which are atomic.

Redis achieves exceptional performance because all data is stored and accessed in memory. It also supports persistence through RDB snapshots and AOF (Append-Only File). Moreover, Redis supports distributed scaling via clustering for handling larger workloads.

Another useful feature of Redis is its built-in pub/sub mechanism, which allows it to act as a producer-consumer system through the PUBLISH and SUBSCRIBE commands. While Redis is not a dedicated message queue, its diverse data structures and in-memory speed make it an excellent lightweight solution for high-performance messaging.

Redis and the Producer-Consumer Model

In [From Basics to Advanced – The Complete Producer-Consumer Model Guide](https://xx/From Basics to Advanced – The Complete Producer-Consumer Model Guide), we learned that the core idea is: producers write to a buffer (queue), and consumers read from it asynchronously, decoupling the two sides.

Redis wasn’t designed exclusively for this model, but it provides several data structures that make it very suitable for implementing it. The three most common Redis-based implementations are:

1. List-based Queue

Redis Lists are implemented as doubly linked lists. Producers can LPUSH items to the list, and consumers can BRPOP (blocking pop) items off. This is the simplest and most widely used Redis queue pattern.

2. Pub/Sub

Redis supports native publish/subscribe messaging: producers PUBLISH to a channel, and consumers SUBSCRIBE to one or more channels.
However, messages are not persisted—if a subscriber disconnects or is offline, it will miss any published messages.

3. Streams

Redis Streams provide a log-like data structure similar to Kafka.
They support features such as persistence, replication, consumer groups, message acknowledgment, and message tracking. Streams are the officially recommended Redis structure for messaging use cases.

These options make Redis a powerful tool for lightweight messaging, especially in scenarios where low latency is critical.

Simulating Producers and Consumers

Let’s implement two Redis-based queues in Go: one using List, and the other using Stream.

List-Based Queue

Producer:

package main

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })

    for i := 1; i <= 10; i++ {
        data := fmt.Sprintf("data-%d", i)
        err := rdb.LPush(ctx, "list-queue", data).Err()
        if err != nil {
            fmt.Printf("Failed to produce: %v\n", err)
            continue
        }
        fmt.Printf("Produced: %s\n", data)
    }
}

Consumer:

package main

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
    "time"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })

    for {
        result, err := rdb.BRPop(ctx, 5*time.Second, "list-queue").Result()
        if err == redis.Nil {
            fmt.Println("No data found, waiting...")
            continue
        } else if err != nil {
            fmt.Printf("Error consuming data: %v\n", err)
            break
        }

        if len(result) == 2 {
            fmt.Printf("Consumed: %s\n", result[1])
        }
    }
}

Stream-Based Queue

Producer:

package main

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
    "strconv"
    "time"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })

    streamKey := "stream-queue"

    for i := 1; i <= 10; i++ {
        id, err := rdb.XAdd(ctx, &redis.XAddArgs{
            Stream: streamKey,
            Values: map[string]interface{}{
                "order_id": fmt.Sprintf("OID-%03d", i),
                "amount":   strconv.Itoa(100 + i),
            },
        }).Result()

        if err != nil {
            fmt.Printf("Failed to produce: %v\n", err)
        } else {
            fmt.Printf("Produced message ID: %s\n", id)
        }

        time.Sleep(500 * time.Millisecond)
    }
}

Consumer (Consumer Group):

package main

import (
    "context"
    "fmt"
    "github.com/redis/go-redis/v9"
    "time"
)

var ctx = context.Background()

func main() {
    rdb := redis.NewClient(&redis.Options{
        Addr: "localhost:6379",
        DB:   0,
    })

    streamKey := "stream-queue"
    groupName := "mygroup"
    consumerName := "consumer-1"

    err := rdb.XGroupCreateMkStream(ctx, streamKey, groupName, "$").Err()
    if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
        panic(err)
    }

    fmt.Println("Start consuming...")

    for {
        streams, err := rdb.XReadGroup(ctx, &redis.XReadGroupArgs{
            Group:    groupName,
            Consumer: consumerName,
            Streams:  []string{streamKey, ">"},
            Count:    5,
            Block:    5 * time.Second,
        }).Result()

        if err == redis.Nil {
            continue
        } else if err != nil {
            fmt.Printf("ReadGroup error: %v\n", err)
            time.Sleep(time.Second)
            continue
        }

        for _, stream := range streams {
            for _, msg := range stream.Messages {
                fmt.Printf("Consumed ID: %s, Values: %v\n", msg.ID, msg.Values)
                if err := rdb.XAck(ctx, streamKey, groupName, msg.ID).Err(); err != nil {
                    fmt.Printf("ACK failed: %v\n", err)
                }
            }
        }
    }
}

Conclusion

Redis is not a dedicated message queue, but thanks to its powerful data structures and high-speed in-memory operations, it can effectively serve as a lightweight messaging platform.

By using Lists or Streams, Redis offers flexible, fast, and simple implementations of the producer-consumer model, making it a great choice for scenarios involving:

• Lightweight asynchronous workflows
• Task distribution
• High-speed message processing with low latency

Compared with Kafka and RabbitMQ, Redis stands out in lightweight, low-cost, and high-speed use cases.