Объекты Condition

Объект Condition - Синхронизация потоков в Python

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

Давайте рассмотрим простой пример, чтобы понять это. В задаче «Производитель-Потребитель», если есть один производитель, производящий какой-либо предмет, и один Потребитель, потребляющий его, то до тех пор, пока Производитель не произведет элемент, Потребитель не может его потреблять. Следовательно, Потребитель ждет, пока Произведенный не произведет предмет. И обязанность производителя - сообщить потребителю, что товар доступен для потребления, как только он будет успешно произведен.

И если есть несколько Потребителей, потребляющих предмет, произведенный Производителем, то Производитель должен проинформировать всех Потребителей о новом произведенном предмете.

Это идеальный вариант использования объекта условия в многопоточности в Python.

Объект Condition: wait (), notify () и notifyAll ()

Теперь, когда мы знаем, для чего используется объект условия в многопоточности Python, давайте посмотрим на его синтаксис:

condition = threading.Condition([lock])

Объект условия принимает в качестве аргумента необязательный объект блокировки. Если мы ничего не предоставляем, создается блокировка по умолчанию.

У объекта условия есть методы acquire() и release(), которые вызывают соответствующие методы связанной блокировки. Он также имеет метод wait() и методы notify() и notifyAll(). Эти три должны вызываться только после того, как вызывающий поток получил блокировку.

Методы класса Condition

Ниже приведены методы класса условий:

acquire(*args) method

Этот метод используется для получения блокировки. Этот метод вызывает соответствующий метод acquire() для базовой блокировки, присутствующей в объекте условия; возвращаемое значение - это то, что возвращает этот метод.

release() method

Этот метод используется для снятия блокировки. Этот метод вызывает соответствующий метод release() для базовой блокировки, присутствующей в объекте условия.

wait([timeout]) method

Этот метод используется, чтобы заблокировать поток и заставить его ждать, пока какой-либо другой поток не уведомит его, вызвав метод notify() или notifyAll() для того же объекта условия или пока не истечет время ожидания.

Это должно быть вызвано только тогда, когда вызывающий поток получил блокировку.

При вызове этот метод снимает блокировку, а затем блокирует поток до тех пор, пока он не будет разбужен вызовом notify() или notifyAll() для той же переменной условия из другого потока или до истечения времени ожидания.

Этот метод возвращает True, если он выпущен из-за метода notify() или notifyAll(), иначе, если произойдет тайм-аут, этот метод вернет логическое значение False.

notify() method

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

notifyAll() method

Он пробуждает все потоки, ожидающие этого условия. Этот метод действует как метод notify(), но пробуждает все ожидающие потоки вместо одного.

Время для примера!

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

import threading
import time
from random import randint

class SomeItem:
  # метод инициализации
  def __init__(self):
    # initialize empty list
    self.list = []
  
  # метод производителя; добавляет в список
  def produce(self, item):
    print("Adding item to list...")
    self.list.append(item)
    
  # метод потребителя; удаляет элемент из списка
  def consume(self):
    print("consuming item from list...")
    item = self.list[0]
    print("Item consumed: ", item)
    self.list.remove(item)

def producer(si, cond):
    r = randint(1,5)
    # создает случайное число элементов
    for i in range(1, r):
      print("working on item creation, it will take: " + str(i) + " seconds")
      time.sleep(i)
      print("acquiring lock...")
      cond.acquire()
      try:
        si.produce(i)
        cond.notify()
      finally:
        cond.release()
      
def consumer(si, cond):
    cond.acquire()
    while True:
      try:
        si.consume()
      except:
        print("No item to consume...")
        # ожидает с максимальным таймаутом 10 сек
        val = cond.wait(10)
        if val:
          print("notification received about item production...")
          continue
        else:
          print("waiting timeout...")
          break
        
    cond.release()
    
if __name__=='__main__':
  # объект условия
  cond = threading.Condition()
  # SomeItem объект
  si = SomeItem()
  # поток производителя
  p = threading.Thread(target=producer, args=(si,cond,))
  p.start()
  # поток потребителя
  c = threading.Thread(target=consumer, args=(si,cond,))
  c.start()

  #print('Waiting for producer and consumer threads...')
  p.join()
  c.join()
  print("Done")

Несколько важных выводов из приведенного выше примера кода:

• Мы создали класс SomeItem, у которого есть список list, который действует как общий ресурс между потоком-производителем и потоком-потребителем.

• Поток-производитель случайным образом генерирует некоторые элементы списка и добавляет их в список.

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

Это очень простой пример, охватывающий все варианты использования объекта условия. Попробуйте запустить указанную выше программу с двумя потоками потребителей и одним потоком производителя.