Потоки в Python — одна из самых обсуждаемых вещей на профильных ресурсах.
Рассмотрим, что такое потоки и почему многопоточность в питоне — бессмысленная трата времени
Многопоточность — нет, мультипроцессорность — да
Есть процесс с общими данными — внутри которого инициализированы потоки. Потоки могут быть созданы главным процессом, но выполняют каждый свою задачу. Python использует внутреннюю глобальную блокировку интерпретатора (Global Interpreter Lock, GIL). Из-за этого в каждый конкретный момент времени может работать только один поток. Вследствие этого программы на Python могут выполняться только на одном процессоре, независимо от их количества в системе. Блокировка GIL делает неэффективной реализацию в виде многопоточного приложения.
Наличие дополнительных процессоров практически не дает преимуществ программе, которая большую часть времени проводит в ожидании событий.
Реализовать многопоточность в питоне, так или иначе, можно, но из-за GIL запуск на нескольких процессоров всё равно не будет осуществлен — все потоки будут работать только на одном процессоре — это не даст никаких преимуществ
Модуль threading
Класс Thread из модуля threading позволяет запускать программу в отдельном потоке. Так выглядит его простейшая реализация:
[code]
from threading import Thread
import time
def clock(interval):
while True:
print(«Текущее время: %s» % time.ctime())
time.sleep(interval)
t = Thread(target=clock, args=(15, ))
t.daemon = True
t.start()
[/code]
Разбираем пример.
1. аrgs — это кортеж, сюда передаем позиционные аргументы функции.
2.daemon — логический флаг, указывающий, будет ли поток демоническим, это означаетпрограмма (процесс) не завершится, пока работает хоть один поток внутри неё. Если назначить поток демоном, то он закроется вместе с закрытием программы. Его нужно прописывать до метода start
3. start — запуск потока на выполнение (внутри target — это название функции, которая будет выполняться)
4. join — указание, что мы будем знать о завершение потока. Без join главная программа будет завершаться сразу после завершения потока. Если мы хотим, чтобы приложение работало после завершения потока — нужно писать join.
Поток в виде класса
Когда мы определяем собственный класс потока, в котором переопределяется метод __init__(), важно вызвать конструктор базового класса super().__init__(), иначе будет ошибка
[code]
from threading import Thread
import time
class ClockThread(Thread): #Создаем класс, которые наследуется от базового класса Thread
def __init__(self, interval):
super().__init__()
self.daemon = True
self.interval = interval
def run(self):
while True:
print(«Текущее время: %s» % time.ctime())
time.sleep(self.interval)
t = ClockThread(3)
t.start()
t.join()
[/code]
Результат:
Разбираем:
1. super().__init__ — мы вызываем конструктор родителя (Thread) базового класса, чтобы использовать все методы
2. start () запускает сам поток, а в run() мы запишем непосредственно блок кода для выполнения
События
Или другой пример выполнения поток по очереди с помощью событий. Событие — простейший объект синхронизации, являющийся, по сути, флажком. Поток может ожидать установки этого флажка, или устанавливать и сбрасывать его самостоятельно.
[code]
import threading
# x = выводимое значение,
# event_for_wait = событие, которое ожидается функция принимает на вход некий параметр, событие, которое ожидают
# event_for_set = событие для которого необходимо установить флаг True
def writer(x, event_for_wait, event_for_set):
for i in range(10):
# ожидаем некое событие
# поток-клиент ожидает установки флага
event_for_wait.wait()
# переводим флаг события на false
# поток-сервер может установить флаг или снять его
event_for_wait.clear()
# выводим параметр
print(x)
# для второго события ставим true — и потоки, которые ждали этого события начинают работать
event_for_set.set()
# Объекты класса Event модуля threading — объекты-наблюдатели
s1 = threading.Event()
s2 = threading.Event()
# Запускаем потоки
# в таргете у нас функция writer, а ей, как известно нужно передать три параметра
# Что мы будем выводить
# Событие на ожидание
# Событие для флага true
flow1 = threading.Thread(target=writer, args=(‘Здесь стартует первый поток’, s1, s2))
flow2 = threading.Thread(target=writer, args=(‘Здесь стартует второй поток’, s2, s1))
# запускаем потоки
flow1.start()
flow2.start()
# устанавливаем значение true для события (s1)
# потоки, которые этого ждут — пробуждаются
s1.set()
# ждем, пока завершатся запущенные потоки
# Ждем, пока поток не закончится. Это блокирует вызывающий поток до тех пор,
# пока поток, чей метод join() вызывается, не завершится
flow1.join()
flow2.join()
[/code]
Результат:
Блокировки
Блокировки — наиболее фундаментальный механизм синхронизации предоставляемый модулем threading. В каждый момент времени блокировка может принадлежать не более чем одному потоку. Если поток пытается получить блокировку, уже принадлежащую другому потоку, выполнение первого потока приостанавливается до тех пор, пока блокировка не будет освобождена.
Если один поток выполняет операции с данными, то второй поток должен быть заблокирован. Блокировка — объект, экземпляр класса Lock(). Есть пример синхронизации потоков при помощи блокировок (мьютексов) показано, как будет работать потоки с блокировкой и без. В идеале — у нас всегда должен быть ноль
[code]
# Как работают потоки с блокировками и без
import threading
lock_yes = 0 # задаем глобальную переменную, которую будем изменять в потоках с блокировками
lock_none = 0 # и без неё
COUNT = 100000 # количество итераций
lock_item = threading.Lock() #создаем экземпляр класса Lock()
# — Работа с блокировками —-
def plus_with_lock():
global lock_yes
for i in range(COUNT):
lock_item.acquire()
lock_yes += 1
lock_item.release()
def minus_with_lock():
global lock_yes
for i in range(COUNT):
lock_item.acquire()
lock_yes -= 1
lock_item.release()
# — Работа без блокировок —
def plus_without_lock():
global lock_none
for i in range(COUNT):
lock_none += 1
def minus_without_lock():
global lock_none
for i in range(COUNT):
lock_none -= 1
if __name__ == «__main__»:
s1 = threading.Thread(target = plus_with_lock)
s2 = threading.Thread(target = minus_with_lock)
s3 = threading.Thread(target = plus_without_lock)
s4 = threading.Thread(target = minus_without_lock)
s1.start()
s2.start()
s3.start()
s4.start()
s1.join()
s2.join()
s3.join()
s4.join()
print(f»Глобальная переменная {lock_yes} при работе с блокировкой»)
print(f»Глобальная переменная {lock_none} при работе без блокировки»)
[/code]
Результат:
Если мы работаем без блокировки — значение переменной меняется случайным образом (да, иногда и даже часто скрипт будет отрабатывать как нужно), но будут случаи, когда происходит как в нашем примере.
Что мы использовали в примере:
1. acquire() — приобрести блокировку
2. release() — освободить блокировку
Семафоры
Еще один механизм синхронизации потоков. Семафор — класс, значение по умолчанию равно нулю, это некий внутренний счетчик. Работает аналогично событиям — уменьшается при каждом вызове метода acquire() и увеличивается при каждом вызове release(). Если счетчик семафора достигает нуля, метод acquire() приостанавливает работу потока, пока другой поток не вызовет release(). Как выглядит синхронизация потоков при помощи семафоров?
[code]
import threading
import time
import random
semaphore = threading.Semaphore(0) # тот самый внутренний счетчик семафора (по умолчанию = 1, мы ставим ноль)
def client():
print(«Продайте мне Вискарь!») # первоначальное сообщение
semaphore.acquire() #здесь у нас семафор — уменьшаем значение счетчика на единицу
# счетчик стал равень нулю, а значит работа потока приостановлена, ждем release()
print(f»Держите свой грязные {item} рублей»)
def vendor():
global item
time.sleep(5) # ждем (типа считаем, ага), чтобы выдать цену покупателю
item = random.randint(0, 1000) #устанавливаем случайную цену на вискарик
print(f»Клади на бар {item} рублей, ковбой»)
semaphore.release() # увеличиваем счетчик на 1 и пробуждаем поток в функции client()
if __name__ == ‘__main__’:
for i in range(5):
t1 = threading.Thread(target=vendor)
t2 = threading.Thread(target=client)
t1.start()
t2.start()
t1.join()
t2.join()
print(«Fin.»)
[/code]
Что использовали в примере:
1. acquire() — приобретает семафор. Если внутренний счетчик имеет значение больше нуля, этот метод уменьшает его на 1 и тут же возвращает управление. Если значение счетчика равно нулю, этот метод приостанавливает работу потока, пока другой поток не вызовет метод release()
2. release() — увеличивает внутренний счетчик семафора на 1. Если перед вызовом метода счетчик был равен нулю и имеется другой поток, ожидающий освобождения семафора, этот поток возобновляет работу
Модуль multiprocessing
С его помощью можно запускать задачи в каждом отдельном процессе. Интерфейс этого модуля имитирует потоки, но процессы не работают с общими данными. Модуль multiprocessing позволяет запускать процессы, которые синхронизируются между собой на нескольких процессорах. Как они взаимодействуют друг с другом? С помощью объектов очереди multiprocessing.JoinableQueue(). Самый простой пример (без очередей)
[code]
import multiprocessing
import time
def clock(interval):
while True:
print(f»The time is {time.ctime()}»)
time.sleep(interval)
if __name__ == «__main__»:
p = multiprocessing.Process(target=clock, args=(1, ))
p.start()
[/code]
Результат:
В следующем примере мы реализуем в виде класса не поток, а подпроцесс (субпроцесс). Наследуемся от базового класса — multiprocessing.Process [code]
import multiprocessing
import time
class Fiveoclock(multiprocessing.Process):
def __init__(self, interval):
super().__init__()
self.interval = interval
def run(self):
while True:
print(f»The time is {time.ctime()}»)
time.sleep(self.interval)
if __name__ == «__main__»:
p = Fiveoclock(1)
p.start()
[/code]
Результат:
В чем плюс использования процесса, а не потока?
Самый главный минус threading, как мы уже сказали в начале статьи — GIL, блокировка интерпритатора. Множество потоков на отдельных ядрах запускаются, но не дают никаких плюсов к производительности. В то время какmultiprocessing — запускает множество процессов на множестве ядер и от этого есть эффективность при высокой нагрузке на проект. Процесс он имеет собственные данные. В процессах изменения локальные — не действуют на другие запущенные процессы.
Синхронизация процессов в модуле multiprocessing. Очереди Queue()
Очереди — механизм для совместного использования данных с помощью уже упомянутых выше объектов очереди — multiprocessing.JoinableQueue(). Очереди позволяют уведомить сервер, что клиенты получили данные, обработали их и ждут новые. Сервер запускается в главном процессе — не подпроцесс, не субпроцесс, как это делается с клиентами.
Сервер должен в цикле перебирать заготовки, обрабатывать их и (put()) поместить обработанную заготовку в очередь. ПОТРЕБИТЕЛЬ — При появлении в очереди объекта, извлечь его, обработать и как-то объявить поставщику, что мы его получили
[code]
import multiprocessing
import time
def client(queue_in):
while True:
print(‘Принимаю данные’)
item = queue_in.get() # Извлекаем элемент из очереди
print(item) # Делаем операции с элементом. В нашем случае, просто выводим
queue_in.task_done() # Сообщаем, что элемент очереди queue_in обработан с помощью метода task_done
time.sleep(4) # Ждем 4 секунды
# функция, в которой сервер добавляет элементы в очередь
def server(sequence, output_q):
for item in sequence:
print(‘Высылаю новые данные’)
output_q.put(item) # добавляем этот элемент в очередь
time.sleep(4) # Ждем 4 секунды
# Настройка
if __name__ == ‘__main__’:
q = multiprocessing.JoinableQueue() # объект очереди
cons_p = multiprocessing.Process(target=client, args=(q,)) # создаем подпроцесс для клиентской функции
cons_p.daemon = True # ставим флаг, что данный процесс является демоническим
cons_p.start() # стартуем процесс
sequence = [1,2,3,4] # последовательность элементов, которая будет передана потребителю. Он получит их по очереди
server(sequence, q) # стартуем сервер
q.join() # ждем пока чтобы клиент успел обработать все элементы
[/code]
Результат:
Что использовали в примере:
1. task_done() — приобретает семафор. Если внутренний счетчик имеет значение больше нуля, этот метод уменьшает его на 1 и тут же возвращает управление. Если значение счетчика равно нулю, этот метод приостанавливает работу потока, пока другой поток не вызовет метод release()
2. put() — добавляет элемент в очередь. Если она заполнена до предела, процесс приостанавливается до появления места.
3. multiprocessing.JoinableQueue() — используется потребителем, чтобы сообщить, что элемент очереди, полученный методом get(), был обработан. Вызывает исключение ValueError, если количество вызовов этого метода превышает число элементов, извлеченных из очереди
