Потоки в Python

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

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

Каждому потоку выделяется свой стек — область оперативной памяти, а также у потоков есть общий доступ к общей памяти процесса. Например, если в общей памяти есть список с данными, то его можно читать и изменять из потоков, созданных в рамках процесса.

В Python работа с потоками производится с помощью модуля threading.

Создание потока

Рассмотрим простейший пример создания потока.

PythonКопировать

import time
from threading import Thread, current_thread


def foo(text):
    time.sleep(2)
    print('[foo] Current thread', current_thread().name)
    print('msg:', text)


t = Thread(target=foo, args=('Hello world',))
t.start()

print('[main] Current thread', current_thread().name)
print('Waiting for thread end...')

t.join()

print('Threads finished')

Пример вывода результата выполнения в консоль:

TerminalКопировать

[main] Current thread MainThread
Waiting for thread end...
[foo] Current thread foo-thread
msg: Hello world
Threads finished

Рассмотрим пример подробнее. Для начала объявляем функцию foo, которая будет запускаться в отдельном потоке. Затем с помощью класса Thread создаём экземпляр потока, передав в параметр target callable-объект (в нашем случае функцию foo), а также кортеж аргументов функции в параметр args. Параметр name позволяет дать потоку осмысленное имя. Метод start() запускает поток. Метод join() позволяет дождаться завершения выполнения потоков в программе.

Также в модуле threading есть функция current_thread(), которая позволяет получить экземпляр текущего потока, связанный с контекстом выполнения callable-объекта. Добавив вывод имени потока, можно увидеть, в каких потоках происходит выполнение кода.

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

Таким образом и работает поток: функция выполняется независимо от основного кода скрипта, или наоборот.

Потоки-демоны

Если не добавлять в код основной программы инструкцию t.join(), то основной код завершится, но при этом, если какие-то потоки будут продолжать работу, общее выполнение программы завершится только после выполнения всех потоков.

Такое поведение не всегда желательно, так как потоки для выполнения фоновых задач (например, монтирование устройств) могут выполняться постоянно. Чтобы указать основной программе, что процесс фоновый и не нужно ждать его завершения, при создании потока нужно добавить параметр daemon=True.

PythonКопировать

t = Thread(target=foo, args=('Hello world',), daemon=True)

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

Класс ThreadPoolExecutor

Часто требуется запуск нескольких потоков одновременно — например, для обработки списка из нескольких файлов или запросов к нескольким URL. Библиотека concurrent.futures предоставляет класс ThreadPoolExecutor для упрощения обработки параллельных потоков.

Обычно экземпляр ThreadPoolExecutor создаётся с контекстным менеджером with, чтобы определять исполняемые блоки и очищать потоки после выполнения.

Основные методы класса — submit и map.

submit позволяет создать поток для одной функции с переданными в неё параметрами. Синтаксис: submit(func: Callable, [*args, **kwargs]).

map позволяет запустить несколько потоков для указанного callable-объекта и массива входных параметров для него. Синтаксис: map(func, *iterables, timeout=None, [chunksize]).

Рассмотрим на примере:

PythonКопировать

from concurrent.futures import ThreadPoolExecutor
import time
from threading import current_thread


def sum_len(data: list[int]) -> int:
    return sum(data)


def foo(text: str) -> int:
    time.sleep(len(text))
    print('[foo] Current thread:', current_thread().name)
    print('msg:', text)
    return len(text)


fruit_list = ['apple', 'banana', 'pineapple']

with ThreadPoolExecutor(max_workers=3) as pool:
    print('[main] Current thread:', current_thread().name)
    results = pool.map(foo, fruit_list)
    print('results type:', type(results))

    result = pool.submit(sum_len, results)
    print('result type:', type(result))

    print('Waiting for thread end...')

print('result:', result.result())
print('Threads finished')

Вывод в консоль:

TerminalКопировать

[main] Current thread: MainThread
results type: <class 'generator'>
result type: <class 'concurrent.futures._base.Future'>
Waiting for thread end...
[foo] Current thread: ThreadPoolExecutor-0_0
msg: apple
[foo] Current thread: ThreadPoolExecutor-0_1
msg: banana
[foo] Current thread: ThreadPoolExecutor-0_2
msg: pineapple
result: 20
Threads finished

Данный код позволяет получить для списка длину строковых значений и затем найти их сумму. Сначала выполняем необходимые импорты и определяем две функции: foo, которая возвращает длину строки, и sum_len, которая возвращает сумму длин элементов списка. Определяем список fruit_list для тестов.

Затем с помощью контекстного менеджера with создаём экземпляр пула потоков. В качестве параметра указываем максимальное количество воркеров через max_workers.

С помощью инструкции pool.map указываем, что нам нужно выполнить функцию foo для каждого элемента списка fruit_list в отдельном потоке и записать результат в переменную results. Обратите внимание на тип переменной results — это генератор.

Инструкция pool.submit указывает, что мы должны применить к результату выполнения предыдущих потоков функцию sum_len и записать полученное значение в переменную result. Типом переменной result будет concurrent.futures._base.Future (или, как её ещё называют, «футура»). Футура — это ожидаемый объект. Чтобы получить его вычисленный результат, можно использовать метод .result().

Заключение

Рассмотрим ещё раз основные достоинства и недостатки потоков в Python.

Достоинства:

Доступна общая память между потоками.

Хорошо подходят для распараллеливания IO-операций (операций ввода-вывода).

Более легковесные, требуют меньше памяти и ресурсов.

Недостатки:

Нет возможности прервать выполнение потока программно.

GIL не позволяет потокам работать одновременно.