Многопоточное программирование в Python: эффективные техники параллелизма

Преимущества потоков

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

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

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

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

Запуск нового потока (модуль `thread`)

Для порождения другого потока нужно вызвать метод из модуля thread:

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

Этот вызов обеспечивает быстрый и эффективный способ создания новых потоков в Linux и Windows.

Вызов возвращается немедленно, и дочерний поток запускается и вызывает function с переданным списком args. Когда функция возвращается, поток завершается.

args — кортеж аргументов; используйте пустой кортеж для вызова без аргументов. kwargs — опциональный словарь именованных аргументов.

Пример

#!/usr/bin/python
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("Error: unable to start thread")

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 эффективен для низкоуровневого трединга, он очень ограничен по сравнению с более новым модулем threading.

Модуль `threading`

Новый модуль threading, включенный в Python 2.4, предоставляет гораздо более мощную высокоуровневую поддержку потоков, чем модуль thread.

Модуль threading раскрывает все методы модуля thread и предоставляет дополнительные:

threading.activeCount() — возвращает количество активных объектов потоков.
threading.currentThread() — возвращает объект потока вызывающего кода.
threading.enumerate() — возвращает список всех активных объектов потоков.

Помимо методов, модуль threading имеет класс Thread, реализующий потоки. Методы класса Thread:

run() — точка входа для потока.
start() — запускает поток, вызывая метод run.
join([time]) — ждет завершения потоков.
isAlive() — проверяет, выполняется ли поток еще.
getName() — возвращает имя потока.
setName() — устанавливает имя потока.

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

Для реализации нового потока с использованием модуля threading нужно:

Определить новый подкласс класса Thread.
Переопределить метод __init__(self[, args]) для добавления дополнительных аргументов.
Переопределить метод run(self[, args]) для реализации того, что должен делать поток при запуске.

После создания подкласса Thread можно создать его экземпляр и запустить новый поток, вызвав start(), который в свою очередь вызывает run().

Пример

#!/usr/bin/python
import threading
import time

exitFlag = 0

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("Starting " + self.name)
        print_time(self.name, 5, self.counter)
        print("Exiting " + self.name)

def print_time(threadName, counter, delay):
    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()
print("Exiting Main Thread")

Результат (пример):

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

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

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

Метод acquire(blocking) нового объекта блокировки используется для принудительного синхронного выполнения потоков. Опциональный параметр blocking управляет тем, ждет ли поток получения блокировки.

Если blocking = 0, поток немедленно возвращает 0, если блокировка не получена, и 1, если получена.
Если blocking = 1, поток блокируется и ждет освобождения блокировки.

Метод release() объекта блокировки используется для освобождения блокировки, когда она больше не нужна.

Пример

#!/usr/bin/python
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("Starting " + self.name)
        # Получить блокировку для синхронизации
        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("Exiting Main Thread")

Результат:

Starting Thread-1
Starting Thread-2
Thread-1: Thu Mar 21 09:11:28 2013
Thread-1: Thu Mar 21 09:11:29 2013
Thread-1: Thu Mar 21 09:11:30 2013
Thread-2: Thu Mar 21 09:11:32 2013
Thread-2: Thu Mar 21 09:11:34 2013
Thread-2: Thu Mar 21 09:11:36 2013
Exiting Main Thread

Многопоточная очередь с приоритетом (Queue)

Модуль Queue позволяет создать новый объект очереди, который может содержать определенное количество элементов. Методы управления очередью:

get() — удаляет и возвращает элемент из очереди.
put() — добавляет элемент в очередь.
qsize() — возвращает количество элементов в очереди.
empty() — возвращает True, если очередь пуста, иначе False.
full() — возвращает True, если очередь полна, иначе False.

Пример

#!/usr/bin/python
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("Starting " + self.name)
        process_data(self.name, self.q)
        print("Exiting " + self.name)

def process_data(threadName, q):
    while not exitFlag:
        queueLock.acquire()
        if not workQueue.empty():
            data = q.get()
            queueLock.release()
            print("%s processing %s" % (threadName, data))
        else:
            queueLock.release()
            time.sleep(1)

threadList = ["Thread-1", "Thread-2", "Thread-3"]
nameList = ["One", "Two", "Three", "Four", "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("Exiting Main Thread")

Результат:

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

Отправка email в Python через SMTP: простое и быстрое руководство Обработка XML в Python: руководство по управлению взаимодействующими данными

Python