¿Cómo utilizar hilos en Python?

Estoy tratando de entender los hilos en Python. He mirado la documentación y los ejemplos, pero francamente, muchos ejemplos son demasiado sofisticados y tengo problemas para entenderlos.

¿Cómo se muestran claramente las tareas que se dividen para subprocesos múltiples?

Desde que se hizo esta pregunta en 2010, ha habido una simplificación real en cómo hacer multihilo simple con python con mapa y grupo .

El código a continuación proviene de un artículo / publicación de blog que definitivamente debería revisar (sin afiliación) – Paralelismo en una línea: un modelo mejor para las tareas diarias de subprocesamiento . Resumiré a continuación: termina siendo solo unas pocas líneas de código:

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

¿Cuál es la versión multiproceso de:

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

Descripción

El mapa es una función pequeña y genial, y la clave para inyectar fácilmente paralelismo en su código Python. Para aquellos que no están familiarizados, el mapa es algo que se levanta de lenguajes funcionales como Lisp. Es una función que mapea otra función sobre una secuencia.

El mapa maneja la iteración sobre la secuencia para nosotros, aplica la función y almacena todos los resultados en una lista práctica al final.

introduzca la descripción de la imagen aquí


Implementación

Las versiones paralelas de la función de mapa son proporcionadas por dos bibliotecas: multiprocesamiento, y también su hijastro poco conocido, pero igualmente fantástico: multiprocessing.dummy.

multiprocessing.dummy es exactamente lo mismo que el módulo multiprocessing, pero en su lugar usa subprocesos ( una distinción importante : use procesos múltiples para tareas que requieren un uso intensivo de la CPU; subprocesos para (y durante) IO ):

multiprocessing.dummy replica la API de multiproceso, pero no es más que un contenedor alrededor del módulo de subprocesos.

 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() 

Y los resultados del tiempo:

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

Pasar múltiples argumentos (funciona así solo en Python 3.3 y versiones posteriores ):

Para pasar múltiples matrices:

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

o pasar una constante y una matriz:

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

Si está utilizando una versión anterior de Python, puede pasar varios argumentos a través de esta solución .

(Gracias al usuario 136036 por el comentario útil)

Aquí hay un ejemplo simple: necesita probar algunas URL alternativas y devolver el contenido de la primera para responder.

 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 

Este es un caso en el que los subprocesos se utilizan como una optimización simple: cada subhilo está esperando que una URL se resuelva y responda, para poner su contenido en la cola; cada subproceso es un demonio (no mantendrá el proceso activo si el subproceso principal finaliza, eso es más común que no); el hilo principal inicia todos los subprocesos, se pone en la cola para esperar hasta que uno de ellos haga un put , luego emite los resultados y termina (lo que elimina cualquier subproceso que aún esté en ejecución, ya que son hilos de daemon).

El uso correcto de los subprocesos en Python está invariablemente conectado a las operaciones de E / S (ya que CPython no usa múltiples núcleos para ejecutar tareas vinculadas a la CPU de todos modos, la única razón para el subprocesamiento no es el locking del proceso mientras que hay que esperar para algunas E / S ). Las colas son casi invariablemente la mejor manera de agrupar el trabajo en subprocesos y / o recostackr los resultados del trabajo, por cierto, y son intrínsecamente seguras para hilos, por lo que evitan que se preocupe por los lockings, las condiciones, los eventos, los semáforos y otras interrelaciones. Conceptos de coordinación / comunicación de hilos.

NOTA : para la paralelización real en Python, debe usar el módulo de multiprocesamiento para dividir varios procesos que se ejecutan en paralelo (debido al locking global del intérprete, los hilos de Python proporcionan intercalación pero, de hecho, se ejecutan en serie, no en paralelo, y solo son útiles cuando entrelazando operaciones de E / S).

Sin embargo, si simplemente está buscando intercalar (o está realizando operaciones de E / S que se pueden paralelizar a pesar del locking global del intérprete), entonces el módulo de subprocesamiento es el lugar para comenzar. Como un ejemplo realmente simple, consideremos el problema de sumr un rango grande al sumr subintervalos en paralelo:

 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 

Tenga en cuenta que lo anterior es un ejemplo muy estúpido, ya que no hace absolutamente ninguna E / S y se ejecutará en serie aunque intercalado (con la sobrecarga agregada de cambio de contexto) en CPython debido al locking global del intérprete.

Como mencionaron otros, CPython puede usar subprocesos solo para I \ O espera debido a GIL. Si desea beneficiarse de varios núcleos para tareas vinculadas a la CPU, utilice el multiprocesamiento :

 from multiprocessing import Process def f(name): print 'hello', name if __name__ == '__main__': p = Process(target=f, args=('bob',)) p.start() p.join() 

