# Потоки и группы в Redis: подробное руководство для Python-разработчиков

2025-10-18 8 min read

Table of Contents

Потоки и группы в Redis: подробное руководство для Python-разработчиков

Введение в Redis Streams

Redis Streams — это тип данных, появившийся в Redis 5.0, который представляет собой лог-ориентированную структуру данных, идеально подходящую для обработки потоков событий, сообщений и данных временных рядов. В отличие от традиционных брокеров сообщений, Redis Streams сочетает в себе простоту использования с высокой производительностью и надежностью.

Что такое потоки Redis?

Поток Redis — это упорядоченная последовательность записей, где каждая запись состоит из:

Уникального идентификатора (обычно основанного на времени)
Набора пар ключ-значение

Основные характеристики:

Упорядоченность: записи хранятся в порядке их добавления
Персистентность: данные сохраняются на диск
Высокая производительность: обработка сотен тысяч операций в секунду
Гибкость: поддержка различных сценариев использования

Основы работы с потоками в Python

Установка и настройка

import redis
import json
from datetime import datetime

# Подключение к Redis
r = redis.Redis(
    host='localhost',
    port=6379,
    db=0,
    decode_responses=True  # автоматическое декодирование из bytes в str
)

Добавление записей в поток

def add_user_activity(user_id, action, details):
    """Добавление активности пользователя в поток"""
    record = {
        'user_id': user_id,
        'action': action,
        'details': json.dumps(details),
        'timestamp': datetime.now().isoformat()
    }

    # XADD добавляет запись в поток
    # * означает автоматическую генерацию ID
    # maxlen ограничивает длину потока
    return r.xadd(
        'user:activities',
        record,
        maxlen=10000,  # сохранять только последние 10000 записей
        approximate=True  # приблизительное ограничение для оптимизации
    )

# Пример использования
activity_id = add_user_activity(
    user_id=123,
    action='login',
    details={'ip': '192.168.1.1', 'browser': 'Chrome'}
)
print(f"Добавлена запись с ID: {activity_id}")

Чтение записей из потока

def read_recent_activities(count=10):
    """Чтение последних записей из потока"""
    # XREVRANGE читает записи в обратном порядке (от новых к старым)
    activities = r.xrevrange(
        'user:activities',
        count=count
    )

    result = []
    for activity_id, data in activities:
        result.append({
            'id': activity_id,
            'user_id': data['user_id'],
            'action': data['action'],
            'details': json.loads(data['details']),
            'timestamp': data['timestamp']
        })

    return result

# Чтение в прямом порядке
def read_activities_range(start='-', end='+', count=50):
    """Чтение записей в диапазоне"""
    activities = r.xrange(
        'user:activities',
        min=start,  # - означает самую старую запись
        max=end,    # + означает самую новую запись
        count=count
    )
    return activities

Потребительские группы (Consumer Groups)

Концепция потребительских групп

Потребительские группы позволяют распределить обработку сообщений из потока между несколькими потребителями, обеспечивая:

Распределенную обработку: несколько worker’ов могут обрабатывать сообщения параллельно
Гарантию доставки: сообщения не теряются при сбоях
Балансировку нагрузки: автоматическое распределение сообщений между потребителями
Отслеживание прогресса: мониторинг обработки сообщений

Создание потребительской группы

def setup_consumer_group(stream_name, group_name):
    """Создание потребительской группы"""
    try:
        # XGROUP CREATE создает группу потребителей
        r.xgroup_create(
            name=stream_name,
            groupname=group_name,
            id='0',  # начать чтение с самой старой записи
            mkstream=True  # создать поток, если не существует
        )
        print(f"Создана группа '{group_name}' для потока '{stream_name}'")
    except redis.exceptions.ResponseError as e:
        if "BUSYGROUP" in str(e):
            print(f"Группа '{group_name}' уже существует")
        else:
            raise

# Создание группы для обработки пользовательских активностей
setup_consumer_group('user:activities', 'activity-processors')

Потребитель (Consumer)

