How do I use threading in Python?

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

С тех пор, как этот вопрос был задан в 2010 году, произошло реальное упрощение того, как выполнять простую многопоточность в Python с помощью map и pool.

Приведенный ниже код взят из статьи / поста в блоге, который вам обязательно следует просмотреть (без указания принадлежности) - Параллелизм в одной строке: лучшая модель для повседневных задач с потоковой обработкой. Я подведу итог ниже - в итоге получается всего несколько строк кода:

from multiprocessing.dummy import Pool as ThreadPool
pool = ThreadPool(4)
results = pool.map(my_function, my_array)

Какая многопоточная версия:

results = []
for item in my_array:
    results.append(my_function(item))

Описание

Map - это классная маленькая функция и ключ к простому внедрению параллелизма в ваш код Python. Для тех, кто не знаком, map - это нечто, заимствованное из функциональных языков, таких как Lisp. Это функция, которая отображает другую функцию поверх последовательности.

Map обрабатывает итерацию последовательности за нас, применяет функцию и сохраняет все результаты в удобном списке в конце.

Реализация

Параллельные версии функции map предоставляются двумя библиотеками: multiprocessing, а также ее малоизвестным, но не менее фантастическим дочерним элементом: multiprocessing.dummy.

multiprocessing.dummy точно такой же, как модуль многопроцессорной обработки, но вместо этого использует потоки (важное различие - используйте несколько процессов для задач с интенсивным использованием ЦП; потоки для (и во время) ввода-вывода):

многопроцессорность.dummy копирует API многопроцессорной обработки, но является не более чем оболочкой вокруг модуля многопоточности.

import urllib2
from multiprocessing.dummy import Pool as ThreadPool

urls = [
  'http://www.python.org',
  'http://www.python.org/about/',
  'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html',
  'http://www.python.org/doc/',
  'http://www.python.org/download/',
  'http://www.python.org/getit/',
  'http://www.python.org/community/',
  'https://wiki.python.org/moin/',
]

# Make the Pool of workers
pool = ThreadPool(4)

# Open the URLs in their own threads
# and return the results
results = pool.map(urllib2.urlopen, urls)

# Close the pool and wait for the work to finish
pool.close()
pool.join()

И временные результаты:

Single thread:   14.4 seconds
       4 Pool:   3.1 seconds
       8 Pool:   1.4 seconds
      13 Pool:   1.3 seconds

Передача нескольких аргументов (работает подобным образом только в Python 3.3 и более поздних версиях):

Для передачи нескольких массивов:

results = pool.starmap(function, zip(list_a, list_b))

Или для передачи константы и массива:

results = pool.starmap(function, zip(itertools.repeat(constant), list_a))

Если вы используете более раннюю версию Python, вы можете передать несколько аргументов с помощью этого обходного пути).

(Спасибо пользователю 136036 за полезный комментарий.)

Here's a simple example: you need to try a few alternative URLs and return the contents of the first one to respond.

import Queue
import threading
import urllib2

# Called by each thread
def get_url(q, url):
    q.put(urllib2.urlopen(url).read())

theurls = ["http://google.com", "http://yahoo.com"]

q = Queue.Queue()

for u in theurls:
    t = threading.Thread(target=get_url, args = (q,u))
    t.daemon = True
    t.start()

s = q.get()
print s

Это случай, когда потоковая обработка используется как простая оптимизация: каждый подпоток ожидает разрешения URL-адреса и ответа, чтобы поместить его содержимое в очередь; каждый поток является демоном (не будет поддерживать процесс, если основной поток завершается - это чаще, чем нет); основной поток запускает все подпотоки, выполняет get в очереди, чтобы дождаться, пока один из них не выполнит put, затем выдает результаты и завершается (что отключает любые подпотоки, которые могли бы все еще работают, поскольку это потоки демонов).

Правильное использование потоков в Python неизменно связано с операциями ввода-вывода (поскольку CPython в любом случае не использует несколько ядер для выполнения задач, связанных с ЦП, единственная причина потоковой обработки - это не блокировка процесса во время ожидания некоторого ввода-вывода). Очереди почти всегда являются лучшим способом передачи работы потокам и / или сбора результатов работы, кстати, и они по сути потокобезопасны, поэтому избавляют вас от беспокойства о блокировках, условиях, событиях, семафорах и других концепциях межпотоковой координации / коммуникации.

ПРИМЕЧАНИЕ: Для фактического распараллеливания в Python вам следует использовать модуль многопроцессорности для разветвления нескольких процессов, которые выполняются параллельно (из-за глобальной блокировки интерпретатора потоки Python обеспечивают чередование, но фактически они выполняются последовательно, а не параллельно, и полезны только при чередовании операций ввода-вывода).

Однако, если вы просто ищете чередование (или выполняете операции ввода-вывода, которые могут быть распараллелены, несмотря на глобальную блокировку интерпретатора), то начать стоит с модуля многопоточность. В качестве действительно простого примера давайте рассмотрим проблему суммирования большого диапазона путем параллельного суммирования поддиапазонов:

import threading

class SummingThread(threading.Thread):
     def __init__(self,low,high):
         super(SummingThread, self).__init__()
         self.low=low
         self.high=high
         self.total=0

     def run(self):
         for i in range(self.low,self.high):
             self.total+=i


thread1 = SummingThread(0,500000)
thread2 = SummingThread(500000,1000000)
thread1.start() # This actually causes the thread to run
thread2.start()
thread1.join()  # This waits until the thread has completed
thread2.join()
# At this point, both threads have completed
result = thread1.total + thread2.total
print result

Обратите внимание, что приведенный выше пример является очень глупым, поскольку он не выполняет абсолютно никакого ввода-вывода и будет выполняться последовательно, хотя и с чередованием (с дополнительными накладными расходами на переключение контекста) в CPython из-за глобальной блокировки интерпретатора.

Как и другие упомянутые, CPython может использовать потоки только для ожидания ввода-вывода из-за GIL.

Если вы хотите использовать несколько ядер для задач, связанных с ЦП, используйте многопроцессорность:

from multiprocessing import Process

def f(name):
    print 'hello', name

if __name__ == '__main__':
    p = Process(target=f, args=('bob',))
    p.start()
    p.join()