Using Redis Streams: Commands, Sample Application and Best Practices

What are Redis Streams?

Redis Streams is a data structure in Redis designed for managing a high-throughput, message-oriented system, often used for event sourcing, messaging queues, or data pipelines. It allows you to append, read, and consume messages in a log-like fashion. Redis Streams can hold an ordered sequence of messages, where each message consists of a unique ID and associated data.

Redis Streams provides features like consumer groups, message acknowledgments, and message trimming, making it highly suitable for scenarios like event-driven architectures and scalable data pipelines. It supports the efficient handling of large volumes of messages with low-latency reads and writes. Compared with Redis Pub/Sub, Streams are considered more feature-rich and reliable in general, which we will explore more in this guide.

Key Features of Redis Streams

Redis Streams offers several powerful features that make it useful for managing high-volume, real-time data:

• Message Ordering: Redis Streams maintain strict ordering of messages. Each message within the stream is assigned a unique ID based on the timestamp and a sequence number when it was written, ensuring that messages can be processed in the exact order they were added.
• Consumer Groups: Consumer groups allow multiple consumers to read messages from the same stream in parallel. Each consumer within a group processes a distinct subset of messages, enabling load balancing and horizontal scaling.
• Message Acknowledgment: Consumers in a group can acknowledge the processing of messages. This acknowledgment helps ensure that messages are not lost and can be retried if necessary, improving the reliability of the system.
• Stream Trimming: Redis Streams allows trimming of old messages based on various criteria like the number of messages or their age. This helps to prevent memory overload by removing messages that are no longer relevant.
• Blocking Reads: Redis Streams support blocking reads, where consumers can wait for new messages to arrive, typically with a timeout. This feature is ideal for real-time systems, where consumers don’t have to constantly poll from the stream.
• Efficient Data Retrieval: Redis Streams provide efficient methods for reading data, whether it's retrieving a range of messages or consuming messages from a particular point. This makes it suitable for time-sensitive data processing applications.
• Persistence and Durability: As with other Redis data structures, Streams benefit from Redis's persistence mechanisms (RDB, AOF), providing durability in case of server restarts. However, persistence settings can be customized based on the use case to balance performance and durability.

Core Redis Stream Commands

Redis Streams comes with a set of commands that allow you to interact with the stream, enabling message insertion, retrieval, and management. Below are some of the key commands you’ll use when working with Redis Streams:

1. XADD

Adds a message to the stream. The command allows you to specify data in the form of key-value pairs, and it automatically generates a unique message ID based on the current time. You can also provide a custom message ID if needed.

Example:

redis$> XADD mystream * name John age 30

This adds a message with a timestamp-based ID containing the fields name=John and age=30 to the stream mystream.

2. XREAD

Reads messages from a stream. It allows you to retrieve messages from one or more streams, specifying a start point for the read operation. You can use this command to fetch messages in real-time or batch processing.

Example:

redis$> XREAD BLOCK 0 STREAMS mystream 0

This command reads messages from the mystream starting at the beginning (0), blocking indefinitely until new messages arrive.

3. XREADGROUP

Reads messages from a stream within a consumer group, which must be created beforehand by using XGROUP CREATE. It helps in distributing message processing across multiple consumers, with each consumer receiving a subset of the messages.

Example:

redis$> XREADGROUP GROUP mygroup consumer1 BLOCK 0 STREAMS mystream >

This command reads messages for the consumer group mygroup and assigns them to consumer1. The > symbol indicates that the consumer should read only messages that have not been processed yet.

4. XACK

Acknowledges the processing of a message. This command is used within consumer groups to confirm that a message has been successfully processed. Once acknowledged, the message is no longer pending.

Example:

redis$> XACK mystream mygroup 1526569487630-0

This acknowledges the message with ID 1526569487630-0 in the stream mystream for the mygroup consumer group.

5. XDEL

Deletes one or more messages from a stream. While this command can be used to manually remove messages, it's often managed automatically through stream trimming.

Example:

redis$> XDEL mystream 1526569487630-0

This deletes the message with ID 1526569487630-0 from the stream mystream.

6. XTRIM

Trims the stream by removing old messages based on specified criteria such as length or age. This helps manage memory usage and ensures the stream only holds relevant messages.

Example:

redis$> XTRIM mystream MAXLEN 1000

