Многопоточность в Python3

Многопоточность похожа на одновременное выполнение нескольких разных программ. Запуск многопоточности имеет следующие преимущества:

  • Использование потоков позволяет помещать задачи, занимающие много времени в программе, в фоновый режим для обработки.
  • Пользовательский интерфейс может быть более привлекательным. Например, когда пользователь нажимает кнопку, чтобы запустить обработку определённых событий, может появиться индикатор выполнения, показывающий ход обработки.
  • Скорость работы программы может увеличиться.
  • Потоки более полезны для реализации некоторых задач ожидания, таких как пользовательский ввод, чтение и запись файлов, а также отправка и приём сетевых данных. В таких случаях мы можем освободить некоторые ценные ресурсы, например используемую память.

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

Каждый поток имеет собственный набор регистров ЦП, называемый контекстом потока, который отражает состояние регистров ЦП во время последнего запуска потока.

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

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

Потоки можно разделить на:

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

Два часто используемых модуля для потоков в Python3:

  • _thread
  • threading (рекомендуется)

Модуль thread устарел. Вместо него пользователи могут использовать модуль threading. Поэтому в Python3 модуль "thread" больше нельзя использовать. Для совместимости Python3 переименовал thread в "_thread".


Начните изучать потоки Python

В Python есть два способа использования потоков: функции или использование классов для обёртки объектов потоков.

Функциональный стиль: вызовите функцию start_new_thread() в модуле _thread, чтобы создать новый поток. Синтаксис следующий:

_thread.start_new_thread ( function, args[, kwargs] )

Описание параметра:

  • function - Функция потока.
  • args - Параметры, передаваемые функции потока; должен иметь тип tuple.
  • kwargs - Необязательные параметры.

Пример

#!/usr/bin/python3

import _thread
import time

# Определить функцию для потока
def print_time( threadName, delay):
   count = 0
   while count < 5:
      time.sleep(delay)
      count += 1
      print ("%s: %s" % ( threadName, time.ctime(time.time()) ))

# Создать два потока
try:
   _thread.start_new_thread( print_time, ("Thread-1", 2, ) )
   _thread.start_new_thread( print_time, ("Thread-2", 4, ) )
except:
   print ("Ошибка: не удалось запустить поток")

while 1:
   pass

Вывод выполнения вышеуказанной программы выглядит следующим образом:

Thread-1: Wed Jan  5 17:38:08 2022
Thread-2: Wed Jan  5 17:38:10 2022
Thread-1: Wed Jan  5 17:38:10 2022
Thread-1: Wed Jan  5 17:38:12 2022
Thread-2: Wed Jan  5 17:38:14 2022
Thread-1: Wed Jan  5 17:38:14 2022
Thread-1: Wed Jan  5 17:38:16 2022
Thread-2: Wed Jan  5 17:38:18 2022
Thread-2: Wed Jan  5 17:38:22 2022
Thread-2: Wed Jan  5 17:38:26 2022

После выполнения вышеуказанной программы вы можете нажать ctrl-c для выхода.

Модуль Thread

Python3 обеспечивает поддержку потоков через две стандартные библиотеки: _thread и threading.

_thread предоставляет низкоуровневые, примитивные потоки и простую блокировку. Его функциональность всё ещё относительно ограничена по сравнению с модулем threading.

В дополнение к тому, что модуль threading содержит все методы модуля _thread, он также предоставляет следующие методы:

  • threading.current_thread(): Возвращает переменную текущего потока.
  • threading.enumerate(): Возвращает список, содержащий все работающие потоки. «Работающий» означает после того, как поток запущен, и до того, как он завершился, не включая потоки до запуска или после завершения.
  • threading.active_count(): Возвращает количество работающих потоков; результат совпадает с len(threading.enumerate()).
  • threading.Thread(target, args=(), kwargs={}, daemon=None)：
    • СоздайтеThreadэкземпляр класса.
    • target: Целевая функция, которую должен выполнить поток.
    • args: Параметры целевой функции, передаваемые в виде кортежа.
    • kwargs: Именованные аргументы целевой функции, передаваемые в виде словаря.
    • daemon: Указывает, является ли поток демоническим.