Solo una nota, la cola no es necesaria para el enhebrado.

Este es el ejemplo más simple que podría imaginar que muestra 10 procesos que se ejecutan simultáneamente.

 import threading from random import randint from time import sleep def print_number(number): # Sleeps a random 1 to 10 seconds rand_int_var = randint(1, 10) sleep(rand_int_var) print "Thread " + str(number) + " slept for " + str(rand_int_var) + " seconds" thread_list = [] for i in range(1, 10): # Instantiates the thread # (i) does not make a sequence, so (i,) t = threading.Thread(target=print_number, args=(i,)) # Sticks the thread in a list so that it remains accessible thread_list.append(t) # Starts threads for thread in thread_list: thread.start() # This blocks the calling thread until the thread whose join() method is called is terminated. # From http://docs.python.org/2/library/threading.html#thread-objects for thread in thread_list: thread.join() # Demonstrates that the main process waited for threads to complete print "Done" 

La respuesta de Alex Martelli me ayudó, sin embargo, aquí está la versión modificada que pensé que era más útil (al menos para mí).

Actualizado: funciona tanto en python2 como en python3.

 try: # for python3 import queue from urllib.request import urlopen except: # for python2 import Queue as queue from urllib2 import urlopen import threading worker_data = ['http://google.com', 'http://yahoo.com', 'http://bing.com'] #load up a queue with your data, this will handle locking q = queue.Queue() for url in worker_data: q.put(url) #define a worker function def worker(url_queue): queue_full = True while queue_full: try: #get your data off the queue, and do some work url = url_queue.get(False) data = urlopen(url).read() print(len(data)) except queue.Empty: queue_full = False #create as many threads as you want thread_count = 5 for i in range(thread_count): t = threading.Thread(target=worker, args = (q,)) t.start() 

Encontré esto muy útil: cree tantos hilos como núcleos y déjelos ejecutar una gran cantidad de tareas (en este caso, llamar a un progtwig shell):

 import Queue import threading import multiprocessing import subprocess q = Queue.Queue() for i in range(30): #put 30 tasks in the queue q.put(i) def worker(): while True: item = q.get() #execute a task: call a shell program and wait until it completes subprocess.call("echo "+str(item), shell=True) q.task_done() cpus=multiprocessing.cpu_count() #detect number of cores print("Creating %d threads" % cpus) for i in range(cpus): t = threading.Thread(target=worker) t.daemon = True t.start() q.join() #block until all tasks are done 

Dada una función, f , hágala así:

 import threading threading.Thread(target=f).start() 

Pasar argumentos a f

 threading.Thread(target=f, args=(a,b,c)).start() 

Python 3 tiene la facilidad de lanzar tareas paralelas . Esto facilita nuestro trabajo.

Tiene para agrupación de hilos y agrupación de procesos .

Lo siguiente da una idea:

Ejemplo de ThreadPoolExecutor

 import concurrent.futures import urllib.request URLS = ['http://www.foxnews.com/', 'http://www.cnn.com/', 'http://europe.wsj.com/', 'http://www.bbc.co.uk/', 'http://some-made-up-domain.com/'] # Retrieve a single page and report the URL and contents def load_url(url, timeout): with urllib.request.urlopen(url, timeout=timeout) as conn: return conn.read() # We can use a with statement to ensure threads are cleaned up promptly with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: # Start the load operations and mark each future with its URL future_to_url = {executor.submit(load_url, url, 60): url for url in URLS} for future in concurrent.futures.as_completed(future_to_url): url = future_to_url[future] try: data = future.result() except Exception as exc: print('%r generated an exception: %s' % (url, exc)) else: print('%r page is %d bytes' % (url, len(data))) 

ProcessPoolExecutor

 import concurrent.futures import math PRIMES = [ 112272535095293, 112582705942171, 112272535095293, 115280095190773, 115797848077099, 1099726899285419] def is_prime(n): if n % 2 == 0: return False sqrt_n = int(math.floor(math.sqrt(n))) for i in range(3, sqrt_n + 1, 2): if n % i == 0: return False return True def main(): with concurrent.futures.ProcessPoolExecutor() as executor: for number, prime in zip(PRIMES, executor.map(is_prime, PRIMES)): print('%d is prime: %s' % (number, prime)) if __name__ == '__main__': main() 

Para mí, el ejemplo perfecto para Threading es monitorear eventos asíncronos. Mira este código.

 # thread_test.py import threading import time class Monitor(threading.Thread): def __init__(self, mon): threading.Thread.__init__(self) self.mon = mon def run(self): while True: if self.mon[0] == 2: print "Mon = 2" self.mon[0] = 3; 

