Как реализовать схему производитель — потребитель?

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

Готовые очереди:

• Java: ArrayBlockingQueue (ограниченная, на массиве), LinkedBlockingQueue (по умолчанию неограниченная — задайте ёмкость), SynchronousQueue (без буфера — передача из рук в руки); методы put/take блокируют, offer/poll с таймаутом;
• Python: queue.Queue(maxsize=...) для потоков, asyncio.Queue для корутин;
• Go: буферизованный канал make(chan Task, 100).

Реализация на мониторе — классическое упражнение на собеседовании: мьютекс и две условные переменные. put ждёт, пока очередь полна, кладёт и будит потребителей; take ждёт, пока пусто, забирает и будит производителей. Ждать — в цикле while, а не if: при пробуждении условие мог перехватить другой поток.

Ограниченная ёмкость. Если производитель быстрее, неограниченная очередь растёт до исчерпания памяти, а каждая задача ждёт всё дольше. Ограничение заставляет производителя ждать или получить отказ — это и есть backpressure: замедление распространяется к источнику. Альтернативы — отбрасывать старые или новые элементы (для метрик и телеметрии), масштабировать потребителей.

Остановка. Потребители висят на take. Варианты: положить в очередь специальный маркер — «ядовитую пилюлю» (по одной на потребителя); прерывание потоков; закрытие канала в Go (range по каналу завершится).

Масштаб выше. Та же схема между сервисами — брокер сообщений (Kafka, RabbitMQ): очередь переживает перезапуски, потребителей можно добавлять, но появляются вопросы гарантий доставки и идемпотентности.

import threading
from collections import deque

class BoundedQueue:
    def __init__(self, capacity):
        self.items, self.capacity = deque(), capacity
        lock = threading.Lock()
        self.not_empty = threading.Condition(lock)
        self.not_full = threading.Condition(lock)

    def put(self, item):
        with self.not_full:
            while len(self.items) >= self.capacity:
                self.not_full.wait()
            self.items.append(item)
            self.not_empty.notify()

    def take(self):
        with self.not_empty:
            while not self.items:
                self.not_empty.wait()
            item = self.items.popleft()
            self.not_full.notify()
            return item