Redis Pub/Sub & Streams Core Knowledge

Full Reference: See advanced.md for Streams consumer groups (Node.js, Java, Python), pending message recovery, TLS configuration, and Sentinel setup.

Deep Knowledge: Use mcp__documentation__fetch_docs with technology: redis-pubsub for comprehensive documentation.

Quick Start (Docker)

services: redis: image: redis:7-alpine ports: - "6379:6379" command: redis-server --appendonly yes

Core Concepts

Feature Pub/Sub Streams

Persistence No Yes

Consumer Groups No Yes

Message History No Yes

Delivery Fire-and-forget At-least-once

Use Case Real-time broadcast Event sourcing, queues

Pub/Sub (Node.js)

import Redis from 'ioredis';

const publisher = new Redis({ host: 'localhost', port: 6379 }); const subscriber = new Redis({ host: 'localhost', port: 6379 });

// Subscriber subscriber.subscribe('orders', 'notifications', (err, count) => { console.log(Subscribed to ${count} channels); });

subscriber.on('message', (channel, message) => { const data = JSON.parse(message); console.log(Received on ${channel}:, data); });

// Pattern subscription subscriber.psubscribe('order.*'); subscriber.on('pmessage', (pattern, channel, message) => { console.log(Pattern ${pattern}, Channel ${channel}:, message); });

// Publisher await publisher.publish('orders', JSON.stringify({ orderId: '123', status: 'created', }));

Pub/Sub (Java - Spring)

@Configuration public class RedisPubSubConfig { @Bean public RedisMessageListenerContainer container( RedisConnectionFactory connectionFactory) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(connectionFactory); container.addMessageListener(orderListener(), new ChannelTopic("orders")); container.addMessageListener(orderPatternListener(), new PatternTopic("order.*")); return container; }

@Bean
public MessageListener orderListener() {
    return (message, pattern) -> {
        String body = new String(message.getBody());
        Order order = objectMapper.readValue(body, Order.class);
        processOrder(order);
    };
}

}

@Service public class OrderPublisher { @Autowired private StringRedisTemplate redisTemplate;

public void publishOrder(Order order) {
    redisTemplate.convertAndSend("orders",
        objectMapper.writeValueAsString(order));
}

}

Pub/Sub (Python)

import redis import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

Subscriber (run in thread)

def subscriber(): pubsub = r.pubsub() pubsub.subscribe('orders') pubsub.psubscribe('order.*')

for message in pubsub.listen():
    if message['type'] in ('message', 'pmessage'):
        data = json.loads(message['data'])
        print(f"Received: {data}")

Publisher

r.publish('orders', json.dumps({'orderId': '123', 'status': 'created'}))

Stream Commands Reference

Command Description

XADD

Add entry to stream

XREAD

Read entries

XREADGROUP

Read with consumer group

XACK

Acknowledge processing

XPENDING

List pending entries

XCLAIM

Claim pending entry

XTRIM

Trim stream size

When NOT to Use This Skill

• Guaranteed message delivery - Pub/Sub is fire-and-forget

• Complex routing patterns - RabbitMQ is better

• High-throughput event sourcing - Kafka provides better throughput

• Multi-datacenter replication - Kafka or Pulsar handle this better

Anti-Patterns

Anti-Pattern Why It's Bad Solution

Pub/Sub for critical data No persistence Use Streams

No XTRIM on streams Memory growth Set MAXLEN or auto-trim

Single consumer group Can't scale Use consumer groups

No XPENDING monitoring Lost messages Monitor and claim pending

No ACK after processing Premature consumption ACK only after success

Quick Troubleshooting

Issue Likely Cause Fix

Messages not received No active subscribers Pub/Sub requires active subscribers

Stream growing unbounded No XTRIM Add MAXLEN limit

High memory usage Large stream Trim streams

Pending messages growing Consumer crashes Implement XCLAIM recovery

Duplicate processing Consumer group issue Ensure unique consumer IDs

Production Checklist

• Authentication enabled

• TLS configured

• Sentinel/Cluster for HA

• Stream trimming configured

• Consumer group monitoring

• Pending message handling

• Memory limits set

• Persistence configured (AOF)

Monitoring Metrics

Metric Alert Threshold

Stream length

1000000

Pending entries

10000

Consumer lag

1000

Memory usage

80%

Reference Documentation

Available topics: basics , pubsub , streams , patterns