Параллелизм в Python - пул потоков

Предположим, нам нужно было создать большое количество потоков для наших многопоточных задач. Это было бы наиболее затратно в вычислительном отношении, поскольку из-за слишком большого количества потоков может возникнуть множество проблем с производительностью. Основная проблема может заключаться в ограничении пропускной способности. Мы можем решить эту проблему, создав пул потоков. Пул потоков может быть определен как группа предварительно созданных и бездействующих потоков, готовых к выполнению работы. Создание пула потоков предпочтительнее, чем создание экземпляров новых потоков для каждой задачи, когда нам нужно выполнить большое количество задач. Пул потоков может управлять одновременным выполнением большого количества потоков следующим образом:

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

Модуль Python - Concurrent.futures

Стандартная библиотека Python включает concurrent.futuresмодуль. Этот модуль был добавлен в Python 3.2 для предоставления разработчикам высокоуровневого интерфейса для запуска асинхронных задач. Это уровень абстракции над модулями потоковой обработки и многопроцессорности Python для предоставления интерфейса для выполнения задач с использованием пула потоков или процессов.

В наших последующих разделах мы узнаем о различных классах модуля concurrent.futures.

Класс исполнителя

Executorэто абстрактный класс concurrent.futuresМодуль Python. Его нельзя использовать напрямую, и нам нужно использовать один из следующих конкретных подклассов -

ThreadPoolExecutor
ProcessPoolExecutor

ThreadPoolExecutor - конкретный подкласс

Это один из конкретных подклассов класса Executor. Подкласс использует многопоточность, и мы получаем пул потоков для отправки задач. Этот пул назначает задачи доступным потокам и планирует их запуск.

Как создать ThreadPoolExecutor?

С помощью concurrent.futures модуль и его конкретный подкласс Executor, мы можем легко создать пул потоков. Для этого нам нужно построитьThreadPoolExecutorс количеством потоков, которые мы хотим в пуле. По умолчанию это число 5. Затем мы можем отправить задачу в пул потоков. Когда мыsubmit() задача, мы возвращаем Future. У объекта Future есть метод, называемыйdone(), который сообщает, решилось ли будущее. Таким образом, для этого конкретного будущего объекта установлено значение. Когда задача завершается, исполнитель пула потоков устанавливает значение для будущего объекта.

пример

from concurrent.futures import ThreadPoolExecutor
from time import sleep
def task(message):
   sleep(2)
   return message

def main():
   executor = ThreadPoolExecutor(5)
   future = executor.submit(task, ("Completed"))
   print(future.done())
   sleep(2)
   print(future.done())
   print(future.result())
if __name__ == '__main__':
main()

Выход

False
True
Completed

В приведенном выше примере ThreadPoolExecutorбыл построен с 5 потоками. Затем исполнителю пула потоков передается задача, которая будет ждать 2 секунды перед отправкой сообщения. Как видно из выходных данных, задача не выполняется до 2 секунд, поэтому первый вызовdone()вернет False. Через 2 секунды задача выполнена и мы получаем результат будущего, вызываяresult() метод на нем.

Создание экземпляра ThreadPoolExecutor - диспетчера контекста

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

with ThreadPoolExecutor(max_workers = 5) as executor

пример

Следующий пример заимствован из документации Python. В этом примере, прежде всего,concurrent.futuresмодуль должен быть импортирован. Затем функция с именемload_url()создается, который загрузит запрошенный URL. Затем функция создаетThreadPoolExecutor с 5 потоками в пуле. ВThreadPoolExecutorбыл использован как менеджер контекста. Мы можем получить результат будущего, позвонивresult() метод на нем.

import concurrent.futures
import urllib.request

URLS = ['http://www.foxnews.com/',
   'http://www.cnn.com/',
   'http://europe.wsj.com/',
   'http://www.bbc.co.uk/',
   'http://some-made-up-domain.com/']

def load_url(url, timeout):
   with urllib.request.urlopen(url, timeout = timeout) as conn:
   return conn.read()

with concurrent.futures.ThreadPoolExecutor(max_workers = 5) as executor:

   future_to_url = {executor.submit(load_url, url, 60): url for url in URLS}
   for future in concurrent.futures.as_completed(future_to_url):
   url = future_to_url[future]
   try:
      data = future.result()
   except Exception as exc:
      print('%r generated an exception: %s' % (url, exc))
   else:
      print('%r page is %d bytes' % (url, len(data)))

Выход

Ниже будет вывод вышеупомянутого скрипта Python -

'http://some-made-up-domain.com/' generated an exception: <urlopen error [Errno 11004] getaddrinfo failed>
'http://www.foxnews.com/' page is 229313 bytes
'http://www.cnn.com/' page is 168933 bytes
'http://www.bbc.co.uk/' page is 283893 bytes
'http://europe.wsj.com/' page is 938109 bytes

Использование функции Executor.map ()

Питон map()функция широко используется в ряде задач. Одна из таких задач - применить определенную функцию к каждому элементу в итерациях. Точно так же мы можем сопоставить все элементы итератора с функцией и отправить их как независимые задания на outThreadPoolExecutor. Рассмотрим следующий пример скрипта Python, чтобы понять, как работает функция.

пример

В этом примере ниже функция карты используется для применения square() для каждого значения в массиве значений.

from concurrent.futures import ThreadPoolExecutor
from concurrent.futures import as_completed
values = [2,3,4,5]
def square(n):
   return n * n
def main():
   with ThreadPoolExecutor(max_workers = 3) as executor:
      results = executor.map(square, values)
for result in results:
      print(result)
if __name__ == '__main__':
   main()

Выход

Вышеупомянутый скрипт Python генерирует следующий вывод -