Класс threading.Thread предоставляет следующие методы и атрибуты:

  1. __init__(self, group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)：

    • ИнициализацияThreadобъект.
    • group: Группа потоков; в настоящее время не используется, зарезервирована для будущего расширения.
    • target: Целевая функция, которую должен выполнить поток.
    • name: Имя потока.
    • args: Параметры целевой функции, передаваемые в виде кортежа.
    • kwargs: Именованные аргументы целевой функции, передаваемые в виде словаря.
    • daemon: Указывает, является ли поток демоническим.

  2. start(self)：

    • Запустите поток. Он вызовет метод потокаrun()метод.

  3. run(self)：

    • Поток определяет код, который должен выполняться в этом методе.

  4. join(self, timeout=None)：

    • Ожидайте завершения потока. По умолчанию,join()он будет блокироваться до завершения вызванного потока. Еслиtimeoutпараметр указан, он ожидает не болееtimeoutсекунд.

  5. is_alive(self)：

    • Возвращает, выполняется ли поток. Если поток был запущен и ещё не завершён, возвращаетTrue, в противном случае он возвращаетFalse。

  6. getName(self)：

    • Возвращает имя потока.

  7. setName(self, name)：

    • Задаёт имя потока.

  8. identАтрибуты:

    • Уникальный идентификатор потока.

  9. daemonАтрибуты:

    • Флаг демона потока, используемый для указания, является ли он потоком-демоном.

  10. isDaemon()Методы:

Простой пример потока:

Пример

import threading
import time

def print_numbers():
    for i in range(5):
        time.sleep(1)
        print(i)

# Создать поток
thread = threading.Thread(target=print_numbers)

# Запустить поток
thread.start()

# Дождаться завершения потока
thread.join()

Результат вывода:

0
1
2
3
4

Создание потоков с помощью модуля threading

Мы можем создать новый подкласс, непосредственно наследуясь от threading.Thread, и после создания экземпляра вызвать метод start(), чтобы запустить новый поток. То есть он вызывает метод run() потока:

Пример

#!/usr/bin/python3

import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, delay):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.delay = delay
    def run(self):
        print ("Запуск потока:" + self.name)
        print_time(self.name, self.delay, 5)
        print ("Выход из потока:" + self.name)

def print_time(threadName, delay, counter):
    while counter:
        if exitFlag:
            threadName.exit()
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

# Создать новый поток
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# Запустить новый поток
thread1.start()
thread2.start()
thread1.join()
thread2.join()
print ("Выход из главного потока")

Результат выполнения вышеприведённой программы следующий;

开始线程：Thread-1
开始线程：Thread-2
Thread-1: Wed Jan  5 17:34:54 2022
Thread-2: Wed Jan  5 17:34:55 2022
Thread-1: Wed Jan  5 17:34:55 2022
Thread-1: Wed Jan  5 17:34:56 2022
Thread-2: Wed Jan  5 17:34:57 2022
Thread-1: Wed Jan  5 17:34:57 2022
Thread-1: Wed Jan  5 17:34:58 2022
退出线程：Thread-1
Thread-2: Wed Jan  5 17:34:59 2022
Thread-2: Wed Jan  5 17:35:01 2022
Thread-2: Wed Jan  5 17:35:03 2022
退出线程：Thread-2
退出主线程

Синхронизация потоков

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

Использование Lock и Rlock объектов Thread позволяет добиться простой синхронизации потоков. Оба объекта имеют методы acquire и release. Для данных, с которыми одновременно разрешено работать только одному потоку, операции можно поместить между методами acquire и release. Следующим образом:

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

Рассмотрим ситуацию: все элементы списка равны 0. Поток "set" изменяет все элементы с конца к началу на 1, а поток "print" читает список с начала к концу и выводит его.

Тогда возможно, что когда поток "set" начинает изменение, поток "print" приходит выводить список, и вывод становится наполовину из 0, наполовину из 1. Это рассинхронизация данных. Чтобы избежать этой ситуации, вводится понятие блокировок.

Блокировка имеет два состояния — заблокирована и разблокирована. Всякий раз, когда поток, например "set", хочет получить доступ к общим данным, он должен сначала захватить блокировку; если другой поток, например "print", уже захватил блокировку, то поток "set" приостанавливается, что представляет собой синхронное блокирование; после того как поток "print" завершит доступ и освободит блокировку, поток "set" продолжит работу.

При такой обработке при выводе списка выводятся либо все 0, либо все 1, и неловкая ситуация с половиной 0 и половиной 1 больше не возникнет.

