Python включает в себя ряд разных параллельных конструкций, таких как как threading, queues и multiprocessing. Модуль threading использовался как главный способ достижения параллельности. Несколько лет назад, модуль multiprocessing был добавлен в пакет стандартных библиотек Python. В этой статье мы сфокусируемся на том, как использовать очереди и потоки (queues и threads).
Использование потоков
Мы начнем с простого примера, который демонстрирует работу потоков. Мы наследуем класс Thread в класс MyThread и укажем, чтобы его имя выводилось как stdout. Попробуем!
|
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 потоков, дав каждому из них уникальное имя. Если вы запустите данный код, вы увидите что-то вроде этого:
|
1 2 3 4 5 |
Thread #2 is running Thread #3 is running Thread #1 is running Thread #4 is running Thread #5 is running |
Порядок выхода каждый раз будет разным. Попробуйте запустить код несколько раз, чтобы увидеть смену порядка. Теперь давайте напишем что-нибудь более практичное!
Написание потокового загрузчика
Предыдущий пример был не слишком полезным в качестве инструмента, показывающего, как именно работают Python потоки. Так что в данном примере, мы создадим класс Thread, который скачивает параллельно файлы из интернета. Мы воспользуемся бесплатным ресурсом в нашем демо. Посмотрим на код:
|
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, извлекаем название файла, после чего используем это название для того, чтобы создать файл на диске.
После этого мы используем цикл для загрузки файла по килобайту за раз, и сохранять его на диск. После того, как сохранение файла завершится, мы выводим название потока и тот url, который загрузился. В Python 3 этот код немного отличаться. Нам нужно импортировать urllib вместо urllib2 и использовать urllib.request.urlopen вместо urllib2.urlopen. Вот код, в котором вы можете увидеть разницу:
|
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. Приступим!
|
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 документов за секунду.
Подведем итоги
Теперь вы знаете, как использовать потоки и очереди, как в теории, так и на практике. Потоки особенно полезны, когда вы создаете пользовательский интерфейс и вам нужно поддерживать этот интерфейс юзабельным. Без потоков, пользовательский интерфейс может перестать реагировать и виснуть, пока вы загружаете большой файл, или создаете большой запрос к базе данных. Во избежание этого, вам нужно выполнять длительные процессы в потоках, и связать их с вашим интерфейсом.
Являюсь администратором нескольких порталов по обучению языков программирования Python, Golang и Kotlin. В составе небольшой команды единомышленников, мы занимаемся популяризацией языков программирования на русскоязычную аудиторию. Большая часть статей была адаптирована нами на русский язык и распространяется бесплатно.
E-mail: vasile.buldumac@ati.utm.md
Образование
Universitatea Tehnică a Moldovei (utm.md)
- 2014 — 2018 Технический Университет Молдовы, ИТ-Инженер. Тема дипломной работы «Автоматизация покупки и продажи криптовалюты используя технический анализ»
- 2018 — 2020 Технический Университет Молдовы, Магистр, Магистерская диссертация «Идентификация человека в киберпространстве по фотографии лица»