На этом шаге мы рассмотрим способ решения этой задачи.

Задача

    В вашей программе есть несколько потоков, и вы хотите безопасно коммуницировать или обмениваться данными между ними.

Решение

    Вероятно, самый безопасный путь переслать данные из одного потока в другой - это использовать Queue из библиотеки queue. Чтобы сделать это, создайте экземпляр Queue, который будет общим для всех потоков. Затем потоки должны использовать операции put() или get(), чтобы добавлять или убирать элементы из очереди. Например:

from queue import Queue
from threading import Thread


# Поток, который производит данные
def producer(out_q):
    while True:
        # Производим данные
        . . .
        out_q.put(data)

# Поток, который потребляет данные
def consumer(in_q):
    while True:
        # Получаем данные
        data = in_q.get()
        # Обрабатываем данные
        . . .

# Создаем разделяемую (shared) очередь и запускаем оба потока
q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=producer, args=(q,))
t1.start()
t2.start()

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

    При использовании очередей иногда довольно сложно скоординировать отключение продюсера (производителя) и консьюмера (потребителя). Общепринятое решение этой проблемы опирается на специальное "сторожевое" значение, которое при помещении в очередь заставляет консьюмеров завершаться. Например:

from queue import Queue
from threading import Thread


# Объект, который сигнализирует об отключении
_sentinel = object()
# Поток, который производит данные
def producer(out_q):
while running:
    # Производим данные
    . . .
    out_q.put(data)

    # Поместить стража в очередь, чтобы сигнализировать о завершении
    out_q.put(_sentinel)

# Поток, который потребляет данные
def consumer(in_q):
    while True:
        # Получаем данные
        data = in_q.get()

        # Проверяем на предмет сигнала о завершении
        if data is _sentinel:
            in_q.put(_sentinel)
            break

        # Обрабатываем данные
        . . .

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

    Хотя очереди - это наиболее распространенный механизм коммуникации, вы можете построить собственные структуры данных, которые просто должны иметь требуемую блокировку и синхронизацию. Самый обычный способ это сделать - обернуть ваши структуры данных в условную переменную. Например, вот так вы можете построить потокобезопасную очередь с приоритетом, обсуждавшуюся в 6 шаге:

import heapq
import threading


class PriorityQueue:
    def __init__(self):
        self._queue = []
        self._count = 0
        self._cv = threading.Condition()

    def put(self, item, priority):
        with self._cv:
            heapq.heappush(self._queue, (-priority, self._count, item))
            self._count += 1
            self._cv.notify()

    def get(self):
        with self._cv:
            while len(self._queue) == 0:
                self._cv.wait()
            return heapq.heappop(self._queue)[-1]

    Коммуникация потоков через очередь - однонаправленный и недетерминистический процесс. В общем случае нет возможности узнать, когда получающий поток в действительности получил и обработал сообщение. Однако объекты Queue предоставляют некоторые базовые возможности завершения, как показано в методах task_done() и join() в этом примере:

from queue import Queue
from threading import Thread


# Поток, который производит данные
def producer(out_q):
    while running:
        # Произвести данные
        . . .
        out_q.put(data)

# Поток, который потребляет данные
def consumer(in_q):
    while True:
        # Получаем данные
        data = in_q.get()
        # Обрабатываем данные
        . . .
        # Сигнализируем о завершении
        in_q.task_done()

# Создаем разделяемую (shared) очередь и запускаем оба потока
q = Queue()
t1 = Thread(target=consumer, args=(q,))
t2 = Thread(target=producer, args=(q,))
t1.start()
t2.start()
# Ждем, пока все произведенные элементы будут потреблены
q.join()

Обсуждение

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

    Предостережение тем, кто будет использовать потоковые очереди: помещение элемента в очередь не означает, что элемент будет скопирован. Поэтому коммуникация в действительности означает передачу ссылки на объект между потоками. Если вы беспокоитесь о появлении общего состояния (shared state), имеет смысл передавать только неизменяемые структуры данных (целые числа, строки или кортежи) или же делать "глубокие копии" помещаемых в очередь элементов. Например:

from queue import Queue
from threading import Thread
import copy


# Поток, который производит данные
def producer(out_q):
    while True:
        # Производим данные
        . . .
        out_q.put(copy.deepcopy(data))

# Поток, который потребляет данные
def consumer(in_q):
    while True:
        # Получаем данные
        data = in_q.get()
        # Обрабатываем данные
        . . .

    Объекты Queue предоставляют несколько дополнительных возможностей, которые могут быть полезны в некоторых контекстах. Если вы создаете Queue, задавая размер - Queue(N) (что необязательно), - то в очередь можно будет поместить ограниченное количество элементов, прежде чем put() заблокирует продюсера. Определение верхней границы количества элементов в очереди может иметь смысл, если есть разница в скорости работы продюсера и консьюмера. Например, продюсер может генерировать элементы гораздо быстрее, чем консьюмер может их обработать. С другой стороны, если очередь блокируется при заполнении, то это может вызвать непредумышленный каскадный эффект, который распространится на всю программу и вызовет дедлок или сбои в работе. В общем, проблема передачи потока управления между общающимися потоками намного сложнее, чем кажется. Если вы внезапно обнаружите, что пытаетесь решить проблему за счет манипуляций с размерами очередей, то это признак хрупкого дизайна или каких-то других врожденных проблем с масштабированием.

    И метод get(), и метод put() поддерживают неблокируемость и тайм-ауты. Например:

import queue


q = queue.Queue()
try:
    data = q.get(block=False)
except queue.Empty:
    . . .

try:
    q.put(item, block=False)
except queue.Full:
    . . .

try:
    data = q.get(timeout=5.0)
except queue.Empty:
    . . .

    Обе эти опции могут быть использованы для избежания проблемы неопределенного блокирования при какой-либо операции с очередью. Например, неблокирующий put() может быть использован с очередью фиксированного размера, чтобы реализовать различные типы обрабатыващего кода, который будет срабатывать при заполнении очереди. Например, можно делать запись в лог и отбрасывать элемент:

def producer(q):
    . . .
    try:
        q.put(item, block=False)
    except queue.Full:
         log.warning('queued item %r discarded!', item)

    Тайм-ауты полезны, если вы пытаетесь заставить потоки-консьюмеры периодически осуществлять операции типа q.get(), чтобы они могли проверить такие вещи, как флаг завершения (описано на 218 шаге):

_running = True
def consumer(q):
    while _running:
        try:
            item = q.get(timeout=5.0)
            # Обрабатываем элемент
            . . .
        except queue.Empty:
            pass

    И последнее: есть методы q.qsize(), q.full() и q.empty(), которые позволяют узнать текущий размер и статус очереди. Однако стоит знать, что все они ненадежны в многопоточном окружении. Например, вызов q.empty() может сообщить вам, что очередь пуста, но за то время, которое пройдет от момента вызова, другой поток может поместить элементы в очередь. Лучше писать код, который не полагается на такие функции.

    На следующем шаге мы рассмотрим блокировку критически важных участков.

Предыдущий шаг Содержание Следующий шаг