This command trims the mystream to keep only the most recent 1000 messages. Note that nearly exact trimming (by using a ~ operator before the threshold value) can also be used for better performance.

Redis Streams vs. Apache Kafka

Apache Kafka is a popular stream processing engine. Unlike Kafka, which was exclusively designed for streaming data, Redis caters to many different use cases, with one of them being stream processing via its Streams feature. 

Let’s explore the differences in more detail.

1. Architecture and Design

Redis Streams is part of the Redis ecosystem, primarily designed to be an in-memory data store. It is suitable for use cases that require low-latency access, such as real-time data processing and event-driven architectures. Redis Streams uses a simple, append-only log model, where each message is appended to a stream and can be consumed in the order they were added.

Apache Kafka, on the other hand, is a distributed event streaming platform built for high-throughput, fault-tolerant, and horizontally scalable message handling. Kafka is designed to handle massive volumes of data across multiple consumers and can scale to support large, distributed environments. It operates on a publish-subscribe model with a focus on durability and retention.

2. Message Storage

Redis Streams stores messages in-memory by default, making it suitable for scenarios where low-latency and fast access are critical. It does support persistence, but this is typically used in scenarios where the data volume can fit in memory, or when durability requirements are more relaxed.

Apache Kafka stores messages on disk, ensuring high durability and long-term storage of messages. Kafka is designed for long-term data retention and can store terabytes of data. It excels in scenarios where message durability and replayability are crucial.

3. Durability and Reliability

Redis Streams offers configurable persistence (via RDB or AOF), but it is generally used in scenarios where the speed of access is prioritized over long-term durability. Redis uses “write-after” persistence, meaning changes are first applied in memory and only later saved to disk, either via periodic snapshots (RDB) or by appending each command to a log file (AOF). This approach prioritizes speed and responsiveness, but it comes with a trade-off: in the event of a crash, some recent changes may be lost if they haven't yet been persisted.

Apache Kafka was built with durability in mind, ensuring that messages are stored on disk and can be replicated across multiple nodes. Kafka guarantees message durability through its log-based storage and replication mechanisms. In particular, Apache Kafka ensures high durability through its in-sync replicas (ISRs), which are broker replicas that have fully caught up with the leader's log. When a producer sends a message, it is written to the leader and then replicated to all ISRs. A message is considered "committed" only after being acknowledged by all in-sync replicas, ensuring that it won’t be lost even if the leader fails.

4. Message Processing and Consumer Groups:

Redis Streams supports consumer groups, allowing multiple consumers to share the load of processing messages. Redis ensures that each consumer in the group processes a subset of messages, providing load balancing and horizontal scaling within the Redis environment.

Kafka also supports consumer groups, but it is built with a more complex mechanism for handling message offsets and allowing consumers to process messages at their own pace. Kafka’s offset management is a core feature, giving consumers fine-grained control over message processing and ensuring that messages are not lost or processed multiple times.

5. Performance and Scalability

Redis Streams is optimized for low-latency operations and is well-suited for real-time processing. Its performance is highly dependent on the system’s available memory and the size of the stream. It is ideal for scenarios that require quick message reads and writes, but it may struggle with extremely large-scale workloads that exceed memory capacity.

Kafka is designed for massive scalability and can handle millions of messages per second. It’s well-suited for high-throughput applications, such as event logging and data pipelines. Kafka's distributed architecture allows it to scale horizontally across clusters, making it more appropriate for handling large-scale data flows across multiple systems.

6. Use Cases

Redis Streams is best suited for real-time data streaming applications, where speed and responsiveness are paramount, such as event-driven systems, real-time analytics, and small-to-medium messaging queues.

Kafka is more appropriate for large-scale event streaming systems, data pipelines, and analytics platforms that require high throughput, long-term storage, and the ability to handle massive amounts of data. It is commonly used in applications like log aggregation, real-time analytics, and stream processing in large distributed systems.

Sample Application: Using Redis Streams to Stream LLM Output

In this tutorial, we will explore how to stream output from a Large Language Model (LLM) to a browser using Redis Streams. This method allows for real-time streaming of data chunks, ensuring a smooth user experience as responses are being generated dynamically. Instructions below are adapted from the Redis documentation.

Prerequisites

To follow this tutorial, you should have:

• Node.js installed.
• A Redis instance up and running.
• Basic knowledge of JavaScript, Node.js, and Redis Streams.