Puedes jugar con este código abriendo una sesión de IPython y haciendo algo como:

 >>>from thread_test import Monitor >>>a = [0] >>>mon = Monitor(a) >>>mon.start() >>>a[0] = 2 Mon = 2 >>>a[0] = 2 Mon = 2 

Espera unos minutos

 >>>a[0] = 2 Mon = 2 

Usando el flamante nuevo módulo concurrent.futures

 def sqr(val): import time time.sleep(0.1) return val * val def process_result(result): print(result) def process_these_asap(tasks): import concurrent.futures with concurrent.futures.ProcessPoolExecutor() as executor: futures = [] for task in tasks: futures.append(executor.submit(sqr, task)) for future in concurrent.futures.as_completed(futures): process_result(future.result()) # Or instead of all this just do: # results = executor.map(sqr, tasks) # list(map(process_result, results)) def main(): tasks = list(range(10)) print('Processing {} tasks'.format(len(tasks))) process_these_asap(tasks) print('Done') return 0 if __name__ == '__main__': import sys sys.exit(main()) 

El enfoque del ejecutor puede parecer familiar para todos aquellos que se han ensuciado las manos con Java antes.

También en una nota: para mantener el universo sano, no olvides cerrar tus pools / ejecutores si no lo usas with contexto (lo cual es tan increíble que lo hace por ti)

La mayoría de las documentaciones y tutoriales utilizan el módulo de Threading y Queue Python que pueden parecer abrumadores para los principiantes.

Quizás considere el módulo concurrent.futures.ThreadPoolExecutor de python 3. Combinado con la comprensión de la cláusula y la lista, podría ser un verdadero encanto.

 from concurrent.futures import ThreadPoolExecutor, as_completed def get_url(url): # Your actual program here. Using threading.Lock() if necessary return "" # List of urls to fetch urls = ["url1", "url2"] with ThreadPoolExecutor(max_workers = 5) as executor: # Create threads futures = {executor.submit(get_url, url) for url in urls} # as_completed() gives you the threads once finished for f in as_completed(futures): # Get the results rs = f.result() 

Aquí está el ejemplo muy simple de la importación CSV utilizando subprocesos. [La inclusión en la biblioteca puede diferir para diferentes propósitos]

Funciones de ayuda:

 from threading import Thread from project import app import csv def import_handler(csv_file_name): thr = Thread(target=dump_async_csv_data, args=[csv_file_name]) thr.start() def dump_async_csv_data(csv_file_name): with app.app_context(): with open(csv_file_name) as File: reader = csv.DictReader(File) for row in reader: #DB operation/query 

Función del conductor:

 import_handler(csv_file_name) 

Hilo múltiple con un ejemplo simple que será útil. Puede ejecutarlo y entender fácilmente cómo funciona el subproceso múltiple en Python. Utilicé el locking para evitar acceder a otro hilo hasta que los hilos anteriores terminaron su trabajo. Por el uso de

tLock = threading.BoundedSemaphore (value = 4)

En esta línea de código, puede permitir la cantidad de procesos a la vez y mantener el rest del hilo que se ejecutará más tarde o después de que hayan finalizado los procesos anteriores.

 import threading import time #tLock = threading.Lock() tLock = threading.BoundedSemaphore(value=4) def timer(name, delay, repeat): print "\r\nTimer: ", name, " Started" tLock.acquire() print "\r\n", name, " has the acquired the lock" while repeat > 0: time.sleep(delay) print "\r\n", name, ": ", str(time.ctime(time.time())) repeat -= 1 print "\r\n", name, " is releaseing the lock" tLock.release() print "\r\nTimer: ", name, " Completed" def Main(): t1 = threading.Thread(target=timer, args=("Timer1", 2, 5)) t2 = threading.Thread(target=timer, args=("Timer2", 3, 5)) t3 = threading.Thread(target=timer, args=("Timer3", 4, 5)) t4 = threading.Thread(target=timer, args=("Timer4", 5, 5)) t5 = threading.Thread(target=timer, args=("Timer5", 0.1, 5)) t1.start() t2.start() t3.start() t4.start() t5.start() print "\r\nMain Complete" if __name__ == "__main__": Main() 

