Dwa paradygmaty przesyłania wiadomości w Redis

Redis oferuje dwa radykalnie różne mechanizmy przesyłania wiadomości. Pub/sub jest to system transmisji w czasie rzeczywistym: wiadomości są dostarczane do wszystkich podłączonych abonentów w tym momencie a potem zapomniane. Jeśli abonent tego nie robi zostanie podłączony, wiadomość zostanie utracona na zawsze.

Strumienie Redisa (wprowadzony w Redis 5.0) to dziennik przeznaczony tylko do dodawania trwałe inspirowane koncepcją strumienia Apache Kafka. Wiadomości są zachowywane o ile nie zostaną wyraźnie usunięte, można je przeczytać ponownie razy i wspieraj Grupy Konsumentów w przetwarzaniu rozproszonym za pomocą potwierdzeń.

Czego się nauczysz

  • Pub/Sub: SUBSKRYBUJ, PUBLIKUJ, PSUBSKRYBUJ ze wzorem globowania
  • Kiedy używać Pub/Sub a kiedy strumieni: kluczowe różnice
  • Strumienie: XADD, XREAD, XRANGE dla dzienników tylko do dołączania
  • Grupy konsumentów: XGROUP CREATE, XREADGROUP, XACK
  • Co najmniej jednokrotne dostarczanie i zarządzanie oczekującymi wiadomościami
  • Wzorzec: pozyskiwanie zdarzeń i dziennik inspekcji ze strumieniami

Pub/Sub: nadawanie w czasie rzeczywistym

Pub/Sub jest idealny do powiadomień w czasie rzeczywistym w przypadku utraty części akceptowana jest wiadomość: aktualizacje cen na żywo, powiadomienia push, pokój rozmów, unieważnianie rozproszonej pamięci podręcznej. 

# Terminal 1: Subscriber
SUBSCRIBE notifications:user:1001
# Attende messaggi sul canale...
# Quando arriva un messaggio:
# 1) "message"
# 2) "notifications:user:1001"
# 3) "{"type":"payment","amount":100}"

# Puoi iscriverti a piu canali
SUBSCRIBE channel:1 channel:2 channel:3

# PSUBSCRIBE: subscribe con pattern glob
PSUBSCRIBE notifications:user:*    # Tutti gli utenti
PSUBSCRIBE events:order:*          # Tutti gli ordini

# Terminal 2: Publisher
PUBLISH notifications:user:1001 '{"type":"payment","amount":100}'
# (integer) 1  -- numero di subscriber che hanno ricevuto il messaggio

PUBLISH notifications:user:1001 '{"type":"login","ip":"192.168.1.1"}'
# (integer) 1

# Se nessun subscriber e connesso:
PUBLISH notifications:user:9999 '{"type":"alert"}'
# (integer) 0  -- messaggio PERSO, nessuno lo ricevera mai
# Python: Pub/Sub con redis-py
import redis
import json
import threading

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

# --- SUBSCRIBER ---
def message_handler(message):
    if message['type'] == 'message':
        channel = message['channel']
        data = json.loads(message['data'])
        print(f"[{channel}] Received: {data}")

def start_subscriber(user_id: int):
    """Subscriber in background thread."""
    sub = r.pubsub()
    sub.subscribe(**{
        f'notifications:user:{user_id}': message_handler,
        'notifications:broadcast': message_handler,
    })
    # Pattern subscribe
    sub.psubscribe(**{'events:*': message_handler})

    # Blocca in ascolto (in un thread separato)
    thread = sub.run_in_thread(sleep_time=0.01, daemon=True)
    return thread, sub

# --- PUBLISHER ---
def publish_notification(user_id: int, notification_type: str, data: dict):
    """Pubblica una notifica a un utente specifico."""
    payload = json.dumps({'type': notification_type, **data})
    subscribers_count = r.publish(f'notifications:user:{user_id}', payload)
    print(f"Delivered to {subscribers_count} subscribers")
    return subscribers_count

