Мастерство межпоточного взаимодействия в Python: синхронизация и обмен данными
Межпоточное взаимодействие (Inter-Thread Communication) относится к процессу обеспечения коммуникации и синхронизации между потоками в многопоточной программе на Python.
Как правило, потоки в Python разделяют одно и то же пространство памяти внутри процесса, что позволяет им обмениваться данными и координировать свою активность через общие переменные, объекты и специализированные механизмы синхронизации, предоставляемые модулем threading.
Для облегчения межпоточного взаимодействия модуль threading предоставляет различные примитивы синхронизации, такие как Locks (блокировки), Events (события), Conditions (условия) и Semaphores (семафоры). В этом руководстве вы узнаете, как использовать объекты Event и Condition для обеспечения коммуникации между потоками в многопоточной программе.
Объект Event
Объект Event управляет состоянием внутреннего флага, чтобы потоки могли ждать или устанавливать его. Объект Event предоставляет методы для контроля состояния этого флага, позволяя потокам синхронизировать свою активность на основе общих условий.
Флаг изначально false и становится true с методом set() и сбрасывается в false с методом clear(). Метод wait() блокирует до тех пор, пока флаг не станет true.
Ниже приведены ключевые методы объекта Event:
- is_set(): Возвращает True тогда и только тогда, когда внутренний флаг true.
- set(): Устанавливает внутренний флаг в true. Все потоки, ждущие его становления true, пробуждаются. Потоки, вызывающие wait() после того, как флаг стал true, не блокируются вообще.
- clear(): Сбрасывает внутренний флаг в false. Впоследствии потоки, вызывающие wait(), будут блокироваться до вызова set() для установки флага в true снова.
- wait(timeout=None): Блокирует до тех пор, пока внутренний флаг не станет true. Если внутренний флаг true при входе, возвращается немедленно. Иначе блокирует до тех пор, пока другой поток не вызовет set() для установки флага в true, или пока не наступит опциональный таймаут. Когда аргумент timeout присутствует и не None, это должно быть число с плавающей точкой, указывающее таймаут для операции в секундах.
Пример
Следующий код пытается смоделировать дорожный трафик, контролируемый состоянием светофора: ЗЕЛЕНЫЙ или КРАСНЫЙ.
В программе два потока, целевые для двух разных функций. Функция signal_state() периодически устанавливает и сбрасывает событие, указывая на смену сигнала с ЗЕЛЕНОГО на КРАСНЫЙ.
Функция traffic_flow() ждет установки события и запускает цикл, пока оно остается установленным.
from threading import Event, Thread
import time
terminate = False
def signal_state():
global terminate
while not terminate:
time.sleep(0.5)
print("Полиция дорожного движения дает ЗЕЛЕНЫЙ сигнал")
event.set()
time.sleep(1)
print("Полиция дорожного движения дает КРАСНЫЙ сигнал")
event.clear()
def traffic_flow():
global terminate
num = 0
while num < 10 and not terminate:
print("Ожидание ЗЕЛЕНОГО сигнала")
event.wait()
print("ЗЕЛЕНЫЙ сигнал ... Транспорт может двигаться")
while event.is_set() and not terminate:
num += 1
print("Транспорт №:", num, " Проезжает перекресток")
time.sleep(1)
print("КРАСНЫЙ сигнал ... Транспорт должен ждать")
event = Event()
t1 = Thread(target=signal_state)
t2 = Thread(target=traffic_flow)
t1.start()
t2.start()
# Завершение потоков через некоторое время
time.sleep(5)
terminate = True
# join всех потоков для завершения
t1.join()
t2.join()
print("Выход из главного потока")
Вывод
При выполнении вышеуказанного кода вы получите следующий вывод:
Ожидание ЗЕЛЕНОГО сигнала Полиция дорожного движения дает ЗЕЛЕНЫЙ сигнал ЗЕЛЕНЫЙ сигнал ... Транспорт может двигаться Транспорт №: 1 Проезжает перекресток Полиция дорожного движения дает КРАСНЫЙ сигнал КРАСНЫЙ сигнал ... Транспорт должен ждать Ожидание ЗЕЛЕНОГО сигнала Полиция дорожного движения дает ЗЕЛЕНЫЙ сигнал ЗЕЛЕНЫЙ сигнал ... Транспорт может двигаться Транспорт №: 2 Проезжает перекресток Транспорт №: 3 Проезжает перекресток Полиция дорожного движения дает КРАСНЫЙ сигнал Полиция дорожного движения дает ЗЕЛЕНЫЙ сигнал Транспорт №: 4 Проезжает перекресток Полиция дорожного движения дает КРАСНЫЙ сигнал КРАСНЫЙ сигнал ... Транспорт должен ждать Полиция дорожного движения дает ЗЕЛЕНЫЙ сигнал Полиция дорожного движения дает КРАСНЫЙ сигнал Выход из главного потока
Объект Condition
Объект Condition в модуле threading Python предоставляет более продвинутый механизм синхронизации. Он позволяет потокам ждать уведомления от другого потока перед продолжением. Объекты Condition всегда ассоциированы с блокировкой и предоставляют механизмы сигнализации между потоками.
Ниже приведен синтаксис класса threading.Condition():
threading.Condition(lock=None)
Ниже приведены ключевые методы объекта Condition:
- acquire(*args): Получает базовую блокировку. Этот метод вызывает соответствующий метод на базовой блокировке; возвращаемое значение — то, что возвращает тот метод.
- release(): Освобождает базовую блокировку. Этот метод вызывает соответствующий метод на базовой блокировке; возвращаемого значения нет.
- wait(timeout=None): Этот метод освобождает базовую блокировку, а затем блокирует до пробуждения вызовом notify() или notify_all() для той же условной переменной в другом потоке, или до наступления опционального таймаута. После пробуждения или таймаута он повторно получает блокировку и возвращается.
- wait_for(predicate, timeout=None): Этот утилитарный метод может вызывать wait() повторно до удовлетворения предиката или до таймаута. Возвращаемое значение — последнее возвращаемое значение предиката и будет оцениваться как False, если метод завершился по таймауту.
- notify(n=1): Этот метод пробуждает не более n потоков, ждущих условной переменной; это no-op, если нет ждущих потоков.
- notify_all(): Пробуждает все потоки, ждущие на этом условии. Этот метод действует как notify(), но пробуждает все ждущие потоки вместо одного. Если вызывающий поток не получил блокировку при вызове этого метода, поднимается RuntimeError.
Пример
Этот пример демонстрирует простую форму межпоточного взаимодействия с использованием объекта Condition модуля threading Python. Здесь thread_a и thread_b общаются с помощью объекта Condition, thread_a ждет, пока не получит уведомление от thread_b. thread_b спит 2 секунды перед уведомлением thread_a и затем завершается.
from threading import Condition, Thread
import time
c = Condition()
def thread_a():
print("Поток А запущен")
with c:
print("Поток А ждет разрешения...")
c.wait()
print("Поток А получил разрешение!")
print("Поток А завершен")
def thread_b():
print("Поток Б запущен")
with c:
time.sleep(2)
print("Уведомление Потока А...")
c.notify()
print("Поток Б завершен")
Thread(target=thread_a).start()
Thread(target=thread_b).start()
Вывод
При выполнении вышеуказанного кода вы получите следующий вывод:
Поток А запущен Поток А ждет разрешения... Поток Б запущен Уведомление Потока А... Поток Б завершен Поток А получил разрешение! Поток А завершен
Пример
Здесь еще один код, демонстрирующий, как объект Condition используется для обеспечения коммуникации между потоками. В этом поток t2 запускает функцию taskB(), а поток t1 запускает функцию taskA(). Поток t1 получает условие и уведомляет его.
К этому времени поток t2 находится в состоянии ожидания. После освобождения условия ожидающий поток продолжает потреблять случайное число, сгенерированное уведомляющей функцией.
from threading import Condition, Thread
import time
import random
numbers = []
def taskA(c):
for _ in range(5):
with c:
num = random.randint(1, 10)
print("Сгенерировано случайное число:", num)
numbers.append(num)
print("Уведомление выпущено")
c.notify()
time.sleep(0.3)
def taskB(c):
for i in range(5):
with c:
print("ожидание обновления")
while not numbers:
c.wait()
print("Получено случайное число", numbers.pop())
time.sleep(0.3)
c = Condition()
t1 = Thread(target=taskB, args=(c,))
t2 = Thread(target=taskA, args=(c,))
t1.start()
t2.start()
t1.join()
t2.join()
print("Готово")
При выполнении этого кода будет получен следующий вывод:
ожидание обновления Сгенерировано случайное число: 2 Уведомление выпущено Получено случайное число 2 Сгенерировано случайное число: 5 Уведомление выпущено ожидание обновления Получено случайное число 5 Сгенерировано случайное число: 1 Уведомление выпущено ожидание обновления Получено случайное число 1 Сгенерировано случайное число: 9 Уведомление выпущено ожидание обновления Получено случайное число 9 Сгенерировано случайное число: 2 Уведомление выпущено ожидание обновления Получено случайное число 2 Готово
Python
- Мастерство преобразования типов в Python: Полное руководство
- Python DSL: адаптированные решения для специализированных доменов
- Доступ к веб-данным с urllib в Python: практическое руководство
- Мастерство отладки Python: эффективные техники и лучшие практики
- Мастерство условных операторов if…else в Python
- Освоение enumerate() в Python: циклы с индексами для списков, кортежей, строк и словарей
- Шпаргалка по основам Python: основные типы данных, словари, списки и функции
- Операции с матрицами в Python с NumPy: транспонирование, сложение, умножение и срезы — практическое руководство
- Освоение типов данных в Python: практическое руководство
- Мастерство парсинга XML в Python: практическое руководство с Minidom и ElementTree