Поток Python ждёт данные: Condition вместо цикла с sleep

Фоновый поток раз в сто миллисекунд проверяет список заданий. Пусто — снова sleep. При остановке сервиса возникает ещё один флаг, а вместе с ним вопрос: как разбудить потребителя и дать ему закончить уже принятые задания?

Сначала состояние, затем уведомление

Ожидание связывает два действия: проверить условие и заснуть до его изменения. Если между ними нет общей защиты, производитель может добавить задание и отправить уведомление раньше, чем потребитель действительно начнёт ждать. Список уже непустой, а разбудить поток больше некому. Подбор длительности sleep скрывает этот разрыв ценой задержки и лишних проверок.

Condition объединяет блокировку и механизм ожидания. Все чтения списка и признака закрытия выполняются с одной Condition. Потребитель вызывает wait_for с предикатом. Во время ожидания связанная блокировка освобождается, поэтому производитель может войти, изменить данные и уведомить ожидающих. После пробуждения проверка условия повторяется под блокировкой.

Уведомление само по себе не является заданием. Оно сообщает, что состояние могло поменяться. Поэтому проверяем наличие данных либо закрытие очереди, а не запоминаем число notify. Для обычной очереди заданий сначала рассмотрите queue.Queue: готовая реализация удобнее. Собственная Condition оправдана, когда ожидание связано с несколькими частями состояния или специальным правилом завершения.

Определить смысл закрытия

Здесь close запрещает новые задания, но разрешает забрать уже принятые. get возвращает None только после опустошения закрытого буфера. Настоящими заданиями служат целые числа, поэтому None можно выделить под сигнал завершения. Если None допустим во входных данных, используйте отдельный уникальный объект или собственное исключение закрытого буфера.

Под блокировкой выполняется только работа со списком и флагом. Обработка задания происходит после выхода из get, иначе медленный обработчик удерживал бы очередь закрытой для производителей. Пример запускает одного потребителя, добавляет значения, закрывает вход и дожидается завершения. Никакой sleep не нужен: порядок задают операции с состоянием, а не предположение о скорости планировщика.

from collections import deque
from threading import Condition, Thread

class Inbox:
    def __init__(self):
        self.cv = Condition()
        self.items = deque()
        self.closed = False
    def put(self, value):
        with self.cv:
            if self.closed:
                raise RuntimeError("closed")
            self.items.append(value)
            self.cv.notify()
    def get(self):
        with self.cv:
            self.cv.wait_for(lambda: self.items or self.closed)
            return self.items.popleft() if self.items else None
    def close(self):
        with self.cv:
            self.closed = True
            self.cv.notify_all()

inbox = Inbox()
results = []
def consume():
    while (value := inbox.get()) is not None:
        results.append(value * value)
worker = Thread(target=consume, daemon=True)
worker.start()
for value in [2, 3, 4]:
    inbox.put(value)
inbox.close()
worker.join(timeout=2)
assert not worker.is_alive()
assert results == [4, 9, 16]
try:
    inbox.put(5)
except RuntimeError:
    pass
else:
    raise AssertionError("closed inbox accepted an item")
print(results)

Остановка нескольких потребителей

Обычная вставка вызывает notify, потому что одно новое задание достаточно отдать одному ожидающему. Закрытие вызывает notify_all: каждый потребитель должен увидеть, что дальнейших заданий не будет. Если разбудить только одного, остальные смогут остаться ждать навсегда. Уведомляющий поток ещё удерживает блокировку, поэтому проснувшиеся получат доступ после выхода из with.

Тайм-аут у join в примере нужен проверке: зависание превращается в AssertionError. В рабочем приложении определите, что делать с зависшим обработчиком, как записать причину и куда сохранить незавершённое задание. Возврат из ожидания с тайм-аутом также требует решения вызывающего кода: истечение времени не означает закрытия буфера и не должно молча превращаться в потерю работы.

Для расширения добавьте несколько потребителей и проверяйте набор результатов, поскольку их порядок уже может измениться. Отдельно закройте пустой буфер до старта потока. Такая проверка хорошо обнаруживает ошибочную зависимость от уведомления: корректный get сразу увидит closed и завершится, даже если исторический notify никто не услышал.

Следующее практическое решение касается размера буфера. Неограниченная очередь способна расти быстрее, чем потребитель успевает выполнять работу. Если входной поток может быть длинным, задайте вместимость и правило для производителя: ждать свободного места, отказывать или передавать нагрузку дальше. В таком варианте появится второй предикат ожидания и уведомление после извлечения элемента. Не добавляйте ограничение одной проверкой длины вне блокировки: два производителя могут одновременно увидеть свободное место. Для стандартного случая готовая queue.Queue с maxsize уже реализует нужное взаимодействие. Самостоятельную реализацию стоит оставлять тогда, когда специальное состояние приложения действительно требует общей Condition, а не просто ради замены готовой очереди. При этом смысл закрытия должен оставаться понятным обеим сторонам.

Следующий шаг в практике

В бесплатном уроке о тестах pytest разберите проверку нормального результата и отказа, затем перенесите туда сценарии закрытого буфера. Откройте программу профессионального Python: темы GIL и устройства интерпретатора помогут дальше выбирать механизм конкурентного выполнения под конкретную нагрузку.