# Avvia subscriber
thread, pubsub = start_subscriber(user_id=1001)

# Pubblica messaggi
import time
time.sleep(0.1)  # Aspetta che subscriber sia pronto

publish_notification(1001, 'payment', {'amount': 100, 'currency': 'EUR'})
publish_notification(1001, 'login', {'ip': '192.168.1.1', 'city': 'Milano'})

# IMPORTANTE: Pub/Sub non e' persistente
# Se il subscriber si disconnette, perde tutti i messaggi inviati nel frattempo
time.sleep(0.5)
pubsub.unsubscribe()
thread.stop()

Strumienie Redis: dziennik trwały i umożliwiający tylko dołączanie

Streamy korzystają z zupełnie innego modelu: każda wiadomość ma unikalny identyfikator (sekwencja znacznika czasu) jest zachowywany w dzienniku i można go ponownie odczytać w dowolnym momencie. Jest to właściwy wybór w przypadku pozyskiwania zdarzeń, dzienników audytu, kolejki zadań i każdy scenariusz, w którym utrata wiadomości jest niedopuszczalna. 

# Redis Streams: XADD e XREAD

# XADD: aggiunge un messaggio allo stream
# ID auto-generato: {milliseconds}-{sequence}
XADD orders:stream * user_id 1001 product_id 555 quantity 2 total 89.99
# "1710000001234-0"  <-- ID generato automaticamente

XADD orders:stream * user_id 1002 product_id 123 quantity 1 total 29.99
# "1710000002345-0"

# ID esplicito (raramente necessario)
XADD orders:stream 1710000003000-0 user_id 1003 product_id 789 quantity 3 total 149.97

# XLEN: numero di messaggi nello stream
XLEN orders:stream    # 3

# XRANGE: leggi messaggi da un range di ID (o -)
XRANGE orders:stream - +
# 1) 1) "1710000001234-0"
#    2) 1) "user_id"
#       2) "1001"
#       3) "product_id"
#       4) "555"
#       5) "quantity"
#       6) "2"
#       7) "total"
#       8) "89.99"
# ...

# XRANGE con COUNT per paginazione
XRANGE orders:stream - + COUNT 10

# XREVRANGE: ordine inverso (dal piu recente)
XREVRANGE orders:stream + - COUNT 5

# XREAD: leggi messaggi dall'ultimo ID letto (blocking possible)
XREAD COUNT 10 STREAMS orders:stream 0   # Da ID 0, leggi al massimo 10

# Leggi solo messaggi nuovi ($ = solo futuri)
XREAD COUNT 10 BLOCK 5000 STREAMS orders:stream $
# Blocca per max 5 secondi aspettando nuovi messaggi

Grupy konsumentów: przetwarzanie rozproszone z potwierdzeniem

Grupy konsumentów to najpotężniejsza funkcja strumieni Redis. Pozwalają wielu odbiorców do przetwarzania komunikatów z tego samego strumienia w sposób rozproszony, z gwarancją, że każdą wiadomość przetwarza dokładnie jeden konsument w grupie, oraz z automatycznym ponownym dostarczaniem niepotwierdzonych wiadomości. 

# Consumer Groups: setup e utilizzo

# 1. Crea il Consumer Group
# $ = inizia dai messaggi futuri, 0 = include messaggi esistenti
XGROUP CREATE orders:stream order-processors $ MKSTREAM
# "OK"

# Crea gruppo che inizia dall'inizio dello stream
XGROUP CREATE orders:stream order-processors-v2 0

# 2. Consumer legge messaggi dal gruppo
# > = leggi il prossimo messaggio non assegnato a nessun consumer
XREADGROUP GROUP order-processors worker-1 COUNT 10 STREAMS orders:stream >
# 1) 1) "orders:stream"
#    2) 1) 1) "1710000001234-0"
#          2) 1) "user_id"
#             2) "1001"
#             ...