Vi muchos ejemplos aquí donde no se estaba realizando ningún trabajo real + en su mayoría estaban vinculados a la CPU. Este es un ejemplo de una tarea vinculada a la CPU que calcula todos los números primos entre 10 millones y 10.05 millones. He usado los 4 métodos aquí

 import math import timeit import threading import multiprocessing from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor def time_stuff(fn): """ Measure time of execution of a function """ def wrapper(*args, **kwargs): t0 = timeit.default_timer() fn(*args, **kwargs) t1 = timeit.default_timer() print("{} seconds".format(t1 - t0)) return wrapper def find_primes_in(nmin, nmax): """ Compute a list of prime numbers between the given minimum and maximum arguments """ primes = [] #Loop from minimum to maximum for current in range(nmin, nmax + 1): #Take the square root of the current number sqrt_n = int(math.sqrt(current)) found = False #Check if the any number from 2 to the square root + 1 divides the current numnber under consideration for number in range(2, sqrt_n + 1): #If divisible we have found a factor, hence this is not a prime number, lets move to the next one if current % number == 0: found = True break #If not divisible, add this number to the list of primes that we have found so far if not found: primes.append(current) #I am merely printing the length of the array containing all the primes but feel free to do what you want print(len(primes)) @time_stuff def sequential_prime_finder(nmin, nmax): """ Use the main process and main thread to compute everything in this case """ find_primes_in(nmin, nmax) @time_stuff def threading_prime_finder(nmin, nmax): """ If the minimum is 1000 and the maximum is 2000 and we have 4 workers 1000 - 1250 to worker 1 1250 - 1500 to worker 2 1500 - 1750 to worker 3 1750 - 2000 to worker 4 so lets split the min and max values according to the number of workers """ nrange = nmax - nmin threads = [] for i in range(8): start = int(nmin + i * nrange/8) end = int(nmin + (i + 1) * nrange/8) #Start the thrread with the min and max split up to compute #Parallel computation will not work here due to GIL since this is a CPU bound task t = threading.Thread(target = find_primes_in, args = (start, end)) threads.append(t) t.start() #Dont forget to wait for the threads to finish for t in threads: t.join() @time_stuff def processing_prime_finder(nmin, nmax): """ Split the min, max interval similar to the threading method above but use processes this time """ nrange = nmax - nmin processes = [] for i in range(8): start = int(nmin + i * nrange/8) end = int(nmin + (i + 1) * nrange/8) p = multiprocessing.Process(target = find_primes_in, args = (start, end)) processes.append(p) p.start() for p in processes: p.join() @time_stuff def thread_executor_prime_finder(nmin, nmax): """ Split the min max interval similar to the threading method but use thread pool executor this time This method is slightly faster than using pure threading as the pools manage threads more efficiently This method is still slow due to the GIL limitations since we are doing a CPU bound task """ nrange = nmax - nmin with ThreadPoolExecutor(max_workers = 8) as e: for i in range(8): start = int(nmin + i * nrange/8) end = int(nmin + (i + 1) * nrange/8) e.submit(find_primes_in, start, end) @time_stuff def process_executor_prime_finder(nmin, nmax): """ Split the min max interval similar to the threading method but use the process pool executor This is the fastest method recorded so far as it manages process efficiently + overcomes GIL limitations RECOMMENDED METHOD FOR CPU BOUND TASKS """ nrange = nmax - nmin with ProcessPoolExecutor(max_workers = 8) as e: for i in range(8): start = int(nmin + i * nrange/8) end = int(nmin + (i + 1) * nrange/8) e.submit(find_primes_in, start, end) def main(): nmin = int(1e7) nmax = int(1.05e7) print("Sequential Prime Finder Starting") sequential_prime_finder(nmin, nmax) print("Threading Prime Finder Starting") threading_prime_finder(nmin, nmax) print("Processing Prime Finder Starting") processing_prime_finder(nmin, nmax) print("Thread Executor Prime Finder Starting") thread_executor_prime_finder(nmin, nmax) print("Process Executor Finder Starting") process_executor_prime_finder(nmin, nmax) main() 

Aquí están los resultados en mi máquina central Mac OSX 4

 Sequential Prime Finder Starting 9.708213827005238 seconds Threading Prime Finder Starting 9.81836523200036 seconds Processing Prime Finder Starting 3.2467174359990167 seconds Thread Executor Prime Finder Starting 10.228896902000997 seconds Process Executor Finder Starting 2.656402041000547 seconds 

Ninguna de las soluciones anteriores realmente usó múltiples núcleos en mi servidor GNU / Linux (donde no tengo derechos de administrador). Simplemente corrieron en un solo núcleo. Utilicé la interfaz os.fork nivel inferior para generar múltiples procesos. Este es el código que me funcionó:

 from os import fork values = ['different', 'values', 'for', 'threads'] for i in range(len(values)): p = fork() if p == 0: my_function(values[i]) break 
 import threading import requests def send(): r = requests.get('https://www.stackoverlow.com') thread = [] t = threading.Thread(target=send()) thread.append(t) t.start()