Освоение межпоточного взаимодействия в Python:синхронизация и обмен данными

Межпоточное взаимодействие — это процесс обеспечения связи и синхронизации между потоками в многопоточной программе Python.

Как правило, потоки в Python используют одно и то же пространство памяти внутри процесса, что позволяет им обмениваться данными и координировать свои действия с помощью общих переменных, объектов и специализированных механизмов синхронизации, предоставляемых модулем потоков.

Чтобы облегчить взаимодействие между потоками, модуль потоков предоставляет различные примитивы синхронизации, такие как объекты «Блокировки», «События», «Условия» и «Семафоры». В этом уроке вы узнаете, как использовать объект Event и Condition для обеспечения связи между потоками в многопоточной программе.

Объект события

Объект Event управляет состоянием внутреннего флага, чтобы потоки могли ждать или устанавливаться. Объект события предоставляет методы для управления состоянием этого флага, позволяя потокам синхронизировать свои действия на основе общих условий.

Первоначально флаг имеет значение false и становится истинным с помощью метода set() и сбрасывается в значение false с помощью методаclear(). Метод wait() блокируется до тех пор, пока флаг не станет истинным.

Ниже приведены ключевые методы объекта Event –

• is_set():возвращает значение True тогда и только тогда, когда внутренний флаг имеет значение true.
• set():установите для внутреннего флага значение true. Все потоки, ожидающие, пока это станет правдой, пробуждаются. Потоки, которые вызывают wait(), когда флаг имеет значение true, вообще не блокируются.
• clear():сброс внутреннего флага в значение false. Впоследствии потоки, вызывающие wait(), будут блокироваться до тех пор, пока не будет вызвана функция set(), чтобы снова установить внутренний флаг в значение true.
• wait(timeout=None):блокировать до тех пор, пока внутренний флаг не станет истинным. Если внутренний флаг при входе равен true, немедленно вернитесь. В противном случае блокируйте до тех пор, пока другой поток не вызовет set(), чтобы установить флаг в значение true, или пока не наступит необязательный тайм-аут. Если присутствует аргумент таймаута, а не 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("Traffic Police Giving GREEN Signal")
 event.set()
 time.sleep(1)
 print("Traffic Police Giving RED Signal")
 event.clear()
def traffic_flow():
 global terminate
 num = 0
 while num < 10 and not terminate:
 print("Waiting for GREEN Signal")
 event.wait()
 print("GREEN Signal ... Traffic can move")
 while event.is_set() and not terminate:
 num += 1
 print("Vehicle No:", num," Crossing the Signal")
 time.sleep(1)
 print("RED Signal ... Traffic has to wait")
event = Event()
t1 = Thread(target=signal_state)
t2 = Thread(target=traffic_flow)
t1.start()
t2.start()
# Terminate the threads after some time
time.sleep(5)
terminate = True
# join all threads to complete
t1.join()
t2.join()
print("Exiting Main Thread")

Вывод

Выполнив приведенный выше код, вы получите следующий результат –

Waiting for GREEN Signal
Traffic Police Giving GREEN Signal
GREEN Signal ... Traffic can move
Vehicle No: 1 Crossing the Signal
Traffic Police Giving RED Signal
RED Signal ... Traffic has to wait
Waiting for GREEN Signal
Traffic Police Giving GREEN Signal
GREEN Signal ... Traffic can move
Vehicle No: 2 Crossing the Signal
Vehicle No: 3 Crossing the Signal
Traffic Police Giving RED Signal
Traffic Police Giving GREEN Signal
Vehicle No: 4 Crossing the Signal
Traffic Police Giving RED Signal
RED Signal ... Traffic has to wait
Traffic Police Giving GREEN Signal
Traffic Police Giving RED Signal
Exiting Main Thread

Объект условия

Объект Condition в модуле потоков 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 потоков, ожидающих условную переменную; это пустая операция, если ни один поток не ожидает.
• notify_all():пробуждает все потоки, ожидающие выполнения этого условия. Этот метод действует как notify(), но пробуждает все ожидающие потоки вместо одного. Если вызывающий поток не получил блокировку при вызове этого метода, возникает ошибка RuntimeError.

Пример

В этом примере демонстрируется простая форма межпоточного взаимодействия с использованием объекта Condition модуля потоков 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("Thread A started")
 with c:
 print("Thread A waiting for permission...")
 c.wait()
 print("Thread A got permission!")
 print("Thread A finished")
def thread_b():
 print("Thread B started")
 with c:
 time.sleep(2)
 print("Notifying Thread A...")
 c.notify()
 print("Thread B finished")
Thread(target=thread_a).start()
Thread(target=thread_b).start()

Вывод

Выполнив приведенный выше код, вы получите следующий результат –

Thread A started
Thread A waiting for permission...
Thread B started
Notifying Thread A...
Thread B finished
Thread A got permission!
Thread A finished

Пример

Вот еще один код, демонстрирующий, как объект 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("Generated random number:", num)
 numbers.append(num)
 print("Notification issued")
 c.notify()
 time.sleep(0.3)
def taskB(c):
 for i in range(5):
 with c:
 print("waiting for update")
 while not numbers: 
 c.wait()
 print("Obtained random number", 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("Done")

Когда вы выполните этот код, он выдаст следующий результат –

waiting for update
Generated random number: 2
Notification issued
Obtained random number 2
Generated random number: 5
Notification issued
waiting for update
Obtained random number 5
Generated random number: 1
Notification issued
waiting for update
Obtained random number 1
Generated random number: 9
Notification issued
waiting for update
Obtained random number 9
Generated random number: 2
Notification issued
waiting for update
Obtained random number 2
Done