Thread me: Threads ME Store | Call us 800-Threads (8473237)
Многопоточность в Python, функции threading с примерами работы потоков
Предыдущий урок: Приватные переменные
Модуль threading в Python используется для реализации многопоточности в программах. В этом материале разберемся с Thread и разными функциями этого модуля.
Что такое поток?
В информатике поток — это минимальная единица работы, запланированная для выполнения операционной системой.
О потоках нужно знать следующее:
- Они существуют внутри процесса;
- В одном процессе может быть несколько потоков;
- Потоки в одном процессе разделяют состояние и память родительского процесса.
Модуль threading в Python можно представить таким простым примером:
import time
from threading import Threaddef sleepMe(i):
print("Поток %i засыпает на 5 секунд.\n" % i)
time.sleep(5)
print("Поток %i сейчас проснулся.\n" % i)for i in range(10):
th = Thread(target=sleepMe, args=(i, ))
th.start()После запуска скрипта вывод будет следующий:
Поток 0 засыпает на 5 секунд. Поток 3 засыпает на 5 секунд. Поток 1 засыпает на 5 секунд. Поток 4 засыпает на 5 секунд. Поток 2 засыпает на 5 секунд. Поток 5 засыпает на 5 секунд. Поток 6 засыпает на 5 секунд. Поток 7 засыпает на 5 секунд. Поток 8 засыпает на 5 секунд. Поток 9 засыпает на 5 секунд. Поток 0 сейчас проснулся. Поток 3 сейчас проснулся. Поток 1 сейчас проснулся. Поток 4 сейчас проснулся. Поток 2 сейчас проснулся. Поток 5 сейчас проснулся. Поток 6 сейчас проснулся. Поток 7 сейчас проснулся. Поток 8 сейчас проснулся. Поток 9 сейчас проснулся.У вас он может отличаться, потому что у параллельных потоков нет определенного порядка.
Функции threading в Python
Возьмем программу из первого примера и воспользуемся ею для демонстрации разных функций модуля.
threading.active_count()
Эта функция возвращает количество исполняемых на текущий момент потоков. Изменим последнюю программу, чтобы она выглядела вот так:
import time
import threading
from threading import Threaddef sleepMe(i):
print("Поток %i засыпает на 5 секунд." % i)
time.sleep(5)
print("Поток %i сейчас проснулся." % i)for i in range(10):
th = Thread(target=sleepMe, args=(i, ))
th.start()
print("Запущено потоков: %i." % threading.active_count())Подписывайтесь на телеграм каналы
Теперь в выводе будет показываться количество активных на текущий момент потоков:
Поток 0 засыпает на 5 секунд.Запущено потоков: 3. Запущено потоков: 4.Поток 1 засыпает на 5 секунд. Запущено потоков: 5.Поток 2 засыпает на 5 секунд. Поток 3 засыпает на 5 секунд.Запущено потоков: 6. Запущено потоков: 7.Поток 4 засыпает на 5 секунд. Поток 5 засыпает на 5 секунд.Запущено потоков: 8. Поток 6 засыпает на 5 секунд.Запущено потоков: 9. Запущено потоков: 10.Поток 7 засыпает на 5 секунд. Поток 8 засыпает на 5 секунд.Запущено потоков: 11. Поток 9 засыпает на 5 секунд.Запущено потоков: 12. Поток 0 сейчас проснулся. Поток 1 сейчас проснулся. Поток 2 сейчас проснулся. Поток 3 сейчас проснулся. Поток 4 сейчас проснулся. Поток 5 сейчас проснулся. Поток 6 сейчас проснулся. Поток 7 сейчас проснулся. Поток 8 сейчас проснулся. Поток 9 сейчас проснулся.Также обратите внимание, что после запуска всех потоков счетчик показывает число 11, а не 10. Причина в том, что основной поток также учитывается наравне с 10 остальными.
threading.current_thread()
Эта функция возвращает исполняемый прямо сейчас поток. С ее помощью можно выполнять определенные действия с ним. Поменяем все тот же скрипт:
import time
import threading
from threading import Threaddef sleepMe(i):
print("Поток %s засыпает на 5 секунд.\n" % threading.current_thread())
time.sleep(5)
print("Поток %s сейчас проснулся." % threading.current_thread())# Cоздаем только четыре потока
for i in range(10):
th = Thread(target=sleepMe, args=(i, ))
th.start()Теперь вывод будет таким:
Поток <Thread(Thread-4, started 13324)> засыпает на 5 секунд. Поток <Thread(Thread-1, started 15748)> засыпает на 5 секунд. Поток <Thread(Thread-5, started 12164)> засыпает на 5 секунд. Поток <Thread(Thread-2, started 15972)> засыпает на 5 секунд. Поток <Thread(Thread-6, started 13540)> засыпает на 5 секунд. Поток <Thread(Thread-3, started 14396)> засыпает на 5 секунд. Поток <Thread(Thread-7, started 15620)> засыпает на 5 секунд. Поток <Thread(Thread-8, started 7644)> засыпает на 5 секунд. Поток <Thread(Thread-9, started 15424)> засыпает на 5 секунд. Поток <Thread(Thread-10, started 15852)> засыпает на 5 секунд. Поток <Thread(Thread-4, started 13324)> сейчас проснулся. Поток <Thread(Thread-1, started 15748)> сейчас проснулся. Поток <Thread(Thread-5, started 12164)> сейчас проснулся.Поток <Thread(Thread-2, started 15972)> сейчас проснулся. Поток <Thread(Thread-6, started 13540)> сейчас проснулся. Поток <Thread(Thread-3, started 14396)> сейчас проснулся.Поток <Thread(Thread-7, started 15620)> сейчас проснулся. Поток <Thread(Thread-8, started 7644)> сейчас проснулся.Поток <Thread(Thread-9, started 15424)> сейчас проснулся. Поток <Thread(Thread-10, started 15852)> сейчас проснулся.threading.main_thread()
Эта функция возвращает основной поток программы. Именно из него создаются новые потоки.
import threading
print(threading.main_thread())Запустим:
<_MainThread(MainThread, started 10476)>
threading.enumerate()
Эта функция возвращает список всех активных потоков. Пользоваться ею проще простого:
import threading
for thread in threading.enumerate():
print("Имя потока %s." % thread.getName())Запустим скрипт:
Имя потока MainThread.
Имя потока SockThread.Сейчас работает только 2 потока, поэтому в выводе есть только они.
threading.Timer()
Эта функция модуля
threadingиспользуется для создания нового потока и указания времени, через которое он должен запуститься. После запуска поток вызывает определенную функцию. Разберем на примере:
import threadingdef delayed():
print("Вывод через 5 секунд!")thread = threading.Timer(5, delayed)
thread.start()Запустим скрипт:
Вывод через 5 секунд!
Многопоточность в Python
В этом материале были разобраны некоторые функции модуля
threading. Они предоставляют удобные методы для управления потоками в многопоточной среде.Далее: Assert
Python в три ручья: работаем с потоками (часть 1) | GeekBrains
В каких случаях вам нужна многопоточность, как реализовать её на Python и что нужно знать о глобальной блокировке GIL.
https://d2xzmw6cctk25h.cloudfront.net/post/1582/og_cover_image/b8f59e927c66e03053113b8036b95103
Из этой статьи вы узнаете, как с Python выполнять несколько операций одновременно и распределять нагрузку между ядрами процессора, какие особенности языка учитывать. Но главное — поймете, когда многопоточность в Python нужна, а когда только мешает.
Небольшое предупреждение для тех, кто впервые слышит о параллельных вычислениях. Что такое поток и чем он отличается от процесса, мы выяснили в статье «Внутри процесса: многопоточность и пинг-понг mutex’ом». Тогда мы приводили примеры на Java, но теоретические основы многопоточности верны и для Python. Совпадают, в том числе, механизмы синхронизации потоков: семафоры, взаимные исключения (mutex), условия, события. Поэтому сегодня сделаем акцент на особенностях Python, его механизмах и инструментах, связанных с многопоточностью.
Организовать параллельные вычисления в Python без внешних библиотек можно с помощью модулей:
- threading — для управления потоками.
- queue — для организации очередей.
- multiprocessing — для управления процессами.
Пока нас интересует только первый пункт списка.
Как создавать потоки в Python
Метод 1 — «функциональный»
Для работы с потоками из модуля threading импортируем класс Thread. В начале кода пишем:
from threading import Thread
После этого нам будет доступна функция Thread() — с ней легко создавать потоки. Синтаксис такой:
variable = Thread(target=function_name, args=(arg1, arg2,))
Первый параметр target — это «целевая» функция, которая определяет поведение потока и создаётся заранее. Следом идёт список аргументов. Если судьбу аргументов (например, кто будет делимым, а кто делителем в уравнении) определяет их позиция, их записывают как args=(x,y). Если же вам нужны аргументы в виде пар «ключ-значение», используйте запись вида kwargs={‘prop’:120}.
Ради удобства отладки можно также дать новому потоку имя. Для этого среди параметров функции прописывают name=«Имя потока». По умолчанию name хранит значение null. А ещё потоки можно группировать с помощью параметра group, который по умолчанию — None.
За дело! Пусть два потока параллельно выводят каждый в свой файл заданное число строк. Для начала нам понадобится функция, которая выполнит задуманный нами сценарий. Аргументами целевой функции будут число строк и имя текстового файла для записи.
Давайте попробуем:
#coding: UTF-8
from threading import Thread
def prescript(thefile, num):
with open(thefile, 'w') as f:
for i in range(num):
if num > 500:
f.write('МногоБукв\n')
else:
f.write('МалоБукв\n')
thread1 = Thread(target=prescript, args=('f1.txt', 200,))
thread2 = Thread(target=prescript, args=('f2.txt', 1000,))
thread1.start()
thread2.start()
thread1.join()
thread2.join()
Что start() запускает ранее созданный поток, вы уже догадались. Метод join() останавливает поток, когда тот выполнит свои задачи. Ведь нужно закрыть открытые файлы и освободить занятые ресурсы. Это называется «Уходя, гасите свет». Завершать потоки в предсказуемый момент и явно — надёжнее, чем снаружи и неизвестно когда. Меньше риск, что вмешаются случайные факторы. В качестве параметра в скобках можно указать, на сколько секунд блокировать поток перед продолжением его работы.
Метод 2 — «классовый»
Для потока со сложным поведением обычно пишут отдельный класс, который наследуют от Thread из модуля threading. В этом случае программу действий потока прописывают в методе run() созданного класса. Ту же петрушку мы видели и в Java.
#coding: UTF-8
import threading
class MyThread(threading.Thread):
def __init__(self, num):
super().__init__(self, name="threddy" + num)
self.num = num
def run(self):
print ("Thread ", self.num),
thread1 = MyThread("1")
thread2 = MyThread("2")
thread1.start()
thread2.start()
thread1.join()
thread2.join()
Стандартные методы работы с потоками
Чтобы управлять потоками, нужно следить, как они себя ведут. И для этого в threading есть специальные методы:
current_thread() — смотрим, какой поток вызвал функцию;
active_count() — считаем работающие в данный момент экземпляры класса Thread;
enumerate() — получаем список работающих потоков.
Ещё можно управлять потоком через методы класса:
is_alive() — спрашиваем поток: «Жив ещё, курилка?» — получаем true или false;
getName() — узнаём имя потока;
setName(any_name) — даём потоку имя;
У каждого потока, пока он работает, есть уникальный идентификационный номер, который хранится в переменной ident.
thread1.start() print(thread1.ident)
Отсрочить операции в вызываемых потоком функциях можно с помощью таймера. В инициализаторе объектов класса Timer всего два аргумента — время ожидания в секундах и функция, которую нужно в итоге выполнить:
import threading
print ("Waiting...")
def timer_test():
print ("The timer has done its job!")
tim = threading.Timer(5.0, timer_test)
tim.start()
Таймер можно один раз создать, а затем запускать в разных частях кода.
Потусторонние потоки
Обычно Python-приложение не завершается, пока работает хоть один его поток. Но есть особые потоки, которые не мешают закрытию программы и останавливается вместе с ней. Их называют демонами (daemons). Проверить, является ли поток демоном, можно методом isDaemon(). Если является, метод вернёт истину.
Назначить поток демоном можно при создании — через параметр “daemon=True” или аргумент в инициализаторе класса.
thread0 = Thread(target=target_func, kwargs={‘x’:10}, daemon=True)
Не поздно демонизировать и уже существующий поток методом setDaemon(daemonic).
Всё бы ничего, но это даже не верхушка айсберга, потому что прямо сейчас нас ждут великие открытия.
Приключение начинается. У древнего шлюза
Питон слывёт дружелюбным и простым в общении, но есть у него причуды. Нельзя просто взять и воспользоваться всеми преимуществами многопоточности в Python! Дорогу вам преградит огромный шлюз… Даже так — глобальный шлюз (Global Interpreter Lock, он же GIL), который ограничивает многопоточность на уровне интерпретатора. Технически, это один на всех mutex, созданный по умолчанию. Такого нет ни в C, ни в Java.
Задача шлюза — пропускать потоки строго по одному, чтоб не летали наперегонки, как печально известные стритрейсеры, и не создавали угрозу работе интерпретатора.
Без шлюза потоки подрезали бы друг друга, чтобы первыми добраться до памяти, но это еще не всё. Они имеют обыкновение внезапно засыпать за рулём! Операционная система не спрашивает, вовремя или невовремя — просто усыпляет их в ей одной известный момент. Из-за этого неупорядоченные потоки могут неожиданно перехватывать друг у друга инициативу в работе с общими ресурсами.
Дезориентированный спросонок поток, который видит перед собой совсем не ту ситуацию, при которой засыпал, рискует разбиться и повалить интерпретатор, либо попасть в тупиковую ситуацию (deadlock). Например, перед сном Поток 1 начал работу со списком, а после пробуждения не нашёл в этом списке элементов, т.к. их удалил или перезаписал Поток 2.
Чтобы такого не было, GIL в предсказуемый момент (по умолчанию раз в 5 миллисекунд для Python 3.2+) командует отработавшему потоку: «СПАААТЬ!» — тот отключается и не мешает проезжать следующему желающему. Даже если желающего нет, блокировщик всё равно подождёт, прежде чем вернуться к предыдущему активному потоку.
Благодаря шлюзу однопоточные приложения работают быстро, а потоки не конфликтуют. Но, к сожалению, многопоточные программы при таком подходе выполняются медленнее — слишком много времени уходит на регулировку «дорожного движения». А значит обработка графики, расчет математических моделей и поиск по большим массивам данных c GIL идут неприемлемо долго.
В статье «Understanding Python GIL»технический директор компании Gaglers Inc. и разработчик со стажем Chetan Giridhar приводит такой пример:
from datetime import datetime
import threading
def factorial(number):
fact = 1
for n in range(1, number+1):
fact *= n
return fact
number = 100000
thread = threading.Thread(target=factorial, args=(number,))
startTime = datetime.now()
thread.start()
thread.join()
endTime = datetime.now()
print "Время выполнения: ", endTime - startTime
Код вычисляет факториал числа 100 000 и показывает, сколько времени ушло у машины на эту задачу. При тестировании на одном ядре и с одним потоком вычисления заняли 3,4 секунды. Тогда Четан создал и запустил второй поток. Расчет факториала на двух ядрах длился 6,2 секунды. А ведь по логике скорость вычислений не должна была существенно измениться! Повторите этот эксперимент на своей машине и посмотрите, насколько медленнее будет решена задача, если вы добавите thread2. Я получила замедление ровно вдвое.
Глобальный шлюз — наследие времён, когда программисты боролись за достойную реализацию многозадачности и у них не очень получалось. Но зачем он сегодня, когда есть много- и очень многоядерные процессоры? Как объяснил Гвидо ван Россум, без GIL не будут нормально работать C-расширения для Python. Ещё упадёт производительность однопоточных приложений: Python 3 станет медленнее, чем Python 2, а это никому не нужно.
Что делать?
«Нормальные герои всегда идут в обход»
Шлюз можно временно отключить. Для этого интерпретатор Python нужно отвлечь вызовом функции из внешней библиотеки или обращением к операционной системе. Например, шлюз выключится на время сохранения или открытия файла. Помните наш пример с записью строк в файлы? Как только вызванная функция возвратит управление коду Python или интерфейсу Python C API, GIL снова включается.
Как вариант, для параллельных вычислений можно использовать процессы, которые работают изолированно и неподвластны GIL. Но это большая отдельная тема. Сейчас нам важнее найти решение для многопоточности.
Если вы собираетесь использовать Python для сложных научных расчётов, обойти скоростную проблему GIL помогут библиотеки Numba, NumPy, SciPy и др. Опишу некоторые из них в двух словах, чтобы вы поняли, стоит ли разведывать это направление дальше.
Numba для математики
Numba — динамически, «на лету» компилирует Python-код, превращая его в машинный код для исполнения на CPU и GPU. Такая технология компиляции называется JIT — “Just in time”. Она помогает оптимизировать производительность программ за счет ускорения работы циклов и компиляции функций при первом запуске.
Суть в том, что вы ставите аннотации (декораторы) в узких местах кода, где вам нужно ускорить работу функций.
Для математических расчётов библиотеку удобно использовать в связке c NumPy. Допустим, нужно сложить одномерные массивы — элемент за элементом.
def arr_sum (x , y):
result_arr = nupmy.empty_like ( x)
for i in range (len (x)) :
result_arr [i ] = x[i ] + y[i ]
return result_arr
Метод nupmy.empty_like() принимает массив и возвращает (но не инициализирует!) другой — соответствующий исходному по форме и типу. Чтобы ускорить выполнение кода, импортируем класс jit из модуля numba и добавляем в начало кода аннотацию @jit:
from numba import jit @jit def arr_sum(x,y):
Это скромное дополнение способно ускорить выполнение операции более чем в 100 раз! Если интересно, посмотрите замеры скорости математических расчётов при использовании разных библиотек для Python.
PyCUDA и Numba для графики
В графических вычислениях Numba тоже кое-что может. Она умеет работать с программной моделью CUDA, чтобы визуализировать научные данные и работу алгоритмов, выдавать информацию о GPU и др. Подробнее о том, как работают графический процессор и CUDA — здесь. И снова мы встретимся с многопоточностью.
При работе с многомерными массивами в CUDA, чтобы понять, какой поток сейчас работает с элементами массива, нужно отследить, кто и когда вызывает функцию ядра. Например, поток может определять свою позицию в сетке блоков и рассчитать соответствующий элемент массива:
from numba import cuda
@cuda.jit
def call_for_kernel(io_arr):
# Идентификатор потока в одномерном блоке
thread_x = cuda.threadIdx.x
# Идентификатор блока в одномерной сетке
thread_y = cuda.blockIdx.x
# Число потоков на блок (т.е. ширина блока)
block_width = cuda.blockDim.x
# Находим положение в массиве
t_position = thread_x + thread_y * block_width
if t_position < io_arr.size: # Убеждаемся, что не вышли за границы массива
io_arr[ t_position] *= 2 # Считаем
Главный плюс этого кода даже не в скорости исполнения, а в прозрачности и простоте. Снова сошлюсь на Хабр, где есть сравнение скорости GPU-расчетов при использовании Numba, PyCUDA и эталонного С CUDA. Небольшой спойлер: PyCUDA позволяет достичь скорости вычислений, сопоставимой с Cи, а Numba подходит для небольших задач.
Когда многопоточность в Python оправдана
Стоит ли преодолевать связанные c GIL сложности и тратить время на реализацию многопоточности? Вот примеры ситуаций, когда многопоточность несёт с собой больше плюсов, чем минусов.
- Для длительных и несвязанных друг с другом операций ввода-вывода. Например, нужно обрабатывать ворох разрозненных запросов с большой задержкой на ожидание. В режиме «живой очереди» это долго — лучше распараллелить задачу.
- Вычисления занимают более миллисекунды и вы хотите сэкономить время за счёт их параллельного выполнения. Если операции укладываются в 1 мс, многопоточность не оправдает себя из-за высоких накладных расходов.
- Число потоков не превышает количество ядер. В противном случае параллельной работы всех потоков не получается и мы больше теряем, чем выигрываем.
Когда лучше с одним потоком
- При взаимозависимых вычислениях. Считать что-то в одном потоке и передавать для дальнейшей обработки второму — плохая идея. Возникает лишняя зависимость, которая приводит к снижению производительности, а в случае ошибки — к ступору и краху программы.
- При работе через GIL. Это мы уже выяснили выше.
- Когда важна хорошая переносимость на разных устройствах. Правильно подобрать число потоков для машины пользователя — задача не из легких. Если вы пишете под известное вам «железо», всё можно решить тестированием. Если же нет — понадобится дополнительно создавать гибкую систему подстройки под аппаратную часть, что потребует времени и умения.
Анонс — взаимные блокировки в Python
Самое смешное, что по умолчанию GIL защищает только интерпретатор и не предохраняет наш код от взаимных блокировок (deadlock) и других логических ошибок синхронизации. Поэтому разводить потоки по углам, как и в Java, нужно принудительно — с помощью блокирующих механизмов. Об этом и о не упомянутых в статье компонентах модуля threading мы поговорим в следующий раз.
перевод, произношение, транскрипция, примеры использования
Эта нитка слишком толста, чтобы пролезть в игольное ушко. ☰
Хирург соединил края разреза хирургической нитью. ☰
Will you thread the needle for me?
Продень мне, пожалуйста, нитку в иголку. ☰
She knotted the end of the thread.
Она завязала на конце нити узелок. ☰
The waiters threaded between the crowded tables.
Официанты пробирались между близко стоящими столиками. ☰
She had her eyebrows threaded.
У неё были тонкие брови «в ниточку». / Она выщипала себе брови нитью. ☰
Waiters threaded through the crowd.
Официанты протискивались сквозь толпу. ☰
a river that threads through narrow valleys
река, которая, словно нитка, вьётся через узкие долины ☰
She untwisted the ball of thread.
Она распустила этот клубок ниток. ☰
His garment was seamed with silk thread.
Его одежда была сшита шёлковыми нитками. ☰
The melody which threads the first duet.
Мелодия, являющаяся лейтмотивом первого дуэта. ☰
He lost the thread of his argument.
Он потерял нить своих рассуждений. ☰
She unwound some thread from the spool.
Она отмотала от катушки немного ниток . ☰
Chaplin feared that the art of pantomime was under thread.
Чаплин боялся, что искусство пантомимы может вымереть. ☰
A thin glistening thread of moisture ran along the rough concrete sill.
Тонкая поблескивающая струйка жидкости стекала по неровному бетонному подоконнику. ☰
She’s handy with a needle and thread.
Она ловко обращается с иголкой и ниткой. ☰
She threaded her shoelace through the holes.
Она продела шнурок через отверстия /люверсы, блочки/ в ботинках. ☰
The prince’s coat was of black silk laced with gold thread.
Плащ принца был из чёрного шёлка с золотым шитьём. ☰
I’m looking for a needle and thread.
Я ищу иголку и нитку. ☰
She coiled the loose thread around her finger.
Она обмотала свободную нитку вокруг пальца. ☰
the spider’s thread attached to the window sill
паутинка, прикреплённая к подоконнику ☰
He is still in hospital, his life hanging by a thread.
Он всё ещё находится в больнице, и его жизнь висит на волоске. ☰
I’d like to refer to something that was posted in an earlier thread.
Я хотел бы сослаться на одну вещь, которая была опубликована в более ранней ветке (форума). ☰
The thread wouldn’t go through the eye.
Нитка никак не продевалась в игольное ушко. ☰
Tom threaded the rope through the safety harness.
Том продел верёвку сквозь страховочный пояс. ☰
Пример простейшей многопоточной программы на WinAPI
Продолжаем вспоминать Windows API. До сих пор мы писали грустные и унылые однопоточные программки из серии «вызови правильную процедуру с нужными аргументами и посмотри на ее код возврата». Сегодня же мы наконец-то напишем хоть и простенькую, но все же самую что ни на есть многопоточную программу, с теми самыми тредами и мьютексами, которые все постоянно ругают.
Из новых процедур нам понадобятся следующие.
CreateThread(NULL, 0, &ThreadProc, &hMutex, 0, NULL);
Как нетрудно догадаться, CreateThread создает новую нитку. Аргументы слева направо — (1) нечто, о чем сейчас знать не нужно, (2) размер стека в байтах, округляется до размера страницы, если ноль, то берется размер по умолчанию, (3) указатель на процедуру, с которой следует начать выполнение потока, (4) аргумент процедуры, переданной предыдущим аргументом, обычно здесь передается указатель на некую структуру, (5) флаги, например, можно создать приостановленный поток (CREATE_SUSPENDED), а затем запустить его с помощью ResumeThread, (6) куда записать ThreadId созданного потока. В случае успеха процедура возвращает хэндл созданного потока. В случае ошибки возвращается NULL, а подробности можно узнать через GetLastError.
CreateMutex(NULL, FALSE, NULL);
Процедура CreateMutex создает новый мьютекс. О первом аргументе сейчас знать не нужно. Если второй аргумент равен TRUE, созданный мьютекс будет сразу залочен текущим потоком, если же второй аргумент равен FALSE, создается разлоченный мьютекс. Третий аргумент задает имя мьютекса, если мы хотим создать именованный мьютекс. Именованные мьютексы нам пока что не понадобятся, поэтому передаем NULL. Возвращаемые значения в точности такие же, как и у CreateThread.
WaitForSingleObject(hHandle, dwTimeout);
WaitForSingleObject ждет, когда объект, хэндл которого был передан первым аргументом, перейдет в сигнальное состояние (signaled state). Кроме того, в зависимости от типа объекта, процедура может менять его состояние. WaitForSingleObject может быть применен к мьютексам, семафорам, потокам, процессам и не только. Если hHandle представляет собой хэндл мьютекса, процедура ждет, когда мьютекс освободится, а затем лочит его. Если же hHandle является хэндлом потока, то процедура просто ждет его завершения. Второй аргумент задает время ожидания в миллисекундах. Можно ждать вечно, передав специальное значение INFINITE. Если указать ноль, процедура не переходит в режим ожидания, а возвращает управление немедленно.
В случае ошибки процедура возвращает WAIT_FAILED, а подробности поможет узнать GetLastError. В случае успеха возвращается WAIT_OBJECT_0, если мы дождались перехода объекта в сигнальное состояние, и WAIT_TIMEOUT, если отвалились по таймауту. Также мы можем получить WAIT_ABANDONED. Это происходит в случае, если нить, державшая мьютекс, завершилась, не освободив его. В этом случае мьютекс становится залочен текущей нитью, но целостность данных, доступ к которым ограничивался мьютексом, по понятным причинам находится под вопросом.
WaitForMultipleObjects(dwCount, lpHandles, bWaitAll, dwTimeout);
Процедура WaitForMultipleObjects работает аналогично WaitForSingleObject, но для массива объектов. Первый аргумент задает размер массива, должен быть строго больше нуля и не превышать MAXIMUM_WAIT_OBJECTS (который равняется в точности 64). Вторым аргументом задается указатель на массив хэндлов. В массиве могут содержаться хэндлы на объекты разных типов, но многократное включение одного и того же хэндла считается ошибкой. Если третий аргумент равен TRUE, процедура ждет перехода в сигнальное состояние всех объектов, иначе — любого из указанных объектов. Семантика последнего аргумента точно такая же, как и в случае с WaitForSingleObject.
Процедура возвращает WAIT_FAILED в случае ошибки и WAIT_TIMEOUT в случае отваливания по таймауту. Если процедура вернула значение от WAIT_OBJECT_0 до WAIT_OBJECT_0 + dwCount — 1, то в случае bWaitAll == TRUE все объекты из массива перешли в сигнальное состояние, а в случае bWaitAll == FALSE в сигнальное состояние перешел объект с индексом код возврата минус WAIT_OBJECT_0. Если процедура вернула значение от WAIT_ABANDONED_0 до WAIT_ABANDONED_0 + dwCount — 1, то в случае bWaitAll == TRUE все ожидаемые объекты перешли в сигнальное состояние и по крайней мере один из объектов оказался брошенным мьютексом, а в случае bWaitAll == FALSE брошенным оказался мьютекс с индексом код возврата минус WAIT_ABANDONED_0.
Если bWaitAll == TRUE, процедура не меняет состояния объектов до тех пор, пока все они не перейдут в сигнальное состояние. Таким образом, пока вы ждете, мьютексы могут продолжать лочиться и анлочиться другими потоками. Если bWaitAll == FALSE и сразу несколько объектов перешли в сигнальное состояние, процедура всегда работает с объектом в массиве, имеющим минимальный индекс.
ReleaseMutex анлочит ранее залоченный мьютекс. В случае успеха процедура возвращает значение, отличное от нуля. В случае ошибки возвращается ноль, а подробности доступны через GetLastError. Что характерно, для предотвращения дэдлоков Windows позволяет потокам лочить один и тот же мьютекс несколько раз, но и количество вызовов ReleaseMutex должно быть соответствующим.
Процедура Sleep блокирует текущий поток на заданное количество миллисекунд. Если в качестве параметра передать INFINITE, поток будет заблокирован навсегда. Если передать ноль, поток уступает оставшуюся часть своей доли процессорного времени любой другой нити с таким же приоритетом. У процедуры нет возвращаемого значения (VOID).
ExitThread завершает текущий поток с заданным кодом возврата. Объект потока при этом переходит в сигнальное состояние. Если это был последний поток в текущем процессе, процесс завершается. Код возврата конкретного потока может быть получен с помощью GetExitCodeThread.
А теперь посмотрим на все это хозяйство в действии:
#include <windows.h>
#define THREADS_NUMBER 10
#define ITERATIONS_NUMBER 100
#define PAUSE 10 /* ms */
DWORD dwCounter = 0;
DWORD WINAPI ThreadProc(CONST LPVOID lpParam) {
CONST HANDLE hMutex = (CONST HANDLE)lpParam;
DWORD i;
for(i = 0; i < ITERATIONS_NUMBER; i++) {
WaitForSingleObject(hMutex, INFINITE);
dwCounter++;
ReleaseMutex(hMutex);
Sleep(PAUSE);
}
ExitThread(0);
}
VOID Error(CONST HANDLE hStdOut, CONST LPCWSTR szMessage) {
DWORD dwTemp;
TCHAR szError[256];
WriteConsole(hStdOut, szMessage, lstrlen(szMessage), &dwTemp, NULL);
wsprintf(szError, TEXT(«LastError = %d\r\n»), GetLastError());
WriteConsole(hStdOut, szError, lstrlen(szError), &dwTemp, NULL);
ExitProcess(0);
}
INT main() {
TCHAR szMessage[256];
DWORD dwTemp, i;
HANDLE hThreads[THREADS_NUMBER];
CONST HANDLE hStdOut = GetStdHandle(STD_OUTPUT_HANDLE);
CONST HANDLE hMutex = CreateMutex(NULL, FALSE, NULL);
if(NULL == hMutex) {
Error(hStdOut, TEXT(«Failed to create mutex.\r\n»));
}
for(i = 0; i < THREADS_NUMBER; i++) {
hThreads[i] = CreateThread(NULL, 0, &ThreadProc, hMutex, 0, NULL);
if(NULL == hThreads[i]) {
Error(hStdOut, TEXT(«Failed to create thread.\r\n»));
}
}
WaitForMultipleObjects(THREADS_NUMBER, hThreads, TRUE, INFINITE);
wsprintf(szMessage, TEXT(«Counter = %d\r\n»), dwCounter);
WriteConsole(hStdOut, szMessage, lstrlen(szMessage), &dwTemp, NULL);
for(i = 0; i < THREADS_NUMBER; i++) {
CloseHandle(hThreads[i]);
}
CloseHandle(hMutex);
ExitProcess(0);
}
Здесь мы создаем десять потоков, каждый из которых увеличивает глобальный счетчик на единицу сто раз. Чтобы в счетчике не оказалась записана фигня, доступ к нему ограничивается с помощью мьютекса. Перед вызовом ExitProcess закрываются хэндлы мьютекса и всех созданных потоков. Строго говоря, в данном конкретном случае это не требуется, но в более общем случае все хэндлы за собой, разумеется, нужно вовремя закрывать. Также вы могли обратить внимание, что я прислушался к совету, который мне дали в комментариях к предыдущей заметке, и стал оборачивать строки в макрос TEXT.
Как обычно, программа может быть скомпилирована как в Visual Studio и запущена под Windows, так и с помощью MinGW и запущена под Wine. Ну и по традиции напоминаю, что ничто не делает блогера более грустным котиком, чем отсутствие каких-либо комментариев к его постам 🙂
Дополнение: Кстати, при помощи именованных мьютексов можно сделать так, чтобы пользователь мог запустить только один экземпляр приложения. Здесь можно посмотреть пример как это делается.
Дополнение: Также вас могут заинтересовать статьи Многопоточность в C/C++ с использованием pthreads и Написание многопоточных приложений на C++11 и старше.
Метки: C/C++, WinAPI, Параллелизм и многопоточность.
Многопоточность на примерах — модуль threading
Python включает в себя ряд разных параллельных конструкций, таких как как threading, queues и multiprocessing. Модуль threading использовался как главный способ достижения параллельности. Несколько лет назад, модуль multiprocessing был добавлен в пакет стандартных библиотек Python. В этой статье мы сфокусируемся на том, как использовать очереди и потоки (queues и threads).
Использование потоков
Мы начнем с простого примера, который демонстрирует работу потоков. Мы наследуем класс Thread в класс MyThread и укажем, чтобы его имя выводилось как stdout. Попробуем!
# -*- coding: utf-8 -*-
import random
import time
from threading import Thread
class MyThread(Thread):
«»»
A threading example
«»»
def __init__(self, name):
«»»Инициализация потока»»»
Thread.__init__(self)
self.name = name
def run(self):
«»»Запуск потока»»»
amount = random.randint(3, 15)
time.sleep(amount)
msg = «%s is running» % self.name
print(msg)
def create_threads():
«»»
Создаем группу потоков
«»»
for i in range(5):
name = «Thread #%s» % (i+1)
my_thread = MyThread(name)
my_thread.start()
if __name__ == «__main__»:
create_threads()
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35
| # -*- coding: utf-8 -*- import random import time from threading import Thread
class MyThread(Thread): «»» A threading example «»»
def __init__(self, name): «»»Инициализация потока»»» Thread.__init__(self) self.name = name
def run(self): «»»Запуск потока»»» amount = random.randint(3, 15) time.sleep(amount) msg = «%s is running» % self.name print(msg)
def create_threads(): «»» Создаем группу потоков «»» for i in range(5): name = «Thread #%s» % (i+1) my_thread = MyThread(name) my_thread.start()
if __name__ == «__main__»: create_threads() |
В этом коде мы импортировали модули random и time, также мы импортировали класс Thread из модуля threading Python. Далее, мы наследуем класс Thread, и переопределили его метод __init__ для принятия аргумента, под названием name. Для начала потока, вам нужно вызывать метод start().
После запуска потока, он автоматически вызовет метод run. Мы переопределили метод run таким образом, чтобы он выбирал случайный отсчет времени для «сна». Пример random.randint указывает Python выбрать случайное число от 3 до 15. После этого мы указываем потоку «спать» столько секунд, сколько было выбрано случайным способом, для симуляции его настоящей работы. Далее мы ввели имя потока, чтобы сказать пользователю, что он закончился. Функция create_threads создаст 5 потоков, дав каждому из них уникальное имя. Если вы запустите данный код, вы увидите что-то вроде этого:
Thread #2 is running
Thread #3 is running
Thread #1 is running
Thread #4 is running
Thread #5 is running
|
| Thread #2 is running Thread #3 is running Thread #1 is running Thread #4 is running Thread #5 is running |
Порядок выхода каждый раз будет разным. Попробуйте запустить код несколько раз, чтобы увидеть смену порядка. Теперь давайте напишем что-нибудь более практичное!
Написание потокового загрузчика
Предыдущий пример был не слишком полезным в качестве инструмента, показывающего, как именно работают Python потоки. Так что в данном примере, мы создадим класс Thread, который скачивает параллельно файлы из интернета. Мы воспользуемся бесплатным ресурсом в нашем демо. Посмотрим на код:
# -*- coding: utf-8 -*-
# Python 2.7 версии
import os
import urllib2
from threading import Thread
class DownloadThread(Thread):
«»»
Пример многопоточной загрузки файлов из интернета
«»»
def __init__(self, url, name):
«»»Инициализация потока»»»
Thread.__init__(self)
self.name = name
self.url = url
def run(self):
«»»Запуск потока»»»
handle = urllib2.urlopen(self.url)
fname = os.path.basename(self.url)
with open(fname, «wb») as f_handler:
while True:
chunk = handle.read(1024)
if not chunk:
break
f_handler.write(chunk)
msg = «%s закончил загрузку %s!» % (self.name, self.url)
print(msg)
def main(urls):
«»»
Run the program
«»»
for item, url in enumerate(urls):
name = «Поток %s» % (item+1)
thread = DownloadThread(url, name)
thread.start()
if __name__ == «__main__»:
urls = [«http://www.irs.gov/pub/irs-pdf/f1040.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040a.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040ez.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040es.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040sb.pdf»]
main(urls)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51
| # -*- coding: utf-8 -*- # Python 2.7 версии
import os import urllib2 from threading import Thread
class DownloadThread(Thread): «»» Пример многопоточной загрузки файлов из интернета «»»
def __init__(self, url, name): «»»Инициализация потока»»» Thread.__init__(self) self.name = name self.url = url
def run(self): «»»Запуск потока»»» handle = urllib2.urlopen(self.url) fname = os.path.basename(self.url)
with open(fname, «wb») as f_handler: while True: chunk = handle.read(1024) if not chunk: break f_handler.write(chunk)
msg = «%s закончил загрузку %s!» % (self.name, self.url) print(msg)
def main(urls): «»» Run the program «»» for item, url in enumerate(urls): name = «Поток %s» % (item+1) thread = DownloadThread(url, name) thread.start()
if __name__ == «__main__»: urls = [«http://www.irs.gov/pub/irs-pdf/f1040.pdf», «http://www.irs.gov/pub/irs-pdf/f1040a.pdf», «http://www.irs.gov/pub/irs-pdf/f1040ez.pdf», «http://www.irs.gov/pub/irs-pdf/f1040es.pdf», «http://www.irs.gov/pub/irs-pdf/f1040sb.pdf»]
main(urls) |
Это, в общем, полностью переписанный первый скрипт. Здесь мы импортировали наши модули os, urllib2, и threading python. Мы используем urllib2 для непосредственной загрузки в класс потока. Модуль os мы используем для извлечения имени файла, который мы загружаем, так что мы можем использовать его для создания файла с таким же названием на нашем компьютере. В классе DownloadThread мы настраиваем __init__ для принятия url и наименований для потока. В методе run, мы открываем url, извлекаем название файла, после чего используем это название для того, чтобы создать файл на диске.
Есть вопросы по Python?
На нашем форуме вы можете задать любой вопрос и получить ответ от всего нашего сообщества!
Telegram Чат & Канал
Вступите в наш дружный чат по Python и начните общение с единомышленниками! Станьте частью большого сообщества!
Паблик VK
Одно из самых больших сообществ по Python в социальной сети ВК. Видео уроки и книги для вас!
После этого мы используем цикл для загрузки файла по килобайту за раз, и сохранять его на диск. После того, как сохранение файла завершится, мы выводим название потока и тот url, который загрузился. В Python 3 этот код немного отличаться. Нам нужно импортировать urllib вместо urllib2 и использовать urllib.request.urlopen вместо urllib2.urlopen. Вот код, в котором вы можете увидеть разницу:
# -*- coding: utf-8 -*-
# Python 3 версии
import os
import urllib.request
from threading import Thread
class DownloadThread(Thread):
«»»
Пример скачивание файла используя многопоточность
«»»
def __init__(self, url, name):
«»»Инициализация потока»»»
Thread.__init__(self)
self.name = name
self.url = url
def run(self):
«»»Запуск потока»»»
handle = urllib.request.urlopen(self.url)
fname = os.path.basename(self.url)
with open(fname, «wb») as f_handler:
while True:
chunk = handle.read(1024)
if not chunk:
break
f_handler.write(chunk)
msg = «%s закончил загрузку %s!» % (self.name, self.url)
print(msg)
def main(urls):
«»»
Запускаем программу
«»»
for item, url in enumerate(urls):
name = «Поток %s» % (item+1)
thread = DownloadThread(url, name)
thread.start()
if __name__ == «__main__»:
urls = [«http://www.irs.gov/pub/irs-pdf/f1040.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040a.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040ez.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040es.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040sb.pdf»]
main(urls)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51
| # -*- coding: utf-8 -*- # Python 3 версии
import os import urllib.request from threading import Thread
class DownloadThread(Thread): «»» Пример скачивание файла используя многопоточность «»»
def __init__(self, url, name): «»»Инициализация потока»»» Thread.__init__(self) self.name = name self.url = url
def run(self): «»»Запуск потока»»» handle = urllib.request.urlopen(self.url) fname = os.path.basename(self.url)
with open(fname, «wb») as f_handler: while True: chunk = handle.read(1024) if not chunk: break f_handler.write(chunk)
msg = «%s закончил загрузку %s!» % (self.name, self.url) print(msg)
def main(urls): «»» Запускаем программу «»» for item, url in enumerate(urls): name = «Поток %s» % (item+1) thread = DownloadThread(url, name) thread.start()
if __name__ == «__main__»: urls = [«http://www.irs.gov/pub/irs-pdf/f1040.pdf», «http://www.irs.gov/pub/irs-pdf/f1040a.pdf», «http://www.irs.gov/pub/irs-pdf/f1040ez.pdf», «http://www.irs.gov/pub/irs-pdf/f1040es.pdf», «http://www.irs.gov/pub/irs-pdf/f1040sb.pdf»]
main(urls) |
Использование Queues
Очередь(Queues Python) может быть использована для стековых реализаций «пришел первым – ушел первым» (first-in-first-out (FIFO)) или же «пришел последним – ушел последним» (last-in-last-out (LILO)) , если вы используете их правильно.
В данном разделе, мы смешаем потоки и создадим простой скрипт файлового загрузчика, чтобы продемонстрировать, как работает Queues Python со случаями, которые мы хотим паралеллизировать. Чтобы помочь объяснить, как работает Queues, мы перепишем загрузочный скрипт из предыдущей секции для использования Queues. Приступим!
# -*- coding: utf-8 -*-
import os
import threading
import urllib.request
from queue import Queue
class Downloader(threading.Thread):
«»»Потоковый загрузчик файлов»»»
def __init__(self, queue):
«»»Инициализация потока»»»
threading.Thread.__init__(self)
self.queue = queue
def run(self):
«»»Запуск потока»»»
while True:
# Получаем url из очереди
url = self.queue.get()
# Скачиваем файл
self.download_file(url)
# Отправляем сигнал о том, что задача завершена
self.queue.task_done()
def download_file(self, url):
«»»Скачиваем файл»»»
handle = urllib.request.urlopen(url)
fname = os.path.basename(url)
with open(fname, «wb») as f:
while True:
chunk = handle.read(1024)
if not chunk:
break
f.write(chunk)
def main(urls):
«»»
Запускаем программу
«»»
queue = Queue()
# Запускаем потом и очередь
for i in range(5):
t = Downloader(queue)
t.setDaemon(True)
t.start()
# Даем очереди нужные нам ссылки для скачивания
for url in urls:
queue.put(url)
# Ждем завершения работы очереди
queue.join()
if __name__ == «__main__»:
urls = [«http://www.irs.gov/pub/irs-pdf/f1040.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040a.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040ez.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040es.pdf»,
«http://www.irs.gov/pub/irs-pdf/f1040sb.pdf»]
main(urls)
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67
| # -*- coding: utf-8 -*-
import os import threading import urllib.request from queue import Queue
class Downloader(threading.Thread): «»»Потоковый загрузчик файлов»»»
def __init__(self, queue): «»»Инициализация потока»»» threading.Thread.__init__(self) self.queue = queue
def run(self): «»»Запуск потока»»» while True: # Получаем url из очереди url = self.queue.get()
# Скачиваем файл self.download_file(url)
# Отправляем сигнал о том, что задача завершена self.queue.task_done()
def download_file(self, url): «»»Скачиваем файл»»» handle = urllib.request.urlopen(url) fname = os.path.basename(url)
with open(fname, «wb») as f: while True: chunk = handle.read(1024) if not chunk: break f.write(chunk)
def main(urls): «»» Запускаем программу «»» queue = Queue()
# Запускаем потом и очередь for i in range(5): t = Downloader(queue) t.setDaemon(True) t.start()
# Даем очереди нужные нам ссылки для скачивания for url in urls: queue.put(url)
# Ждем завершения работы очереди queue.join()
if __name__ == «__main__»: urls = [«http://www.irs.gov/pub/irs-pdf/f1040.pdf», «http://www.irs.gov/pub/irs-pdf/f1040a.pdf», «http://www.irs.gov/pub/irs-pdf/f1040ez.pdf», «http://www.irs.gov/pub/irs-pdf/f1040es.pdf», «http://www.irs.gov/pub/irs-pdf/f1040sb.pdf»]
main(urls) |
Давайте притормозим. В первую очередь, нам нужно взглянуть на определение главной функции для того, чтобы увидеть, как все протекает. Здесь мы видим, что она принимает список url адресов. Далее, функция main создаете экземпляр очереди, которая передана пяти демонизированным потокам. Основная разница между демонизированным и недемонизированным потоком в том, что вам нужно отслеживать недемонизированные потоки и закрывать их вручную, в то время как поток «демон» нужно только запустить и забыть о нем. Когда ваше приложение закроется, закроется и поток. Далее мы загрузили очередь (при помощи метода put) вместе с переданными url. Наконец, мы указываем очереди подождать, пока потоки выполнят свои процессы через метод join. В классе download у нас есть строчка self.queue.get(), которая выполняет функцию блока, пока очередь делает что-либо для возврата. Это значит, что потоки скромно будут дожидаться своей очереди. Также это значит, чтобы поток получал что-нибудь из очереди, он должен вызывать метод очереди под названием get. Таким образом, добавляя что-нибудь в очередь, пул потоков, поднимет или возьмет эти объекты и обработает их. Это также известно как dequeing. После того, как все объекты в очередь обработаны, скрипт заканчивается и закрывается. На моем компьютере были загружены первые 5 документов за секунду.
Подведем итоги
Теперь вы знаете, как использовать потоки и очереди, как в теории, так и на практике. Потоки особенно полезны, когда вы создаете пользовательский интерфейс и вам нужно поддерживать этот интерфейс юзабельным. Без потоков, пользовательский интерфейс может перестать реагировать и виснуть, пока вы загружаете большой файл, или создаете большой запрос к базе данных. Во избежание этого, вам нужно выполнять длительные процессы в потоках, и связать их с вашим интерфейсом.
Модуль threading на примерах
Модуль threading впервые был представлен в Python 1.5.2 как продолжение низкоуровневого модуля потоков. Модуль threading значительно упрощает работу с потоками и позволяет программировать запуск нескольких операций одновременно. Обратите внимание на то, что потоки в Python лучше всего работают с операциями I/O, такими как загрузка ресурсов из интернета или чтение файлов и папок на вашем компьютере.
Если вам нужно сделать что-то, для чего нужен интенсивный CPU, тогда вам, возможно, захочется взглянуть на модуль multiprocessing, вместо threading. Причина заключается в том, что Python содержит Global Interpreter Lock (GIL), который запускает все потоки внутри главного потока. По этой причине, когда вам нужно запустить несколько интенсивных операций с потоками, вы заметите, что все работает достаточно медленно. Так что мы сфокусируемся на том, в чем потоки являются лучшими: операции I/O.
Небольшое интро
Поток позволяет вам запустить часть длинного кода так, как если бы он был отдельной программой. Это своего рода вызов наследуемого процесса, за исключением того, что вы вызываете функцию или класс, вместо отдельной программы. Я всегда находил конкретные примеры крайне полезными. Давайте взглянем на нечто совершенно простое:
import threading
def doubler(number):
"""
A function that can be used by a thread
"""
print(threading.currentThread().getName() + '\n')
print(number * 2)
print()
if __name__ == '__main__':
for i in range(5):
my_thread = threading.Thread(target=doubler, args=(i,))
my_thread.start()Здесь мы импортируем модуль threading и создаем обычную функцию под названием doubler. Наша функция принимает значение и удваивает его. Также она выводи название потока, который вызывает функцию и выводит бланк-строчку в конце. Далее, в последнем блоке кода, мы создаем пять потоков, и запускаем каждый из них по очереди.
Вам больше не придется долгими часами настраивать дорогостоящую рекламу в Фейсбуке и тратить на нее весомый бюджет. Сайт Avi1 готов помочь Вам купить ее по самым приятным ценам, и как можно быстрее развиться в данной социальной сети. Здесь Вам будут доступны пакетные предложения с разнообразными ресурсами: лайки, подписчики, друзья и пр.
Используя многопоточность можно решить много рутинных моментов. Например загрузка видео или другого материала в социальные сети, такие как Youtube или Facebook. Для развития своего Youtube канала можно использовать https://publbox.com/ru/youtube который возьмет на себя администрирование вашего канала. Youtube отличный источник заработка и чем больше каналов тем лучше. Без Publbox вам не обойтись.
Обратите внимание на то, что когда мы определяем поток, мы устанавливаем его целью на нашу функцию doubler, и мы также передаем аргумент функции. Причина, по которой параметр args выглядит немного непривычно, заключается в том, что нам нужно передать sequence функции doubler, и она принимает только один аргумент, так что нужно добавить запятую в конце, чтобы создать sequence одной из них. Обратите внимание на то, что если вы хотите подождать, пока поток определится, вы можете вызвать его метод join(). Когда вы запустите этот код, вы получите следующую выдачу:
Thread-1
0
Thread-2
2
Thread-3
4
Thread-4
6
Thread-5
8Конечно, вам скорее всего не захочется выводить вашу выдачу в stdout. Это может закончиться сильным беспорядком. Вместо этого, вам нужно использовать модуль Python под названием logging. Это защищенный от потоков модуль и он прекрасно выполняет свою работу. Давайте немного обновим указанный ранее пример и добавим модуль logging, и заодно назовем наши потоки:
import logging
import threading
def get_logger():
logger = logging.getLogger("threading_example")
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler("threading.log")
fmt = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s'
formatter = logging.Formatter(fmt)
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
def doubler(number, logger):
"""
A function that can be used by a thread
"""
logger.debug('doubler function executing')
result = number * 2
logger.debug('doubler function ended with: {}'.format(result))
if __name__ == '__main__':
logger = get_logger()
thread_names = ['Mike', 'George', 'Wanda', 'Dingbat', 'Nina']
for i in range(5):
my_thread = threading.Thread(
target=doubler, name=thread_names[i], args=(i,logger))
my_thread.start()Самое большое изменение в этом коде – это добавление функции get_logger. Эта часть кода создаст логгер, который настроен на дебаг. Это сохранит log в нынешнюю рабочую папку (другими словами, туда, откуда запускается скрипт) и затем мы настраиваем формат каждой линии на логированный. Формат включает временной штамп, название потока, уровень логгирования и логгированое сообщение. В функции doubler мы меняем наши операторы вывода на операторы логгирования.
Обратите внимание на то, что мы передаем логгер в функцию doubler, когда создаем поток. Это связанно с тем, что если вы определяет объект логгирования в каждом потоке, вы получите несколько singletons и ваш журнал будет содержать множество повторяющихся строк. В конце мы называем наши потоки, создав список наименований, и затем устанавливаем каждый поток на особое наименование, использую параметр name. Когда вы запустите этот код, вы должны получить лог файл со следующим содержимым:
Эта выдача достаточно понятная, так что давайте пойдем дальше. Я хочу разобрать еще один вопрос в этой статье. Мы поговорим о наследовании класса под названием threading.Thread. Давайте снова рассмотрим предыдущий пример, только вместо вызова потока напрямую, мы создадим свой собственный подкласс. Вот обновленный код:
import logging
import threading
class MyThread(threading.Thread):
def __init__(self, number, logger):
threading.Thread.__init__(self)
self.number = number
self.logger = logger
def run(self):
"""
Run the thread
"""
logger.debug('Calling doubler')
doubler(self.number, self.logger)
def get_logger():
logger = logging.getLogger("threading_example")
logger.setLevel(logging.DEBUG)
fh = logging.FileHandler("threading_class.log")
fmt = '%(asctime)s - %(threadName)s - %(levelname)s - %(message)s'
formatter = logging.Formatter(fmt)
fh.setFormatter(formatter)
logger.addHandler(fh)
return logger
def doubler(number, logger):
"""
A function that can be used by a thread
"""
logger.debug('doubler function executing')
result = number * 2
logger.debug('doubler function ended with: {}'.format(result))
if __name__ == '__main__':
logger = get_logger()
thread_names = ['Mike', 'George', 'Wanda', 'Dingbat', 'Nina']
for i in range(5):
thread = MyThread(i, logger)
thread.setName(thread_names[i])
thread.start()В этом примере мы только что унаследовали класс threading.Thread. Мы передали число, которое хотим удвоить, а также передали объект логгированмя, как делали это ранее. Но на этот раз, мы настроим название потока по-другому, вызвав функцию setName в объекте потока. Нам все еще нужно вызвать старт в каждом потоке, но запомните, что нам не нужно определять это в наследуемом классе. Когда вы вызываете старт, он запускает ваш поток, вызывая метод run. В нашем классе мы вызываем функцию doubler для выполнения наших вычислений. Выдача сильно похожа на ту, что была в примере ранее, за исключением того, что я добавил дополнительную строку в выдаче. Попробуйте сами и посмотрите, что получится.
Замки и Синхронизация
Когда у вас в распоряжении более одного потока, тогда вам, возможно, понадобится понять, как избежать конфликтов. Под этим я имею ввиду то, что вы можете использовать случай, где более одного потока нуждаются в доступе к одном и тому же ресурсу в одно и то же время. Если вы не думаете о таких проблемах и соответственном планировании, тогда вы можете столкнуться с самыми худшими проблемами в крайне неудобное время, и, как правило, в момент выпуска кода.
Решение проблемы – это использовать замки. Замок предоставлен модулем Python threading и может держать один поток, или не держать поток вообще. Если поток пытается acquire замок на ресурсе, который уже закрыт, этот поток будет ожидать до тех пор, пока замок не откроется. Давайте посмотрим на практичный пример одного кода, который не имеет никакого замочного функционала, но мы попробуем его добавить:
import threading
total = 0
def update_total(amount):
"""
Updates the total by the given amount
"""
global total
total += amount
print (total)
if __name__ == '__main__':
for i in range(10):
my_thread = threading.Thread(target=update_total, args=(5,))
my_thread.start()Мы можем сделать этот пример еще интереснее, добавив вызов time.sleep. Следовательно, проблема здесь в том, что один поток может вызывать update_total и перед тем, как он обновится, другой поток может вызвать его и тоже попытается обновить его. В зависимости от порядка операций, значение может быть добавлено единожды. Давайте добавим замок к функции. Существует два способа сделать эта. Первый – это использование try/finally, если мы хотим убедиться, что замок снят. Вот пример:
import threading
total = 0
lock = threading.Lock()
def update_total(amount):
"""
Updates the total by the given amount
"""
global total
lock.acquire()
try:
total += amount
finally:
lock.release()
print (total)
if __name__ == '__main__':
for i in range(10):
my_thread = threading.Thread(target=update_total, args=(5,))
my_thread.start()Здесь мы просто вешаем замок, перед тем как сделать что-либо другое. Далее, мы пытаемся обновить total и finally, мы снимаем замок и выводим нынешний total. Мы можем упростить данную задачу, используя оператор Python под названием with:
import threading
total = 0
lock = threading.Lock()
def update_total(amount):
"""
Updates the total by the given amount
"""
global total
with lock:
total += amount
print (total)
if __name__ == '__main__':
for i in range(10):
my_thread = threading.Thread(target=update_total, args=(5,))
my_thread.start()Как вы видите, нам больше не нужны try/finally, так как контекстный менеджер, предоставленный оператором with, сделал все это за нас. Конечно, вы можете обнаружить, что пишите код там, где необходимы несколько потоков с доступом к нескольким функциям. Когда вы впервые начнете писать конкурентный код, вы можете сделать что-нибудь на подобии следующего:
import threading
total = 0
lock = threading.Lock()
def do_something():
lock.acquire()
try:
print('Lock acquired in the do_something function')
finally:
lock.release()
print('Lock released in the do_something function')
return "Done doing something"
def do_something_else():
lock.acquire()
try:
print('Lock acquired in the do_something_else function')
finally:
lock.release()
print('Lock released in the do_something_else function')
return "Finished something else"
if __name__ == '__main__':
result_one = do_something()
result_two = do_something_else()Этот код хорошо работает в данном случае, но подразумевается, что у вас есть несколько потоков, вызывающих обе эти функции. Пока один поток работает над функциями, второй может, в свою очередь, обновлять данные и вы получите некорректный результат. Проблема в том, что вы сначала можете не заметить, что с результатами что-то неладное. Как найти решение этой проблеме? Давайте в ней разберемся. Первое, к чему можно прийти, это повесить замок на двух вызовах функций. Давайте попробуем обновить указанный ранее пример, чтобы получить что-то вроде следующего:
import threading
total = 0
lock = threading.RLock()
def do_something():
with lock:
print('Lock acquired in the do_something function')
print('Lock released in the do_something function')
return "Done doing something"
def do_something_else():
with lock:
print('Lock acquired in the do_something_else function')
print('Lock released in the do_something_else function')
return "Finished something else"
def main():
with lock:
result_one = do_something()
result_two = do_something_else()
print (result_one)
print (result_two)
if __name__ == '__main__':
main()Когда вы запустите этот код, вы увидите, что он просто висит. Причина в том, что мы просто указываем модулю threading повесить замок. Так что когда мы вызываем первую функцию, она видит, что замок уже висит и блокируется. Это будет длиться до тех пор, пока замок не снимут, что никогда и не случится, так как это не предусмотрено в коде. Хорошее решение в данном случае – использовать re-entrant замок. Модуль threading предоставляет такой, в виде функции RLock. Просто замените строку lock = threading.Lock() на lock = threading.RLock() и попробуйте перезапустить код. Теперь он должен заработать. Если вы хотите попробовать код выше но добавить в него потоки, то мы можем заменить call на main следующим образом:
if __name__ == '__main__':
for i in range(10):
my_thread = threading.Thread(target=main)
my_thread.start()Так мы запустим основную функцию в каждом потоке, что в свою очередь приведет к вызову остальных двух функций. В конце вы получите достаточно крупную выдачу.
Таймеры
Модуль threading включает в себя один очень удобный класс, под названием Timer, который вы можете использовать запуска действия, спустя определенный отрезок времени. Данный класс запускает собственный поток и начинают работу с того же метода start(), как и обычные потоки. Вы также можете остановить таймер, используя метод cancel. Обратите внимание на то, что вы можете отменить таймер еще до того, как он стартовал. Однажды у меня был случай, когда мне нужно было наладить связь с под-процессом, который я начал, но мне нужен было обратный отсчет. Несмотря на существования ряда различных способов решения этой отдельной проблемы, моим любимым решением всегда было использование класса Timer модуля threading. Для этого примера мы взглянем на применение команды ping. В Linux, команда ping будет работать, пока вы её не убьете. Так что класс Timer становится особенно полезным для мира Linux. Вот пример:
import subprocess
from threading import Timer
kill = lambda process: process.kill()
cmd = ['ping', 'www.google.com']
ping = subprocess.Popen(
cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
my_timer = Timer(5, kill, [ping])
try:
my_timer.start()
stdout, stderr = ping.communicate()
finally:
my_timer.cancel()
print (str(stdout))Здесь мы просто настраиваем лямбду, которую мы можем использовать, чтобы убить процесс. Далее мы начинаем нашу работу над ping и создаем объект Timer. Обратите внимание на то, что первый аргумент – это время ожидания в секундах, затем – функция, которую нужно вызвать и аргумент, который будет передан функции. В нашем случае, наша функция – это лямбда, и мы передаем её список аргументов, где список содержит только один элемент. Если вы запустите этот код, он будет работать примерно 5 секунд, после чего выведет результат пинга.
Другие Компоненты Потоков
Модуль threading также включает в себя поддержу других объектов. Например, вы можете создать семафор, который является одним из древнейших синхронизационных примитивов в компьютерной науке. Семафор управляет внутренним счетчиком, который будет декрементируется когда вы вызываете acquire и увеличивается, когда вы вызываете release. Счетчик разработан таким образом, что он не может падать ниже нуля. Так что если так вышло, что вы вызываете acquire когда он равен нулю, то он заблокируется.
Еще один полезный инструмент, который содержится в модуле, это Event. С его помощью вы можете получить связь между двумя потоками, используя сигналы. Мы рассмотрим примеры применения Event в следующей статье. Наконец-то, в версии Python 3.2 был добавлен объект Barrier. Это примитив, который управляет пулом потока, при этом не важно, где потоки должны ждать своей очереди. Для передачи барьера, потоку нужно вызвать метод wait(), который будет блокировать до тех пор, пока все потоки не сделают вызов. После чего все потоки пройдут дальше одновременно.
Связь потоков
Существует ряд случаев, когда вам нужно сделать так, чтобы потоки были связанны друг с другом. Как я упоминал ранее, вы можете использовать Event для этой цели. Но более удобный способ – использовать Queue. В нашем примере мы используем оба способа! Давайте посмотрим, как это будет выглядеть:
import threading
from queue import Queue
def creator(data, q):
"""
Creates data to be consumed and waits for the consumer
to finish processing
"""
print('Creating data and putting it on the queue')
for item in data:
evt = threading.Event()
q.put((item, evt))
print('Waiting for data to be doubled')
evt.wait()
def my_consumer(q):
"""
Consumes some data and works on it
In this case, all it does is double the input
"""
while True:
data, evt = q.get()
print('data found to be processed: {}'.format(data))
processed = data * 2
print(processed)
evt.set()
q.task_done()
if __name__ == '__main__':
q = Queue()
data = [5, 10, 13, -1]
thread_one = threading.Thread(target=creator, args=(data, q))
thread_two = threading.Thread(target=my_consumer, args=(q,))
thread_one.start()
thread_two.start()
q.join()Давайте немного притормозим. Во первых, у нас есть функция creator (также известная, как producer), которую мы используем для создания данных, с которыми мы хотим работать (или использовать). Далее мы получаем еще одну функцию, которую мы используем для обработки данных, под названием my_consumer. Функция creator использует метод Queue под названием put, чтобы добавить данные в очередь, затем потребитель, в свою очередь, будет проверять, есть ли новые данные и обрабатывать их, когда такие появятся. Queue обрабатывает все закрытия и открытия замков, так что лично вам эта участь не грозит.
В данном примере мы создали список значений, которые мы хотим дублировать. Далее мы создаем два потока, один для функции creator/producer, второй для consumer (потребитель). Обратите внимание на то, что мы передаем объект Queue каждому потоку, что является прямо таки магией, учитывая то, как обрабатываются замки. Очередь начнется с первого потока, который передает данные второму. Когда первый поток передает те или иные данные в очередь, он также передает их к Event, после чего дожидается, когда произойдет события, чтобы закончить. Далее, в функции consumer, данные обрабатываются, и после этого вызывается метод настройки Event, который указывает первому потоку, что второй закончил обработку, так что он может продолжать. Последняя строка кода вызывает метод join объекта Queue, который указывает Queue подождать, пока потоки закончат обработку. Первый поток заканчивает, когда ему больше нечего передавать в Queue.
Подведем итоги
Мы рассмотрели достаточно много материала. А именно:
- Основы работы с модулем threading
- Как работают замки
- Что такое Event и как его можно использовать
- Как использовать таймер
- Внутрипотоковая связь с использованием Queue/Event
Теперь вы знаете, как использовать потоки, и в чем они хороши. Надеюсь, вы найдете им применение в своем собственном коде!
thread — Wiktionary
English[edit]
Alternative forms[edit]
Etymology[edit]
From Middle English thred, þred, threed, from Old English þrǣd, from Proto-Germanic *þrēduz, from Proto-Indo-European *treh₁-tu-s, from *terh₁- (“rub, twist”). Cognates with Saterland Frisian Träid (“thread, wire”), West Frisian tried, Dutch draad, German Draht, Norwegian, Danish and Swedish tråd, and Icelandic þráður. Non-Germanic cognates include Albanian dredh (“twist, turn”). More at throw.
Pronunciation[edit]
- IPA(key): /θɹɛd/
- Rhymes: -ɛd
- Hyphenation: thread
Noun[edit]
thread (plural threads)
- A long, thin and flexible form of material, generally with a round cross-section, used in sewing, weaving or in the construction of string.
- 1922, Michael Arlen, “Ep./1/2”, in “Piracy”: A Romantic Chronicle of These Days:
He walked. To the corner of Hamilton Place and Picadilly, and there stayed for a while, for it is a romantic station by night. The vague and careless rain looked like threads of gossamer silver passing across the light of the arc-lamps.
- 1922, Michael Arlen, “Ep./1/2”, in “Piracy”: A Romantic Chronicle of These Days:
- A continued theme or idea.
- Synonym: topic
All of these essays have a common thread.
I’ve lost the thread of what you’re saying.
- (engineering) A screw thread.
- A sequence of connections.
- 1847, Charlotte Brontë, Jane Eyre, Chapter XVIII:
I was pondering these things, when an incident, and a somewhat unexpected one, broke the thread of my musings.
- 1897, Bram Stoker, Dracula, Chapter 21:
‘Let him go on. Do not interrupt him. He cannot go back, and maybe could not proceed at all if once he lost the thread of his thought.’
- 1847, Charlotte Brontë, Jane Eyre, Chapter XVIII:
- The line midway between the banks of a stream.
- (computing) A unit of execution, lighter in weight than a process, usually sharing memory and other resources with other threads executing concurrently.
- (Internet) A series of messages, generally grouped by subject, in which all messages except the first are replies to previous messages in the thread.
- A filament, as of a flower, or of any fibrous substance, as of bark.
- (figurative) Composition; quality; fineness.
Hyponyms[edit]
Derived terms[edit]
Translations[edit]
long, thin and flexible form of material
|
|
the line midway between the banks of a stream
(computing): a unit of execution
(Internet): a series of messages
|
Verb[edit]
thread (third-person singular simple present threads, present participle threading, simple past threaded or (archaic) thrid, past participle threaded or (archaic) thridden)
- (transitive) To put thread through.
- thread a needle
- (transitive) To pass (through a narrow constriction or around a series of obstacles).
I think I can thread my way through here, but it’s going to be tight.
- 1961 February, D. Bertram, “The lines to Wetherby and their traffic”, in Trains Illustrated, page 101:
On the descent the line is often in cuttings; some are high, such as at Scarcroft, where a cut through firestone and fireclay was necessary, and near Bardsey, where the line threads a deep tree-lined gorge.
- 2013, Ben Smith, «[1]», BBC Sport, 19 October 2013:
- Picking the ball up in his own half, Januzaj threaded a 40-yard pass into the path of Rooney to slice Southampton open in the blink of an eye.
- To screw on, to fit the threads of a nut on a bolt
Derived terms[edit]
Translations[edit]
See also[edit]
References[edit]
Anagrams[edit]
Etymology[edit]
From English thread.
Pronunciation[edit]
Noun[edit]
thread m (plural threads)
- (anglicism, computing) Thread
- (anglicism, Internet) Thread
Synonyms[edit]
Italian[edit]
Etymology[edit]
From English thread.
Noun[edit]
thread m (invariable)
- (Internet) thread (series of messages)
Portuguese[edit]
Etymology[edit]
From English thread.
Noun[edit]
thread m (plural threads)
- (computing) thread (one of several units of execution running concurrently)
Попробовать новую тему со скидкой
Попробовать новую тему со скидкой | SuperiorThreads.com
НАСЛАЖДАЙТЕСЬ БЕСПЛАТНОЙ ДОСТАВКОЙ НА ЗАКАЗЫ НА 50 $ ИЛИ БОЛЕЕ (НА АДРЕСЫ США)
Попробуйте Me Specials
Категория
Размещение
Вес резьбы
Тип волокна
Размер промышленной резьбы
НОВЫЙ ЗАЗОР
4 доллара.99
2,49 $
50% скидка
НОВЫЙ ЗАЗОР
4 доллара.99
2,49 $
50% скидка
НЕТ НА СКЛАДЕ
6 долларов.99
5,59 долл. США
скидка в 20
НОВЫЙ ЗАЗОР
6,00 долларов США
2,99 доллара США
50% скидка
.
java — Основной поток и текущий поток
Переполнение стека
- Около
Продукты
- Для команд
Переполнение стека
Общественные вопросы и ответыПереполнение стека для команд
Где разработчики и технологи делятся частными знаниями с коллегамиВакансии
Программирование и связанные с ним технические возможности карьерного ростаТалант
Нанимайте технических специалистов и создавайте свой бренд работодателяРеклама
Обратитесь к разработчикам и технологам со всего мира- О компании
Загрузка…
.