Многопоточность. Thread и создание потоков

Многопоточность представляет способность программы выполнять несколько задач одновременно в рамках одного процесса. Потоки незаменимы в операциях, где программа большую часть времени ждет ответа от внешних ресурсов, например, скачивание файлов из интернета, чтение/запись файлов или отправка запросов к базам данных. В Python за работу с потоками отвечает встроенный модуль threading. Ключевым компонентом для работы с потоками в языке Python является класс Thread.

Главный поток

Вне зависимости от того, создаем ли мы вручную потоки или нет, при запуске программы создается как минимум один поток - главный поток программы. Этот поток мы можем получить с помощью встроенной функции threading.main_thread() в виде объекта Thread. В обычных условиях основным потоком является поток, из которого был запущен интерпретатор Python. Например, пропишем скрипте на Python следующий код:

import threading

current_thread = threading.main_thread()
print("Main Thread:")
print(current_thread)

Консоль нам выведет что-то вроде следующего:

Main Thread:
<_MainThread(MainThread, started 124401057440256)>

Этот вывод показывает текущее состояние потока. А именно:

• _MainThread: имя класса в модуле threading. Оно указывает на то, что перед нами именно главный поток программы (тот, который создается автоматически при запуске Python-скрипта). Вообще "MainThread" - стандартное имя этого потока.

• started: статус потока. Он означает, что поток сейчас активен и выполняется в данный момент.

• 124401057440256: уникальный идентификатор потока (ID) в памяти операционной системы. Это число генерируется заново при каждом новом запуске программы.

Свойства потоков

Для объектов класса Thread доступен ряд атрибутов, с помощью которых можно получить состояние/свойства потока:

• name: имя потока. Фактически это строка, которая используется только для идентификации. В реальности нескольким потокам может быть присвоено одно и то же имя. На некоторых платформах имя потока устанавливается на уровне операционной системы при его запуске, чтобы оно было видно в диспетчерах задач.

• ident: идентификатор потока (если поток не был запущен, то равно None). Это ненулевое целое число. Идентификаторы потоков могут быть повторно использованы при завершении одного потока и создании другого. Идентификатор доступен даже после завершения работы потока.

• native_id: нативный идентификатор потока. Это также идентификатор потока, назначенный операционной системой (ядром) - неотрицательное целое число или None, если поток не был запущен. Это значение может использоваться для уникальной идентификации данного потока в масштабах всей системы (до завершения потока, после чего значение может быть повторно использовано операционной системой).

• daemon: логическое значение, которое указывает, является ли данный поток потоком-демоном (True) или нет (False).

Использование свойств:

import threading

current_thread = threading.main_thread()
print("Main Thread name: ", current_thread.name)
print("Main Thread id: ", current_thread.ident)
print("Main Thread native id:", current_thread.native_id)

Пример консольного вывода:

Main Thread name:  MainThread
Main Thread id:  135968807522816
Main Thread native id: 35474

Как видно, у потоков в Python есть два атрибута, которые хранят его идентификатор: ident и native_id. Основное отличие между ними заключается в том, кто присваивает этот идентификатор и где он используется:

• ident - это внутренний ID потока в самом Python. ident удобен для внутренней логики программы на Python. Например, для логирования или использования в качестве ключей в словарях, чтобы разделять данные между потоками внутри самого приложения.

• native_id - это реальный ID потока в операционной системе. native_id незаменим, если нужно связать поток внутри Python с системными утилитами. Например, если поток завис, с помощью native_id можно найти конкретно его в системном мониторе процессов (в Linux это TID в команде top -H или ps), чтобы завершить или профилировать на уровне ОС.

Краткое сравнение этих атрибутов:

Характеристика	ident	native_id
Кто управляет	Виртуальная машина Python (интерпретатор)	Операционная система (Windows, Linux, macOS)
Доступность	Доступен всегда, во всех версиях Python 3.	Появился только в Python 3.8.
Уникальность	Уникален только пока поток жив. После завершения потока Python может использовать этот ID повторно для новых потоков.	Уникален на уровне всей ОС в данный момент времени (как PID для процессов).
Связь с ОС	Может не совпадать с системным ID потока (зависит от реализации Python и ОС).	Гарантированно совпадает с ID, который видит ОС (например, в диспетчере задач).

Для получения идентификаторов текущего потока в модуле threading доступны методы threading.get_ident() и threading.get_native_id():

import threading

ident = threading.get_ident()
native_id = threading.get_native_id()

print("Python ID: ", ident)
print("Native ID:", native_id)

Создание новых потоков

Для создания новых потоков применяется конструктор класса Thread:

class threading.Thread(group=None, name=None, args=(), kwargs={}, *, daemon=None, context=None)

Основные параметры:

• target: действие, которое собственно и будет выполняться в потоке. По умолчанию равно None, что указывает, что никаких действий не вызывается

• name: имя потока. По умолчанию создается уникальное имя вида "Thread-N", где N - небольшое десятичное число, или "Thread-N (target)"", где target - это target.__name__, если указан аргумент target.

К примеру, определим свой поток:

import threading

my_thread = threading.Thread(target=print)
print(my_thread)

Здесь в качестве target - вызываемого действия выступает встроенная функция print(). При выполнении программа выведет что-то наподобие:

<Thread(Thread-1 (print), initial)>

Аналогично можно передать и другую, в том числе свою функцию:

