What are Redis Streams and how are they used for event sourcing?

Redis Streams provide log-like data structure for event sourcing: **Basic Stream Operations:** # Add entries to stream XADD mystream * sensor-id 1234 temperature 19.8 humidity 67.2 XADD mystream * sensor-id 1235 temperature 20.1 humidity 65.8 # Read from stream XREAD COUNT 2 STREAMS mystream 0 XREAD BLOCK 5000 STREAMS mystream $ # Block for new entries # Read range XRANGE mystream - + # All entries **Consumer Groups:** # Create consumer group XGROUP CREATE mystream mygroup 0 MKSTREAM # Add consumers to group XREADGROUP GROUP mygroup consumer1 COUNT 1 STREAMS mystream > XREADGROUP GROUP mygroup consumer2 COUNT 1 STREAMS mystream > # Acknowledge processed messages XACK mystream mygroup 1609459200000-0 **Python Example - Event Processing:** import redis import json from datetime import datetime r = redis.Redis(decode_responses=True) # Producer - Add events def add_event(event_type, data): event = { 'type': event_type, 'timestamp': datetime.now().isoformat(), 'data': json.dumps(data) } return r.xadd('events', event) # Consumer - Process events def process_events(group_name, consumer_name): try: r.xgroup_create('events', group_name, id='0', mkstream=True) except redis.ResponseError: pass # Group already exists while True: messages = r.xreadgroup( group_name, consumer_name, {'events': '>'}, count=1, block=1000 ) for stream, msgs in messages: for msg_id, fields in msgs: # Process message event_type = fields['type'] data = json.loads(fields['data']) print(f"Processing {event_type}: {data}") # Acknowledge message r.xack('events', group_name, msg_id) **Use Cases:** • Event sourcing • Activity feeds • Sensor data collection • Chat applications • Audit logs • Real-time analytics **Advantages over Pub/Sub:** • Message persistence • Consumer groups for load balancing • Message acknowledgment • Replay capability