# 3. Acknowledge (conferma processing completato)
XACK orders:stream order-processors 1710000001234-0
# (integer) 1  -- 1 messaggio acknowledged

# 4. XPENDING: vedi messaggi pendenti (non ancora acked)
XPENDING orders:stream order-processors - + 10
# Mostra: ID, consumer assegnato, tempo dall'ultimo delivery, delivery count

# 5. XCLAIM: riassegna messaggi idle a un altro consumer (dopo timeout)
XCLAIM orders:stream order-processors worker-2 60000 1710000001234-0
# Riassegna il messaggio se e' idle da piu di 60 secondi

# 6. XDEL: elimina messaggi specifici
XDEL orders:stream 1710000001234-0

# 7. XTRIM: mantieni solo gli ultimi N messaggi
XTRIM orders:stream MAXLEN 10000    # Mantieni ultimi 10.000 messaggi
XTRIM orders:stream MAXLEN ~ 10000  # Approssimativo (piu efficiente)
# Python: Consumer Group con at-least-once delivery
import redis
import json
import time
from typing import Callable

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

STREAM_KEY = 'orders:stream'
GROUP_NAME = 'order-processors'
PENDING_TIMEOUT_MS = 30000  # 30 secondi prima di ri-assegnare

class StreamConsumer:
    def __init__(self, consumer_name: str, handler: Callable):
        self.consumer_name = consumer_name
        self.handler = handler
        self._ensure_group()

    def _ensure_group(self):
        """Crea il consumer group se non esiste."""
        try:
            r.xgroup_create(STREAM_KEY, GROUP_NAME, id=', mkstream=True)
            print(f"Created consumer group: {GROUP_NAME}")
        except redis.exceptions.ResponseError as e:
            if 'BUSYGROUP' not in str(e):
                raise
            # Gruppo gia' esiste: ok

    def process_messages(self, batch_size: int = 10) -> int:
        """Processa un batch di messaggi. Ritorna il numero processato."""
        processed = 0

        # Prima: controlla e riassegna messaggi pending di altri consumer (crashati)
        self._reclaim_pending()

        # Leggi nuovi messaggi non assegnati
        results = r.xreadgroup(
            groupname=GROUP_NAME,
            consumername=self.consumer_name,
            streams={STREAM_KEY: '>'},
            count=batch_size,
            block=1000,  # Blocca 1 secondo se non ci sono messaggi
        )

        if not results:
            return 0

        for stream_name, messages in results:
            for message_id, fields in messages:
                try:
                    self.handler(message_id, fields)
                    # Acknowledge dopo processing riuscito
                    r.xack(STREAM_KEY, GROUP_NAME, message_id)
                    processed += 1
                except Exception as e:
                    print(f"Error processing {message_id}: {e}")
                    # Non acknowledge: il messaggio rimane pending
                    # Verra riassegnato dopo PENDING_TIMEOUT_MS

        return processed

    def _reclaim_pending(self):
        """Riassegna a se' stesso messaggi idle di altri consumer."""
        pending = r.xpending_range(
            STREAM_KEY, GROUP_NAME,
            min='-', max='+', count=10,
        )
        for p in pending:
            if p['time_since_delivered'] > PENDING_TIMEOUT_MS:
                if p['name'] != self.consumer_name:
                    # Riassegna il messaggio a questo consumer
                    r.xclaim(
                        STREAM_KEY, GROUP_NAME, self.consumer_name,
                        PENDING_TIMEOUT_MS, [p['message_id']],
                    )
                    print(f"Reclaimed {p['message_id']} from {p['name']}")

# --- Producer ---
class OrderEventProducer:
    def publish_order(self, order: dict) -> str:
        """Pubblica un evento ordine nello stream."""
        message_id = r.xadd(STREAM_KEY, order)
        return message_id

# --- Utilizzo ---
def handle_order(message_id: str, fields: dict):
    print(f"Processing order {message_id}: {fields}")
    # Logica di business (es. email conferma, aggiornamento inventario)
    time.sleep(0.01)  # Simula lavoro

producer = OrderEventProducer()
consumer = StreamConsumer('worker-1', handle_order)

# Pubblica 5 ordini
for i in range(5):
    msg_id = producer.publish_order({
        'user_id': str(1000 + i),
        'total': str(50.0 + i * 10),
        'status': 'pending',
    })
    print(f"Published: {msg_id}")

# Processa messaggi in loop
while True:
    count = consumer.process_messages(batch_size=5)
    if count == 0:
        break  # Nessun messaggio rimasto

Porównanie: kiedy używać Pub/Sub a kiedy strumieni

Pub/Sub: wybierz kiedy

  • Powiadomienia w czasie rzeczywistym, w których dopuszczalne są niektóre nieodebrane wiadomości
  • Unieważnienie rozproszonej pamięci podręcznej (jeśli węzeł nie otrzyma, następne żądanie unieważnia się)
  • Czat lub obecność na żywo (jeśli jesteś offline, nie obchodzi Cię, co się stało)
  • Finansowe aktualizacje cen/tickerów w czasie rzeczywistym
  • Potrzebujesz bardzo małych opóźnień i możesz tolerować straty

Strumienie: wybierz kiedy

  • Każda wiadomość musi zostać przetworzona przynajmniej raz (zamówienia, płatności)
  • Potrzebujesz powtórki (ponowne przeczytanie przeszłych wydarzeń w celu debugowania lub odbudowania)
  • Przetwarzanie rozproszone z wieloma procesami roboczymi równolegle (grupy konsumentów)
  • Dziennik audytu i źródło zdarzeń (historia zdarzeń nie może zostać utracona)
  • Konsumenci, którzy mogą przejść do trybu offline i muszą odzyskać utracone wiadomości
# Riepilogo comandi chiave

# PUB/SUB
SUBSCRIBE channel             # Iscriviti a un canale
PSUBSCRIBE pattern:*          # Iscriviti con pattern glob
PUBLISH channel message       # Pubblica (fire-and-forget)
UNSUBSCRIBE channel           # Disiscrivi

# STREAMS - Producer
XADD stream * field value     # Aggiunge messaggio con ID auto
XLEN stream                   # Lunghezza stream
XTRIM stream MAXLEN ~ 10000   # Tronca stream

# STREAMS - Consumer semplice
XREAD COUNT 10 STREAMS stream 0         # Leggi da inizio
XREAD BLOCK 5000 STREAMS stream $       # Blocca aspettando nuovi

# STREAMS - Consumer Groups
XGROUP CREATE stream group $ MKSTREAM    # Crea gruppo
XREADGROUP GROUP group consumer STREAMS stream >  # Leggi non assegnati
XACK stream group message-id             # Acknowledge
XPENDING stream group - + 10            # Lista pending
XCLAIM stream group consumer 30000 id   # Riassegna dopo timeout

Wnioski

Wybór pomiędzy Pub/Sub i Streams sprowadza się do jednego pytania: czy Cię na to stać przegapić wiadomości? Jeśli tak, Pub/Sub oferuje opóźnienie tak nisko, jak to możliwe. Jeśli nie, Streams with Consumer Groups oferuje gwarancję niezawodnego dostarczania z prostotą operacyjną Redis. Następny artykuł omawia skrypty Lua pod kątem złożonych operacji atomowych.

Nadchodzące artykuły z serii Redis

  • Artykuł 3: Skrypty Lua — API operacji atomowych, funkcji EVAL i Redis
  • Artykuł 4: Ograniczanie szybkości — Wiadro tokenów, przesuwane okno, stały licznik
  • Artykuł 5: Zarządzanie sesjami i wzorce pamięci podręcznej — odkładanie na bok, zapisywanie