Architecture Overview

The application architecture follows these key steps:

1. User Input: The browser sends a user query to the server via WebSockets.
2. LLM Processing: The server forwards the query to OpenAI's LLM for processing, and the LLM returns a response in chunks.
3. Redis Stream: These chunks are added to a Redis Stream, which acts as a message queue.
4. Consumer: A consumer listens to the Redis Stream and broadcasts the chunks of data to connected clients via WebSockets.

Streaming LLM Output Using Redis Streams Architecture Flow (Source)

Setting Up the Demo

To get started, clone the demo repository from GitHub:

$> git clone https://github.com/redis-developer/redis-short-demos.git
$> cd redis-short-demos/streaming-llm-output

Install dependencies:

$> npm install

Create a .env file and add the following environment variables:

OPENAI_API_KEY=your-openai-api-key
OPENAI_ORGANIZATION=your-openai-organization
REDIS_URL="redis://localhost:6379/"

Start the application:

# Start the backend server.
$> npm run backend

# Start the frontend application.
$> npm run frontend

Visit http://127.0.0.1:5400/ in your browser to interact with the demo.

How It Works

Let’s break down the core components of the code.

Redis Utilities:

• The redis-wrapper.ts module provides utility functions to interact with Redis Streams. Functions like addItemToStream and readStream manage adding data to the stream and reading it efficiently.
• addItemToStream: Adds messages to a Redis stream with automatically generated IDs.
• readStream: Listens to a Redis stream and processes data as it arrives.

LLM Prompting:

• The question.ts module handles the creation of prompts for the LLM using the LangChain library. It defines how to ask a question and stream the LLM's response in chunks.
• askQuestion: This function sends the query to the LLM and streams the response to the Redis stream in real-time.

Socket Server:

• The socket-x-read.ts module sets up WebSockets with Socket.IO to manage real-time communication between the client and server. It listens for incoming user questions, processes them, and streams the LLM output to the browser.

Express Server:

• The index.ts module integrates the backend server with the Redis connection and WebSocket server, ensuring real-time communication with clients.

Frontend Implementation

The frontend, built with plain JavaScript, connects to the server using WebSockets and listens for the streamed chunks of data. Upon receiving the chunks, it dynamically updates the page, allowing the user to view the response as it is generated. Example frontend logic:

const socket = io('http://localhost:3000');

// Function to send the question to the server.
function onSearch() {
  const outputDiv = document.getElementById('output');
  const question = document.getElementById('question').value;

  // Clear previous output.
  outputDiv.innerHTML = '';

  // Use socket to emit the question.
  socket.emit('askQuestion', {
    topic: 'Redis',
    topicQuestion: question,
  });
}

// Listen for streamed chunks of LLM's response.
socket.on('chunk', (chunk) => {
  const outputDiv = document.getElementById('output');
  outputDiv.innerHTML += chunk;
});

Redis Insight Monitoring

You can monitor the Redis stream using Redis Insight, a GUI tool for interacting with Redis data. It allows you to visualize the Redis stream OPENAI_STREAM and inspect the messages being processed in real-time.

In this tutorial, we demonstrated how to stream LLM outputs in real-time using Redis Streams. This setup ensures a responsive and interactive user experience as large language models process and generate responses in chunks. Redis Streams provide a powerful, scalable, and fault-tolerant solution for handling real-time data in distributed systems.

Best Practices for Redis Streams

1. Use Consumer Groups for Parallel Processing

Consumer groups in Redis Streams enable efficient parallel processing of messages by distributing the workload across multiple consumers. This is particularly useful in scenarios where the volume of messages is high, and single-threaded processing would become a bottleneck. By using consumer groups, Redis ensures that each message is consumed by one and only one consumer, avoiding duplicate processing and providing fault tolerance. It is essential to design your consumer groups to match the expected workload, ensuring a balanced load among the consumers to avoid resource contention or idle consumers. 

Consider configuring consumer groups based on message types or priorities, and adjust the number of consumers as your system scales to handle traffic spikes or shifting loads. Additionally, regular monitoring of consumer performance and stream consumption can help prevent any single consumer from falling behind in processing.

2. Acknowledge Messages Promptly

