В чем сложность работы с очередями?

Сложность работы с очередями

Очереди (queues) кажутся простыми структурами данных, но в реальных приложениях их использование сопровождается множеством сложностей, особенно при работе с распределёнными системами, параллелизмом и надёжностью доставки.

1. Параллелизм и race conditions

Проблема: Когда несколько потоков или процессов одновременно добавляют и извлекают элементы из очереди, возникают race conditions.

# НЕБЕЗОПАСНО - race condition
class UnsafeQueue:
    def __init__(self):
        self.items = []
    
    def put(self, item):
        # Потокон-небезопасно!
        self.items.append(item)  # может быть interrupted
    
    def get(self):
        if len(self.items) > 0:  # проверка
            return self.items.pop(0)  # извлечение (race condition!)
        return None

# Пример race condition:
# Поток 1: if len(self.items) > 0:  # True
# Поток 2: if len(self.items) > 0:  # True
# Поток 1: return self.items.pop(0)  # извлекает единственный элемент
# Поток 2: return self.items.pop(0)  # IndexError!

Решение: Использовать thread-safe структуры

import threading
from queue import Queue  # thread-safe из стандартной библиотеки

# БЕЗОПАСНО
safe_queue = Queue(maxsize=100)

# Поток 1
def producer():
    for i in range(5):
        safe_queue.put(i)  # thread-safe

# Поток 2
def consumer():
    while True:
        item = safe_queue.get()  # thread-safe, блокирует если пусто
        print(f"Processing {item}")
        safe_queue.task_done()

thread1 = threading.Thread(target=producer)
thread2 = threading.Thread(target=consumer, daemon=True)
thread1.start()
thread2.start()

2. Deadlock и starvation

Deadlock: Когда потребители ждут, что в очереди появятся элементы, а производители ждут места в очереди.

import threading
from queue import Queue

# Очередь с ограниченным размером
queue = Queue(maxsize=2)

def producer():
    for i in range(5):
        print(f"Producer: putting {i}")
        queue.put(i)  # блокирует если очередь полна
        print(f"Producer: put {i}")

def consumer():
    import time
    time.sleep(2)  # потребитель медленный
    for _ in range(5):
        item = queue.get()
        print(f"Consumer: got {item}")
        time.sleep(0.5)

p = threading.Thread(target=producer)
c = threading.Thread(target=consumer)
p.start()
c.start()
p.join()
c.join()

# Очередь наполняется, производитель блокируется
# Потребитель медленный, очередь не пустеет
# Возможен deadlock если не ограничить время

Решение: Использовать timeout

try:
    queue.put(item, timeout=5)  # ждём максимум 5 секунд
except queue.Full:
    print("Queue is full, dropping item")
    # или повторить, или отправить в другую очередь

try:
    item = queue.get(timeout=5)
except queue.Empty:
    print("Queue is empty")

3. Обработка ошибок и повторы

Проблема: Если задача в очереди обработалась с ошибкой, нужно её переобработать. Но это может привести к бесконечным повторам или потере данных.

# Плохо - задача может быть потеряна
def process_item():
    item = queue.get()
    try:
        risky_operation(item)
    except Exception as e:
        print(f"Error: {e}")  # логируем и забываем
        # item потеряна!
    queue.task_done()

# Лучше - с повторами
def process_item_with_retry(max_retries=3):
    item = queue.get()
    retries = 0
    while retries < max_retries:
        try:
            risky_operation(item)
            queue.task_done()
            return
        except Exception as e:
            retries += 1
            if retries < max_retries:
                print(f"Retry {retries}/{max_retries}: {e}")
                time.sleep(2 ** retries)  # exponential backoff
            else:
                print(f"Failed after {max_retries} retries")
                # Отправляем в dead letter queue
                dead_letter_queue.put(item)
                queue.task_done()

4. Порядок обработки и приоритеты

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

from queue import PriorityQueue

# Приоритетная очередь
priority_queue = PriorityQueue()

# Добавляем задачи с приоритетами (меньшее число = выше приоритет)
priority_queue.put((3, "Low priority task"))
priority_queue.put((1, "High priority task"))
priority_queue.put((2, "Medium priority task"))

while not priority_queue.empty():
    priority, task = priority_queue.get()
    print(f"Processing ({priority}): {task}")

# Вывод:
# Processing (1): High priority task
# Processing (2): Medium priority task
# Processing (3): Low priority task

Усложнение: Что если приоритет может меняться во время обработки?

class Task:
    def __init__(self, task_id, priority, data):
        self.task_id = task_id
        self.priority = priority
        self.data = data
    
    def __lt__(self, other):
        return self.priority < other.priority
    
    def __eq__(self, other):
        return self.task_id == other.task_id

task_queue = PriorityQueue()
task_queue.put(Task(1, 5, "data1"))
task_queue.put(Task(2, 1, "data2"))

while not task_queue.empty():
    task = task_queue.get()
    print(f"Processing task {task.task_id}: {task.data}")

5. Требование At-Least-Once доставки

Проблема: В распределённых системах (RabbitMQ, Kafka, Redis) нужно гарантировать, что задача будет обработана, даже если потребитель упадёт.

import redis
import json

