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

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

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

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

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

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

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

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

В Python есть два способа использования потоков: с помощью функций или с помощью класса, оборачивающего объекты потоков.

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

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

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

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

Пример (Python 2.0+)

#!/usr/bin/python # -*- coding: UTF-8 -*- 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: Thu Jan 22 15:42:17 2009
Thread-1: Thu Jan 22 15:42:19 2009
Thread-2: Thu Jan 22 15:42:19 2009
Thread-1: Thu Jan 22 15:42:21 2009
Thread-2: Thu Jan 22 15:42:23 2009
Thread-1: Thu Jan 22 15:42:23 2009
Thread-1: Thu Jan 22 15:42:25 2009
Thread-2: Thu Jan 22 15:42:27 2009
Thread-2: Thu Jan 22 15:42:31 2009
Thread-2: Thu Jan 22 15:42:35 2009

Завершение потока обычно происходит при естественном завершении функции потока; также можно вызвать thread.exit() в функции потока, что вызывает исключение SystemExit для достижения цели выхода из потока.

Модуль thread

Python поддерживает потоки через две стандартные библиотеки: thread и threading. thread предоставляет низкоуровневые примитивные потоки и простую блокировку.

Другие методы, предоставляемые модулем threading:

  • threading.currentThread(): возвращает переменную текущего потока.
  • threading.enumerate(): возвращает список, содержащий работающие потоки. Работающие означает после запуска потока и до его завершения, не включая потоки до запуска и после завершения.
  • threading.activeCount(): возвращает количество работающих потоков, тот же результат, что и len(threading.enumerate()).

Помимо использования методов, модуль thread также предоставляет класс Thread для работы с потоками. Класс Thread предоставляет следующие методы:

  • run():Метод, используемый для представления активности потока.
  • start():Запускает активность потока.
  • join([time]):Ожидает завершения потока. Это блокирует вызывающий поток до тех пор, пока поток, для которого вызван метод join(), не завершится — при нормальном выходе или необработанном исключении — либо не истечёт необязательный тайм-аут.
  • isAlive():Возвращает, активен ли поток.
  • getName():Возвращает имя потока.
  • setName():Задаёт имя потока.

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

Чтобы создать поток с помощью модуля Threading, напрямую унаследуйте от threading.Thread, а затем переопределите метод __init__ и метод run:

Пример (Python 2.0+)

#!/usr/bin/python # -*- coding: UTF-8 -*- import threading import time exitFlag = 0 class myThread (threading.Thread): # Наследуемся от родительского класса threading.Thread def __init__(self, threadID, name, counter): threading.Thread.__init__(self) self.threadID = threadID self.name = name self.counter = counter def run(self): # Запишите код, который должен выполняться, в функцию run. Поток запустит функцию run сразу после создания. print "Запуск " + self.name print_time(self.name, self.counter, 5) print "Завершение " + self.name def print_time(threadName, delay, counter): while counter: if exitFlag: (threading.Thread).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() print "Выход из главного потока"

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

Starting Thread-1
Starting Thread-2
Exiting Main Thread
Thread-1: Thu Mar 21 09:10:03 2013
Thread-1: Thu Mar 21 09:10:04 2013
Thread-2: Thu Mar 21 09:10:04 2013
Thread-1: Thu Mar 21 09:10:05 2013
Thread-1: Thu Mar 21 09:10:06 2013
Thread-2: Thu Mar 21 09:10:06 2013
Thread-1: Thu Mar 21 09:10:07 2013
Exiting Thread-1
Thread-2: Thu Mar 21 09:10:08 2013
Thread-2: Thu Mar 21 09:10:10 2013
Thread-2: Thu Mar 21 09:10:12 2013
Exiting 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 больше не возникнет.

Пример (Python 2.0+)

#!/usr/bin/python # -*- coding: UTF-8 -*- import threading import time class myThread (threading.Thread): def __init__(self, threadID, name, counter): threading.Thread.__init__(self) self.threadID = threadID self.name = name self.counter = counter def run(self): print "Запуск " + self.name # Захватить блокировку; вернуть True после успешного захвата блокировки # Если необязательный параметр timeout не задан, он будет блокироваться до тех пор, пока блокировка не будет захвачена # В противном случае он вернёт False по истечении timeout threadLock.acquire() print_time(self.name, self.counter, 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 "Выход из главного потока"

Очередь с приоритетом потоков

Модуль Queue в Python предоставляет синхронизированные, потокобезопасные классы очередей, включая очередь FIFO (первым пришёл — первым ушёл) Queue, очередь LIFO (последним пришёл — первым ушёл) 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, block=True, timeout=None) Записывает в очередь, timeout — время ожидания
  • Queue.put_nowait(item) Эквивалентно Queue.put(item, False)
  • Queue.task_done() После завершения работы функция Queue.task_done() отправляет в очередь сигнал о том, что задача завершена.
  • Queue.join() Фактически означает ожидание, пока очередь не опустеет, прежде чем выполнять другие операции.

Пример (Python 2.0+)

#!/usr/bin/python # -*- coding: UTF-8 -*- 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 = ["Thread-1", "Thread-2", "Thread-3"] nameList = ["Один", "Два", "Три", "Четыре", "Five"] 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 "Выход из главного потока"

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

Starting Thread-1
Starting Thread-2
Starting Thread-3
Thread-1 processing One
Thread-2 processing Two
Thread-3 processing Three
Thread-1 processing Four
Thread-2 processing Five
Exiting Thread-3
Exiting Thread-1
Exiting Thread-2
Exiting Main Thread
Другие расширения