Redis Streams occupies a practical middle ground for real-time data pipelines. It delivers a fast, persistent event log with built-in consumer groups, acknowledgment tracking, and horizontal scaling. No separate broker infrastructure. No complex operational overhead. Just Redis doing what it does best: staying simple and fast.
If you’re building AI data ingestion systems, real-time analytics pipelines, or event-driven workflows that need to handle concurrent consumers without message loss, this guide walks through the patterns that actually work in production.
Why Redis Streams for Real-Time Pipelines
Redis Streams gives you a few critical capabilities that make it suitable for production data pipelines:
- Persistent event log with automatic ID generation and ordering
- Consumer groups that track which messages each consumer has processed
- Acknowledgment (ACK) mechanism to prevent message loss
- Pending entries list (PEL) to handle failed processing
- Simple interface without the operational overhead of a separate message broker
For AI data ingestion specifically, this means you can ingest raw data events into Redis Streams, then fan them out to multiple processors (embedding generation, data validation, feature extraction) without worrying about duplicate processing or lost messages.
Core Concepts: Streams, Consumer Groups, and Acknowledgments
Before we dive into code, let’s establish the mental model.
A Redis Stream is an append-only log. Every event gets a unique ID (usually a timestamp plus a sequence number). Multiple consumers can read from the same stream independently, and consumer groups let you divide work across parallel processors.
When a consumer reads a message from a group, it doesn’t immediately disappear. The message stays in the stream until the consumer explicitly acknowledges it. If processing fails, the message stays in the pending entries list and can be retried or sent to a dead-letter stream.
This is the foundation of reliable delivery: a message is processed at least once (via retries on failure), and you control when it’s considered truly delivered.
Setting Up a Basic Stream and Consumer Group
Let’s start with a minimal example. We’ll create a stream, add an event, and set up a consumer group.
const redis = require('redis');
const client = redis.createClient({
host: 'localhost',
port: 6379
});
await client.connect();
// Add an event to the stream
const eventId = await client.xAdd('ai-data-ingestion', '*', {
userId: 'user123',
eventType: 'page_view',
timestamp: Date.now().toString(),
metadata: JSON.stringify({ page: '/dashboard', referrer: 'google' })
});
console.log('Event added with ID:', eventId);
// Create a consumer group
// The second argument is the group name, third is the starting message ID
// '$' means start from new messages only
await client.xGroupCreate('ai-data-ingestion', 'embedding-processors', '$', {
MKSTREAM: true // Create stream if it doesn't exist
});
console.log('Consumer group created');
Now we have a stream called ‘ai-data-ingestion’ and a consumer group called ’embedding-processors’. Any consumer that joins this group will receive unprocessed messages in parallel.
Building a Resilient Consumer
Here’s where the real value emerges. A consumer reads from the group, processes the message, and acknowledges only after successful processing.
async function startEmbeddingConsumer(consumerId) {
const BATCH_SIZE = 10;
const BLOCK_TIME = 1000; // milliseconds
while (true) {
try {
// Read up to BATCH_SIZE messages, block if none available
const messages = await client.xReadGroup(
{
key: 'ai-data-ingestion',
group: 'embedding-processors',
consumer: consumerId
},
{
count: BATCH_SIZE,
block: BLOCK_TIME
}
);
if (!messages || messages.length === 0) {
console.log(`[${consumerId}] No messages, waiting...`);
continue;
}
for (const message of messages[0].messages) {
const { id, message: data } = message;
try {
// Process the event (e.g., generate embeddings)
console.log(`[${consumerId}] Processing event ${id}:`, data);
const processed = await processEvent(data);
// Store result (e.g., in a results stream or database)
await client.xAdd('embedding-results', '*', {
sourceId: id,
result: JSON.stringify(processed),
processedAt: Date.now().toString()
});
// Acknowledge only after successful processing
await client.xAck(
'ai-data-ingestion',
'embedding-processors',
id
);
console.log(`[${consumerId}] Acknowledged ${id}`);
} catch (error) {
console.error(`[${consumerId}] Failed to process ${id}:`, error.message);
// Message stays in pending list for retry
}
}
} catch (error) {
console.error(`[${consumerId}] Consumer error:`, error.message);
await new Promise(resolve => setTimeout(resolve, 5000));
}
}
}
// Helper function: simulate event processing
async function processEvent(eventData) {
// In production, this might call an embedding API, ML model, etc.
return {
embeddings: [0.1, 0.2, 0.3],
confidence: 0.95
};
}
// Start multiple consumers
await startEmbeddingConsumer('processor-1');
await startEmbeddingConsumer('processor-2');
await startEmbeddingConsumer('processor-3');
The pattern is straightforward: read, process, acknowledge. If processing fails, the message stays pending and can be retried later. This ensures no data loss.
Handling Failed Messages and Dead-Letter Streams
In production, some messages will fail repeatedly. After a threshold, you need to move them somewhere for manual inspection or alternative handling. This is a dead-letter stream.
async function processEventWithRetry(consumerId, messageId, eventData, maxRetries = 3) {
// Get the current retry count from the pending entries list
const pendingInfo = await client.xPending(
'ai-data-ingestion',
'embedding-processors',
messageId,
messageId,
1
);
const retryCount = pendingInfo[0]?.deliveryCount || 1;
try {
const result = await processEvent(eventData);
await client.xAdd('embedding-results', '*', {
sourceId: messageId,
result: JSON.stringify(result),
processedAt: Date.now().toString()
});
await client.xAck(
'ai-data-ingestion',
'embedding-processors',
messageId
);
console.log(`[${consumerId}] Successfully processed ${messageId}`);
} catch (error) {
console.error(`[${consumerId}] Attempt ${retryCount} failed for ${messageId}:`, error.message);
if (retryCount >= maxRetries) {
// Move to dead-letter stream
await client.xAdd('embedding-dead-letter', '*', {
originalId: messageId,
originalData: JSON.stringify(eventData),
error: error.message,
retries: retryCount.toString(),
failedAt: Date.now().toString()
});
// Acknowledge to remove from pending list
await client.xAck(
'ai-data-ingestion',
'embedding-processors',
messageId
);
console.log(`[${consumerId}] Moved ${messageId} to dead-letter after ${retryCount} retries`);
}
// If retries remaining, don't acknowledge; message stays pending for next attempt
}
}
The dead-letter stream gives you a place to inspect failed events, understand failure patterns, and potentially reprocess them later with fixes.
At-Least-Once Delivery and Idempotency
Redis Streams provides at-least-once delivery through consumer groups and acknowledgments. To handle scenarios where a message might be delivered more than once, you need idempotency in your processing logic.
The simplest approach: use the message ID as an idempotency key. Before processing, check if you’ve already processed this ID. If yes, skip; if no, process and record the ID.
async function processEventIdempotent(messageId, eventData) {
// Check if already processed
const alreadyProcessed = await client.get(`processed:${messageId}`);
if (alreadyProcessed) {
console.log(`Message ${messageId} already processed, skipping`);
return JSON.parse(alreadyProcessed);
}
// Process the event
const result = await processEvent(eventData);
// Store result and mark as processed
await client.set(
`processed:${messageId}`,
JSON.stringify(result),
{ EX: 86400 } // Expire after 24 hours
);
return result;
}
This pattern ensures that even if a message is delivered twice (due to retries or network issues), your downstream systems see the same result.
Monitoring and Observability
In production, you need visibility into your pipelines. Check consumer group status regularly.
async function monitorConsumerGroup() {
const groupInfo = await client.xGroupInfo('ai-data-ingestion');
for (const group of groupInfo) {
console.log(`Group: ${group.name}`);
console.log(` Consumers: ${group.consumers}`);
console.log(` Pending: ${group.pending}`);
console.log(` Last delivered ID: ${group.lastDeliveredId}`);
}
// Get detailed consumer info
const consumerInfo = await client.xGroupConsumers(
'ai-data-ingestion',
'embedding-processors'
);
for (const consumer of consumerInfo) {
console.log(`Consumer: ${consumer.name}`);
console.log(` Pending messages: ${consumer.pending}`);
console.log(` Idle time: ${consumer.idle}ms`);
}
// Check for stuck messages
const pending = await client.xPending(
'ai-data-ingestion',
'embedding-processors',
'-',
'+',
10 // Get top 10 oldest pending
);
for (const msg of pending) {
const idleTime = msg.millisecondsSinceLastDelivery;
if (idleTime > 300000) { // 5 minutes
console.warn(`Message ${msg.id} stuck for ${idleTime}ms, consider redelivery`);
}
}
}
// Run monitoring every 30 seconds
setInterval(monitorConsumerGroup, 30000);
This monitoring loop catches stuck messages, tracks consumer health, and alerts you to pending backlogs before they become problems.
Scaling Horizontally
The beauty of consumer groups is that scaling is straightforward. Each consumer process reads from the same group, and Redis automatically distributes messages across them.
// In your deployment (e.g., Docker or Kubernetes)
// Start multiple instances of the same consumer:
const consumerId = process.env.CONSUMER_ID || `processor-${process.pid}`;
const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';
const client = redis.createClient({
url: redisUrl
});
await client.connect();
// Ensure consumer group exists
await client.xGroupCreate('ai-data-ingestion', 'embedding-processors', '$', {
MKSTREAM: true
}).catch(() => {}); // Ignore if already exists
await startEmbeddingConsumer(consumerId);
Deploy this code in a Docker container, scale to 10 replicas via Kubernetes, and each replica automatically becomes a consumer in the group. Messages are distributed fairly across all active consumers.
Production Deployment Patterns
A few things to consider when moving to production:
Redis Persistence: Use AOF (Append-Only File) mode or RDB snapshots to ensure stream data survives restarts. For critical pipelines, use Redis Sentinel or Cluster for high availability.
Connection Pooling: Use a connection pool (e.g., with ioredis or redis-pool) so you don’t exhaust connections. Each consumer process should share a pool.
Graceful Shutdown: When stopping a consumer, finish processing current messages before exiting. This prevents unnecessary redeliveries.
let isShuttingDown = false;
process.on('SIGTERM', async () => {
console.log('SIGTERM received, shutting down gracefully...');
isShuttingDown = true;
// Allow current batch to finish (adjust timeout as needed)
await new Promise(resolve => setTimeout(resolve, 30000));
await client.disconnect();
process.exit(0);
});
async function startEmbeddingConsumer(consumerId) {
while (!isShuttingDown) {
// ... consumer logic ...
}
}
Environment-Specific Configuration: Use environment variables for stream names, group names, batch sizes, and retry thresholds. This lets you test locally and deploy to production without code changes.
Practical AI Data Ingestion Example
Let’s tie it together with a realistic scenario: ingesting user events, enriching them with embeddings, and storing results for downstream ML models.
// Producer: ingest raw events
async function ingestEvent(event) {
return await client.xAdd('raw-events', '*', {
userId: event.userId,
eventType: event.type,
data: JSON.stringify(event.data),
timestamp: Date.now().toString()
});
}
// Consumer 1: enrich events with embeddings
async function enrichmentConsumer() {
const messages = await client.xReadGroup(
{ key: 'raw-events', group: 'enrichment', consumer: 'enricher-1' },
{ count: 10, block: 1000 }
);
for (const msg of messages[0].messages) {
const eventData = msg.message;
const embedding = await generateEmbedding(eventData.data);
await client.xAdd('enriched-events', '*', {
sourceId: msg.id,
userId: eventData.userId,
embedding: JSON.stringify(embedding),
enrichedAt: Date.now().toString()
});
await client.xAck('raw-events', 'enrichment', msg.id);
}
}
// Consumer 2: store enriched events for ML training
async function storageConsumer() {
const messages = await client.xReadGroup(
{ key: 'enriched-events', group: 'storage', consumer: 'storage-1' },
{ count: 10, block: 1000 }
);
for (const msg of messages[0].messages) {
const enrichedData = msg.message;
// Store in database (e.g., PostgreSQL, MongoDB)
await storeInDatabase(enrichedData);
await client.xAck('enriched-events', 'storage', msg.id);
}
}
async function generateEmbedding(data) {
// Call your embedding API (OpenAI, local model, etc.)
return [0.1, 0.2, 0.3]; // Simplified
}
async function storeInDatabase(data) {
// Your database write logic
}
// Start the pipeline
await ingestEvent({ userId: 'user123', type: 'action', data: 'viewed product' });
await enrichmentConsumer();
await storageConsumer();
This pattern creates a multi-stage pipeline where each stage is independent, scalable, and resilient. If the embedding service is slow, enrichment consumers back up but don’t block ingestion. If storage fails, enriched events stay pending until the database recovers.
Common Pitfalls and How to Avoid Them
Not acknowledging messages: If you never call xAck, messages accumulate in the pending list forever. Always acknowledge after successful processing.
Blocking forever: Set a reasonable block timeout. If you block indefinitely and the process crashes, the consumer is marked as dead and messages are redelivered after a timeout.
Ignoring idle consumers: Monitor for consumers that haven’t checked in. Stale consumers can hold messages indefinitely. Use xGroupDelConsumer to clean them up.
Undersizing Redis: Redis Streams live in memory. Monitor memory usage and plan capacity based on stream size, retention, and throughput. Use MAXLEN to trim old messages if storage is a concern.
// Trim stream to last 100,000 messages
await client.xTrimMaxLen('ai-data-ingestion', 100000);
Wrapping Up
Redis Streams is a powerful tool for building real-time data pipelines. Consumer groups, acknowledgments, and the pending entries list give you the primitives for resilient, at-least-once processing at scale.
For AI data ingestion pipelines, event-driven architectures, and real-time analytics on Node.js, it’s worth serious consideration. Start small, monitor closely, and scale horizontally as your pipeline grows.
What is the difference between Redis Streams and Redis Pub/Sub?
Redis Pub/Sub delivers messages only to subscribers that are actively listening at the moment of publication. If no one is listening, the message is lost. Redis Streams, on the other hand, persist messages to a log, so new subscribers can read historical messages. Streams also support consumer groups for parallel processing and acknowledgment tracking, making them suitable for reliable data pipelines.
How do I achieve exactly-once delivery with Redis Streams?
Redis Streams provides at-least-once delivery via consumer groups and acknowledgments. To achieve exactly-once semantics, you need to implement idempotency in your application logic. Use the message ID as an idempotency key, check if a message has been processed before, and skip or return cached results if it has.
How do I scale Redis Streams to handle high throughput?
Scale horizontally by running multiple consumer processes, all joining the same consumer group. Redis automatically distributes messages across active consumers. Each consumer should use a connection pool and process messages in batches. Monitor consumer lag and pending messages to ensure even distribution. For very high throughput, consider Redis Cluster to shard data across multiple Redis instances.
What happens if a consumer crashes while processing a message?
If a consumer crashes before acknowledging a message, the message remains in the pending entries list. After a timeout (configurable), Redis marks the consumer as dead and redelivers the message to another consumer in the group. This ensures no messages are lost, though you may process the same message multiple times if the original consumer partially succeeded.
How do I monitor Redis Streams in production?
Use xGroupInfo to check consumer group status, including the number of pending messages and consumers. Use xGroupConsumers to see individual consumer health and idle time. Use xPending to identify stuck messages. Set up alerts for growing pending backlogs, idle consumers, and stuck messages that haven’t been redelivered. Export metrics to your monitoring system (Prometheus, DataDog, etc.) for visibility.