Пример

#!/usr/bin/python3

import threading
import time

class myThread (threading.Thread):
    def __init__(self, threadID, name, delay):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.delay = delay
    def run(self):
        print ("Запуск потока: " + self.name)
        # Захватить блокировку для синхронизации потоков
        threadLock.acquire()
        print_time(self.name, self.delay, 3)
        # Освободить блокировку, чтобы запустить следующий поток
        threadLock.release()

def print_time(threadName, delay, counter):
    while counter:
        time.sleep(delay)
        print ("%s: %s" % (threadName, time.ctime(time.time())))
        counter -= 1

threadLock = threading.Lock()
threads = []

# Создать новый поток
thread1 = myThread(1, "Thread-1", 1)
thread2 = myThread(2, "Thread-2", 2)

# Запустить новый поток
thread1.start()
thread2.start()

# Добавить потоки в список потоков
threads.append(thread1)
threads.append(thread2)

# Дождаться завершения всех потоков
for t in threads:
    t.join()
print ("Выход из главного потока")

Выполните вышеуказанную программу, и результат вывода будет:

开启线程： Thread-1
开启线程： Thread-2
Thread-1: Wed Jan  5 17:36:50 2022
Thread-1: Wed Jan  5 17:36:51 2022
Thread-1: Wed Jan  5 17:36:52 2022
Thread-2: Wed Jan  5 17:36:54 2022
Thread-2: Wed Jan  5 17:36:56 2022
Thread-2: Wed Jan  5 17:36:58 2022
退出主线程

Очередь приоритетов потоков (Queue)

Модуль Queue в Python предоставляет синхронизированные потокобезопасные классы очередей, включая очередь FIFO (first-in, first-out) Queue, очередь LIFO (last-in, first-out) LifoQueue и очередь с приоритетом PriorityQueue.

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

Общие методы в модуле Queue:

  • Queue.qsize() возвращает размер очереди.
  • Queue.empty() Если очередь пуста, возвращает True; иначе False
  • Queue.full() Если очередь заполнена, возвращает True; иначе False
  • Queue.full соответствует размеру maxsize
  • Queue.get([block[, timeout]]) получает из очереди; timeout — время ожидания
  • Queue.get_nowait() Эквивалентно Queue.get(False)
  • Queue.put(item) записывает в очередь; timeout — время ожидания
  • Queue.put_nowait(item) Эквивалентно Queue.put(item, False)
  • Queue.task_done() После завершения задачи функция Queue.task_done() посылает очереди сигнал о том, что задача выполнена
  • Queue.join() Фактически означает ожидание, пока очередь не опустеет, после чего выполняются другие операции

Пример

#!/usr/bin/python3

import queue
import threading
import time

exitFlag = 0

class myThread (threading.Thread):
    def __init__(self, threadID, name, q):
        threading.Thread.__init__(self)
        self.threadID = threadID
        self.name = name
        self.q = q
    def run(self):
        print ("Запуск потока:" + self.name)
        process_data(self.name, self.q)
        print ("Выход из потока:" + self.name)

def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print ("%s обрабатывает %s" % (threadName, data))
        else:
            queueLock.release()
        time.sleep(1)

threadList = ["Поток-1", "Поток-2", "Поток-3"]
nameList = ["Один", "Два", "Три", "Четыре", "Пять"]
queueLock = threading.Lock()
workQueue = queue.Queue(10)
threads = []
threadID = 1

# Создание нового потока
for tName in threadList:
    thread = myThread(threadID, tName, workQueue)
    thread.start()
    threads.append(thread)
    threadID += 1

# Заполнение очереди
queueLock.acquire()
for word in nameList:
    workQueue.put(word)
queueLock.release()

# Ожидание опустошения очереди
while not workQueue.empty():
    pass

# Уведомление потока о необходимости завершения
exitFlag = 1

# Ожидание завершения всех потоков
for t in threads:
    t.join()
print ("Выход из главного потока")

Результат выполнения вышеуказанной программы:

开启线程：Thread-1
开启线程：Thread-2
开启线程：Thread-3
Thread-3 processing One
Thread-1 processing Two
Thread-2 processing Three
Thread-3 processing Four
Thread-1 processing Five
退出线程：Thread-3
退出线程：Thread-2
退出线程：Thread-1
退出主线程
Другие расширения