class ActivityConsumer:
    def __init__(self, stream_name, group_name, consumer_name):
        self.stream_name = stream_name
        self.group_name = group_name
        self.consumer_name = consumer_name
        self.redis = redis.Redis(decode_responses=True)

    def process_activity(self, activity_data):
        """Обработка активности пользователя"""
        try:
            print(f"Обработка активности: {activity_data['action']}")
            # Здесь может быть сложная логика обработки
            # Например, сохранение в базу данных, отправка уведомлений и т.д.

            # Имитация обработки
            import time
            time.sleep(0.1)

            return True
        except Exception as e:
            print(f"Ошибка обработки: {e}")
            return False

    def start_consuming(self, batch_size=10, block_time=5000):
        """Запуск потребления сообщений"""
        print(f"Запуск потребителя '{self.consumer_name}'...")

        while True:
            try:
                # XREADGROUP чтение сообщений из группы
                messages = self.redis.xreadgroup(
                    groupname=self.group_name,
                    consumername=self.consumer_name,
                    streams={self.stream_name: '>'},  # > означает новые сообщения
                    count=batch_size,
                    block=block_time  # блокировка в миллисекундах
                )

                if not messages:
                    continue

                for stream_name, stream_messages in messages:
                    for message_id, message_data in stream_messages:
                        print(f"Получено сообщение {message_id}")

                        # Обработка сообщения
                        success = self.process_activity(message_data)

                        if success:
                            # XACK подтверждение успешной обработки
                            self.redis.xack(
                                self.stream_name,
                                self.group_name,
                                message_id
                            )
                            print(f"Сообщение {message_id} обработано и подтверждено")
                        else:
                            print(f"Ошибка обработки сообщения {message_id}")

            except KeyboardInterrupt:
                print("Остановка потребителя...")
                break
            except Exception as e:
                print(f"Ошибка: {e}")
                import time
                time.sleep(5)  # пауза перед повторной попыткой

Запуск нескольких потребителей

import threading

def run_consumer(consumer_id):
    """Запуск потребителя в отдельном потоке"""
    consumer = ActivityConsumer(
        stream_name='user:activities',
        group_name='activity-processors',
        consumer_name=f'consumer-{consumer_id}'
    )
    consumer.start_consuming()

# Запуск нескольких потребителей
def start_consumer_pool(num_consumers=3):
    threads = []

    for i in range(num_consumers):
        thread = threading.Thread(target=run_consumer, args=(i,))
        thread.daemon = True
        threads.append(thread)
        thread.start()

    # Ожидание завершения (в реальном приложении может быть бесконечным)
    for thread in threads:
        thread.join()

# Запуск пула потребителей
start_consumer_pool(3)

Мониторинг и управление

Отслеживание состояния группы

def monitor_consumer_group(stream_name, group_name):
    """Мониторинг состояния потребительской группы"""

    # Информация о группе
    group_info = r.xinfo_groups(stream_name)
    print("Информация о группах:")
    for group in group_info:
        print(f"  Группа: {group['name']}")
        print(f"  Потребители: {group['consumers']}")
        print(f"  Ожидающие: {group['pending']}")
        print(f"  Последнее сообщение: {group['last-delivered-id']}")
        print("  ---")

    # Информация о потребителях
    consumers_info = r.xinfo_consumers(stream_name, group_name)
    print("Информация о потребителях:")
    for consumer in consumers_info:
        print(f"  Потребитель: {consumer['name']}")
        print(f"  Ожидающие: {consumer['pending']}")
        print(f"  Простой: {consumer['idle']}мс")
        print("  ---")

# Мониторинг ожидающих сообщений
def check_pending_messages(stream_name, group_name):
    """Проверка ожидающих сообщений"""
    pending = r.xpending(stream_name, group_name)
    print(f"Всего ожидающих: {pending['pending']}")

    if pending['pending'] > 0:
        # Детальная информация об ожидающих сообщениях
        detailed_pending = r.xpending_range(
            stream_name,
            group_name,
            min='-',  # самая старая
            max='+',  # самая новая
            count=10
        )

        print("Детальная информация об ожидающих сообщениях:")
        for msg in detailed_pending:
            print(f"  ID: {msg['message_id']}")
            print(f"  Потребитель: {msg['consumer']}")
            print(f"  Время ожидания: {msg['idle']}мс")
            print(f"  Количество доставок: {msg['delivered']}")
            print("  ---")

# Перераспределение застрявших сообщений
def claim_stuck_messages(stream_name, group_name, consumer_name, min_idle_time=30000):
    """Забрать сообщения, которые долго обрабатываются"""

    # Поиск сообщений, которые не обрабатываются дольше min_idle_time мс
    pending = r.xpending_range(
        stream_name,
        group_name,
        min='-',
        max='+',
        count=100
    )

    stuck_messages = []
    for msg in pending:
        if msg['idle'] > min_idle_time:
            stuck_messages.append(msg['message_id'])

    if stuck_messages:
        print(f"Найдено {len(stuck_messages)} застрявших сообщений")

        # XCLAIM забирает сообщения у текущего потребителя
        claimed = r.xclaim(
            stream_name,
            group_name,
            consumer_name,
            min_idle_time,
            stuck_messages
        )

        print(f"Перераспределено {len(claimed)} сообщений")
        return claimed

    return []

Практические примеры использования

Система обработки заказов