Acknowledging messages in Redis Streams is crucial for maintaining the integrity of the system and ensuring that messages are not reprocessed unnecessarily. When a consumer successfully processes a message, it should immediately send an acknowledgment (XACK) to Redis. If the acknowledgment is delayed or forgotten, the message will remain in the Pending Entries List (PEL), and the consumer might be reassigned the same message, leading to inefficiencies and potential data inconsistencies. 

In systems where message processing can fail or be delayed, it's also a good idea to implement automatic retries with appropriate backoff mechanisms. However, avoid too many retries, as this can increase load unnecessarily and slow down the overall system. Ensuring prompt acknowledgment improves throughput, reduces memory usage, and prevents potential message duplication.

3. Trim Streams to Manage Memory Usage

Redis Streams can consume a significant amount of memory, especially when the stream grows rapidly due to high message throughput. To avoid running out of memory or compromising system performance, it is important to periodically trim old messages. Redis Streams offers multiple trimming options, such as trimming by maximum length (MAXLEN) or by a time-based threshold (MINID). Implementing stream trimming ensures that your Redis instance only retains the most recent or relevant data, which is crucial when dealing with large-scale systems. 

You can set trimming policies to automatically discard outdated messages without human intervention, making your system more resilient to large message volumes. Be careful when configuring trimming policies, as trimming messages too aggressively could result in the loss of important historical data that might be needed later. Always evaluate your use case carefully and adjust trimming strategies based on the specific retention needs of your application.

4. Implement Idempotent Consumers

Idempotency is a key design pattern in distributed systems, especially when handling message queues. In the context of Redis Streams, it ensures that even if a consumer processes a message more than once (due to a failure, re-delivery, or retry), the outcome remains the same. For instance, consider a message indicating that a transaction was completed. If the consumer fails after processing the transaction and then reprocesses the same message, idempotency ensures the transaction isn't duplicated. 

Implementing idempotency requires careful design, such as ensuring that consumers track their progress and only process messages once, or that the actions triggered by a message are safe to repeat without side effects. You can achieve idempotency through mechanisms such as deduplication of message IDs or checking for the completion status of a task before re-executing it. Idempotent consumers enhance reliability and system consistency, especially in the case of retries, message reordering, or failures.

5. Monitor Pending Entries List (PEL)

The Pending Entries List (PEL) is an important Redis Streams feature for monitoring messages that have been delivered to consumers but not yet acknowledged. The PEL helps track the processing state of messages in real-time, and it’s a valuable tool for identifying potential issues like delayed processing, failed consumers, or system bottlenecks. If the PEL grows too large, it could signal that consumers are struggling to keep up with the stream or are failing to process messages on time. 

Monitoring the PEL involves checking for consumers who are falling behind or have stopped acknowledging messages, which can lead to message duplication or even message loss if the system is not configured to handle retries correctly. Regularly inspecting the PEL, either manually or using automated monitoring tools, is critical for maintaining a healthy system. It’s also advisable to set up alerts based on certain thresholds for the PEL size or the duration that messages remain unacknowledged. By keeping track of this list, you can proactively address performance issues, optimize consumer behavior, and prevent backlogs from affecting overall system performance.

Dragonfly: The Next-Generation In-Memory Data Store

Dragonfly is a modern, source-available, multi-threaded, Redis-compatible in-memory data store that stands out by delivering unmatched performance and efficiency. Designed from the ground up to disrupt existing legacy technologies, Dragonfly redefines what an in-memory data store can achieve. With Dragonfly, you get the familiar API of Redis (including Streams) without the performance bottlenecks, making it an essential tool for modern cloud architectures aiming for peak performance and cost savings. Migrating from Redis to Dragonfly requires zero or minimal code changes.

Key Advancements of Dragonfly

• Multi-Threaded Architecture: Efficiently leverages modern multi-core processors to maximize throughput and minimize latency.
• Unmatched Performance: Achieves 25x better performance than Redis, ensuring your applications run with extremely high throughput and consistent latency.
• Cost Efficiency: Reduces hardware and operational costs without sacrificing performance, making it an ideal choice for budget-conscious enterprises.
• Redis API Compatibility: Offers seamless integration with existing applications and frameworks running on Redis while overcoming its limitations.
• Innovative Design: Built to scale vertically and horizontally, providing a robust solution for rapidly growing data needs.

Dragonfly Cloud is a fully managed service from the creators of Dragonfly, handling all operations and delivering effortless scaling so you can focus on what matters without worrying about in-memory data infrastructure anymore.