import threading

def my_task():
    print("Дочерний поток: Начал работу...")
    time.sleep(2)  # Имитация долгой работы
    print("Дочерний поток: Завершил работу!")

my_thread = threading.Thread(target=my_task)
print(my_thread)    # <Thread(Thread-1 (my_task), initial)>

Запуск потока

Для запуска потока применяется метод start():

import threading

# Функция, выполняемая в другом потоке
def my_task():
    print("Дочерний поток: Начал работу...")
    #time.sleep(2)  # Имитация долгой работы
    print("Дочерний поток: Завершил работу!")

print("Главный поток: Начал работу...")

# Создаем поток
my_thread = threading.Thread(target=my_task)

# Запускаем поток
my_thread.start()

print("Главый поток: Завершил работу!")

Главный поток программы запускает my_thread и сразу идет дальше, не дожидаясь его окончания. Результат программы недетерминарован. Например, он может быть таким:

Главный поток: Начал работу...
Дочерний поток: Начал работу...
Дочерний поток: Завершил работу!
Главый поток: Завершил работу!

А может быть и таким:

Главный поток: Начал работу...
Дочерний поток: Начал работу...
Главый поток: Завершил работу!
Дочерний поток: Завершил работу!

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

import threading
import time

# Функция, выполняемая в другом потоке
def my_task():
    print("Дочерний поток: Начал работу...")
    time.sleep(2)  # Имитация долгой работы
    print("Дочерний поток: Завершил работу!")

print("Главный поток: Начал работу...")

# Создаем поток
my_thread = threading.Thread(target=my_task)

# Запускаем поток
my_thread.start()

print("Главый поток: Завершил работу!")

Здесь в функции my_task с помощью time.sleep(2) устанавливаем задержку на 2 секунды. Это гарантирует, что главный поток завершится раньше, чем my_thread.

Стоит отметить, что у потока Thread также есть метод run(), который также запускает действие. Но в отличие от метода start() новый поток не создается, а действие просто выполняется как обычная функция в рамках главного потока программы:

# Создаем поток
my_thread = threading.Thread(target=my_task)

# новый поток НЕ запускается, my_task выполняется как обычная функция
my_thread.run()

Передача аргументов в функцию потока

Если выполняемой в новом потоке функции требуются аргументы, то для их передачи можно использовать параметр args (список или кортеж аргументов для вызова target) и kwargs (словарь именованных аргументов для вызова target). Например:

import threading

my_thread = threading.Thread(target=print, args=("Hello METANIT.COM",))

my_thread.start()

Здесь в новом потоке запускается встроенная функция print(), которой передается один аргумент - строка "Hello METANIT.COM". Обратите внимание, что аргументу args передается кортеж, поэтому после строки указывается запятая - args=("Hello METANIT.COM",)

Или другой пример:

import threading
import time

def greet_user(name, delay):
    time.sleep(delay)
    print(f"Привет, {name}!")

# создаем два потока
thread1 = threading.Thread(target=greet_user, args=("Алексей", 1))
thread2 = threading.Thread(target=greet_user, args=("Мария", 2))

thread1.start()
thread2.start()

Здесь создается и запускается два потока, каждому из которых передается свой набор данных - два аргумента. В итоге мы получим следующий консольный вывод:

Привет, Алексей!
Привет, Мария!

Ожидание завершения. Метод join()

Часто главному потоку требуется дождаться результатов работы остальных потоков перед тем, как завершить программу или сложить полученные данные. Для этого используется метод join(), который есть у класса Thread. Например:

import threading
import time

def work():
    time.sleep(2)   # задержка в 2 секунды
    print("Work Thread завершил работу")
    
my_tread = threading.Thread(target=work)
my_tread.start()

# Говорим главному потоку подождать завершения остальных потоков
my_tread.join() 

print("Main Thread завершил работу")

Благодаря строке my_tread.join() главный поток на этой строке остановится и будет ждать завершения потока my_tread. Соответственно, если эту строку убрать, то главный поток завершить свое выполнение ранее my_tread.

Создание группы потоков

В реальных приложениях задачи масштабируют с помощью списков. Создадим 3 потока, которые будут выполнять работу параллельно:

import threading
import time

# условная функция загрузки файла
def download_file(file_id):
    print(f"Скачивание файла {file_id}...")
    time.sleep(2)
    print(f"Файл {file_id} успешно скачан!")

# список потоков
threads = []

# Создаем и запускаем 3 потока
for i in range(1, 4):
    t = threading.Thread(target=download_file, args=(i,))
    threads.append(t)
    t.start()

# дожидаемся завершения КАЖДОГО потока
for t in threads: t.join()

print("Все файлы скачаны!")

Здесь функция download_file() имитирует загрузку файла, принимая в качестве параметра условный идентификатор файла. В цикле запускаем три потока, добавляя их в список threads для последующего управления. В итоге все условные три файла начнут скачиваться одновременно. Вместо последовательного ожидания в 6 секунд (3 файла по 2 секунды), программа выполнится всего за 2 секунды.

Ограничение GIL (Global Interpreter Lock)

В стандартном Python (CPython) существует глобальная блокировка - GIL. Она не позволяет нескольким потокам одновременно выполнять чистый Python-код на разных ядрах процессора. Таким образом, в каждый конкретный момент времени только один поток может выполнять байт-код Python.