class OrderProcessingSystem:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)
        self.setup_streams()

    def setup_streams(self):
        """Настройка потоков для системы заказов"""
        streams = [
            'orders:new',
            'orders:processing',
            'orders:completed',
            'orders:failed'
        ]

        for stream in streams:
            try:
                self.redis.xgroup_create(
                    stream,
                    'order-processors',
                    id='0',
                    mkstream=True
                )
            except redis.exceptions.ResponseError:
                pass

    def place_order(self, order_data):
        """Размещение нового заказа"""
        order_id = self.redis.xadd(
            'orders:new',
            order_data,
            maxlen=100000
        )
        print(f"Размещен заказ {order_id}")
        return order_id

    def process_orders(self, consumer_id):
        """Обработка заказов"""
        while True:
            try:
                messages = self.redis.xreadgroup(
                    groupname='order-processors',
                    consumername=f'order-processor-{consumer_id}',
                    streams={'orders:new': '>'},
                    count=1,
                    block=5000
                )

                if not messages:
                    continue

                for stream_name, stream_messages in messages:
                    for message_id, order_data in stream_messages:
                        print(f"Обработка заказа {message_id}")

                        try:
                            # Имитация обработки заказа
                            self.process_order_steps(order_data)

                            # Перемещение в поток завершенных заказов
                            self.redis.xadd(
                                'orders:completed',
                                {**order_data, 'processed_by': consumer_id}
                            )

                            # Подтверждение обработки
                            self.redis.xack('orders:new', 'order-processors', message_id)

                        except Exception as e:
                            print(f"Ошибка обработки заказа {message_id}: {e}")
                            # Перемещение в поток неудачных заказов
                            self.redis.xadd(
                                'orders:failed',
                                {**order_data, 'error': str(e)}
                            )
                            self.redis.xack('orders:new', 'order-processors', message_id)

            except Exception as e:
                print(f"Ошибка потребителя: {e}")

    def process_order_steps(self, order_data):
        """Пошаговая обработка заказа"""
        # Шаг 1: Проверка наличия товара
        self.check_inventory(order_data)

        # Шаг 2: Проверка платежа
        self.process_payment(order_data)

        # Шаг 3: Подготовка к отправке
        self.prepare_shipment(order_data)

        print("Заказ успешно обработан")

    def check_inventory(self, order_data):
        """Проверка наличия товара"""
        # Имитация проверки инвентаря
        import random
        if random.random() < 0.05:  # 5% вероятность отсутствия товара
            raise Exception("Товар отсутствует на складе")

    def process_payment(self, order_data):
        """Обработка платежа"""
        # Имитация обработки платежа
        import random
        if random.random() < 0.02:  # 2% вероятность отказа платежа
            raise Exception("Ошибка обработки платежа")

    def prepare_shipment(self, order_data):
        """Подготовка к отправке"""
        # Имитация подготовки отправки
        import time
        time.sleep(0.2)

Система сбора метрик и аналитики

class MetricsCollector:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)
        self.setup_metrics_streams()

    def setup_metrics_streams(self):
        """Настройка потоков для сбора метрик"""
        try:
            self.redis.xgroup_create(
                'metrics:raw',
                'metrics-processors',
                id='0',
                mkstream=True
            )
        except redis.exceptions.ResponseError:
            pass

    def collect_metric(self, metric_name, value, tags=None):
        """Сбор метрики"""
        metric_data = {
            'name': metric_name,
            'value': str(value),
            'timestamp': datetime.now().isoformat(),
            'tags': json.dumps(tags or {})
        }

        self.redis.xadd('metrics:raw', metric_data)

    def process_metrics(self):
        """Обработка и агрегация метрик"""
        while True:
            try:
                messages = self.redis.xreadgroup(
                    groupname='metrics-processors',
                    consumername='metrics-aggregator',
                    streams={'metrics:raw': '>'},
                    count=100,
                    block=1000
                )

                if not messages:
                    continue

                aggregated = {}

                for stream_name, stream_messages in messages:
                    for message_id, metric_data in stream_messages:
                        metric_name = metric_data['name']
                        value = float(metric_data['value'])

                        # Агрегация по имени метрики
                        if metric_name not in aggregated:
                            aggregated[metric_name] = {
                                'count': 0,
                                'sum': 0,
                                'min': float('inf'),
                                'max': float('-inf')
                            }

                        agg = aggregated[metric_name]
                        agg['count'] += 1
                        agg['sum'] += value
                        agg['min'] = min(agg['min'], value)
                        agg['max'] = max(agg['max'], value)

                        # Подтверждение обработки
                        self.redis.xack('metrics:raw', 'metrics-processors', message_id)

                # Сохранение агрегированных данных
                for metric_name, stats in aggregated.items():
                    if stats['count'] > 0:
                        stats['avg'] = stats['sum'] / stats['count']

                        self.redis.xadd('metrics:aggregated', {
                            'metric': metric_name,
                            'period': '1min',
                            'count': stats['count'],
                            'average': stats['avg'],
                            'min': stats['min'],
                            'max': stats['max'],
                            'timestamp': datetime.now().isoformat()
                        })

            except Exception as e:
                print(f"Ошибка обработки метрик: {e}")