class RobustQueue:
    def __init__(self, redis_client, queue_name):
        self.redis = redis_client
        self.queue_name = queue_name
        self.processing_list = f"{queue_name}:processing"
    
    def put(self, item):
        # Добавляем в основную очередь
        self.redis.rpush(self.queue_name, json.dumps(item))
    
    def get(self):
        # Атомарно: перемещаем из очереди в processing список
        item = self.redis.rpoplpush(self.queue_name, self.processing_list)
        if item:
            return json.loads(item)
        return None
    
    def acknowledge(self, item):
        # После успешной обработки удаляем из processing списка
        self.redis.lrem(self.processing_list, 1, json.dumps(item))
    
    def recover_lost_items(self):
        # При перезагрузке потребителя, переместить items из processing
        # обратно в основную очередь
        items = self.redis.lrange(self.processing_list, 0, -1)
        for item in items:
            self.redis.rpush(self.queue_name, item)
            self.redis.lrem(self.processing_list, 1, item)

6. Масштабирование и распределённые очереди

Проблема: Локальная очередь работает только в одном процессе. Для масштабирования нужны распределённые очереди (Redis, RabbitMQ, Kafka).

# Celery (распределённая очередь задач)
from celery import Celery

app = Celery("tasks", broker="redis://localhost:6379")

@app.task(bind=True, max_retries=3)
def process_data(self, data):
    try:
        # Длительная операция
        result = expensive_operation(data)
        return result
    except Exception as exc:
        # Автоматический retry с exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

# Отправка задачи в очередь
process_data.delay({"key": "value"})

# Мониторинг
celery_app.tasks

7. Обработка дублей

Проблема: При перезагрузке потребителя или сетевых сбоях, одна и та же задача может быть обработана несколько раз (at-least-once доставка).

# Решение через идемпотентность
import hashlib

class IdempotentProcessor:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.processed_ids = "processed_items"
    
    def process_item(self, item):
        # Создаём уникальный ID задачи
        item_id = hashlib.md5(str(item).encode()).hexdigest()
        
        # Проверяем, уже ли была обработана
        if self.redis.sismember(self.processed_ids, item_id):
            print(f"Item {item_id} already processed, skipping")
            return
        
        # Обрабатываем
        result = do_something(item)
        
        # Записываем как обработанную
        self.redis.sadd(self.processed_ids, item_id)
        
        # Устанавливаем TTL на 24 часа (можно забыть старые)
        self.redis.expire(self.processed_ids, 86400)
        
        return result

8. Мониторинг и алерты

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

class MonitoredQueue:
    def __init__(self, queue, max_size=1000):
        self.queue = queue
        self.max_size = max_size
        self.metrics = {
            "total_processed": 0,
            "total_failed": 0,
            "current_size": 0,
        }
    
    def put(self, item):
        self.queue.put(item)
        self.metrics["current_size"] = self.queue.qsize()
        
        # Алерт если очередь переполнена
        if self.queue.qsize() > self.max_size * 0.8:
            self.send_alert(f"Queue at {self.queue.qsize()}/{self.max_size}")
    
    def get_with_stats(self):
        try:
            item = self.queue.get(timeout=5)
            self.metrics["total_processed"] += 1
            return item
        except queue.Empty:
            if self.metrics["total_processed"] == 0:
                self.send_alert("Queue consumer might be stuck")
            raise
    
    def get_health(self):
        return {
            "queue_size": self.queue.qsize(),
            "processed": self.metrics["total_processed"],
            "failed": self.metrics["total_failed"],
            "capacity_usage": f"{self.queue.qsize() / self.max_size * 100:.1f}%",
        }
    
    def send_alert(self, message):
        # Отправить в Slack, PagerDuty, etc.
        print(f"ALERT: {message}")

9. Порядок гарантий

Проблема: Разные очереди дают разные гарантии:

# 1. Локальная очередь (collections.deque, queue.Queue)
# - порядок FIFO
# - теряется при перезагрузке
# - single machine only

# 2. Redis очередь
# - порядок FIFO
# - сохраняется в памяти (с persistence)
# - single machine (или с replication)

# 3. RabbitMQ
# - может гарантировать порядок
# - durable queues (persisted to disk)
# - distributed

# 4. Kafka
# - гарантирует порядок в пределах partition
# - permanent storage
# - highly distributed и scalable

# Выбор зависит от требований
if need_guaranteed_order_and_persistence:
    use_kafka_or_rabbitmq()
elif need_simple_and_fast:
    use_redis_queue()
elif single_process_local_queue:
    use_collections_deque()

10. Graceful shutdown

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

import signal
import time

class GracefulQueue:
    def __init__(self, queue):
        self.queue = queue
        self.shutdown = False
        self.current_item = None
        
        signal.signal(signal.SIGTERM, self._handle_shutdown)
        signal.signal(signal.SIGINT, self._handle_shutdown)
    
    def _handle_shutdown(self, signum, frame):
        print("Shutdown signal received")
        self.shutdown = True
    
    def process_loop(self):
        while not self.shutdown:
            try:
                self.current_item = self.queue.get(timeout=1)
                # Обработка
                process_item(self.current_item)
                self.queue.task_done()
                self.current_item = None
            except queue.Empty:
                continue
        
        # Обработать оставшиеся элементы
        print("Gracefully shutting down, processing remaining items...")
        self.queue.join()  # ждём пока все задачи завершатся
        print("Shutdown complete")

Итог

Очереди на первый взгляд просты, но в production они полны ловушек: race conditions, deadlocks, потеря данных, масштабирование, мониторинг. В реальных приложениях обычно используют готовые решения (Celery с Redis/RabbitMQ для задач, Kafka для потоков данных) с правильной обработкой ошибок, повторами, идемпотентностью и мониторингом.