Лучшие практики и рекомендации

1. Проектирование потоков

# Хорошая практика: использование префиксов и структурированных имен
STREAMS = {
    'user:activities': 'Действия пользователей',
    'orders:events': 'События заказов',
    'notifications:outgoing': 'Исходящие уведомления',
    'metrics:system': 'Системные метрики'
}

# Плохая практика: неструктурированные имена
BAD_STREAMS = ['stream1', 'data', 'messages']

2. Обработка ошибок и повторные попытки

class RobustConsumer:
    def __init__(self, stream_name, group_name, consumer_name, max_retries=3):
        self.stream_name = stream_name
        self.group_name = group_name
        self.consumer_name = consumer_name
        self.max_retries = max_retries
        self.redis = redis.Redis(decode_responses=True)

    def process_with_retry(self, message_id, message_data):
        """Обработка с повторными попытками"""
        for attempt in range(self.max_retries):
            try:
                self.process_message(message_data)
                return True
            except Exception as e:
                print(f"Попытка {attempt + 1} не удалась: {e}")
                if attempt == self.max_retries - 1:
                    # Последняя попытка не удалась
                    self.handle_failure(message_id, message_data, e)
                    return False
                import time
                time.sleep(2 ** attempt)  # Экспоненциальная задержка

        return False

    def handle_failure(self, message_id, message_data, error):
        """Обработка неудачных сообщений"""
        # Запись в поток ошибок
        self.redis.xadd('stream:errors', {
            'original_stream': self.stream_name,
            'message_id': message_id,
            'error': str(error),
            'timestamp': datetime.now().isoformat(),
            'consumer': self.consumer_name
        })

3. Мониторинг и алертинг

class StreamMonitor:
    def __init__(self):
        self.redis = redis.Redis(decode_responses=True)

    def check_stream_health(self, stream_name, warning_threshold=1000):
        """Проверка здоровья потока"""
        stream_length = self.redis.xlen(stream_name)
        pending_messages = self.get_pending_count(stream_name)

        health_status = {
            'stream': stream_name,
            'length': stream_length,
            'pending': pending_messages,
            'timestamp': datetime.now().isoformat()
        }

        # Проверка предупреждений
        if pending_messages > warning_threshold:
            health_status['alert'] = 'HIGH_PENDING_MESSAGES'
            self.send_alert(health_status)

        return health_status

    def get_pending_count(self, stream_name):
        """Получение количества ожидающих сообщений"""
        try:
            groups = self.redis.xinfo_groups(stream_name)
            total_pending = sum(int(group['pending']) for group in groups)
            return total_pending
        except:
            return 0

    def send_alert(self, health_status):
        """Отправка алерта"""
        print(f"ALERT: {health_status['alert']} в потоке {health_status['stream']}")
        print(f"Ожидающие сообщения: {health_status['pending']}")

Заключение

Redis Streams и потребительские группы предоставляют мощный инструментарий для построения масштабируемых и отказоустойчивых систем обработки потоков данных в Python. Ключевые преимущества:

Высокая производительность: обработка сотен тысяч сообщений в секунду
Надежность: гарантированная доставка и отслеживание прогресса
Масштабируемость: простое распределение нагрузки между потребителями
Гибкость: поддержка различных сценариев использования

При правильной реализации Redis Streams может стать основой для построения сложных event-driven архитектур, систем аналитики в реальном времени и распределенных обработчиков данных.

Примеры кода в этой статье демонстрируют основные принципы работы с потоками и группами Redis в Python, но в реальных проектах следует дополнительно учитывать:

Безопасность подключения к Redis
Мониторинг и логирование
Обработку исключительных ситуаций
Масштабирование и балансировку нагрузки

Использование Redis Streams открывает новые возможности для создания эффективных и масштабируемых приложений, способных обрабатывать большие объемы данных в реальном времени.

Next: Эволюционная архитектура в Python

Спасибо, что прочитали статью. Посмотрите другие материалы в архиве, там много практических разборов по разным технологиям.

python Series

# Динамическая типизация, изменяемость типов и система типов в Python

python 1 / 187

2025-03-18 3 min read

Динамическая типизация, изменяемость типов и система типов в Python Python — язык с богатой и гибкой системой типов, которая делает его удобным для разработки, но требует понимания ключевых…