python ejemplos - ¿Cómo utilizar hilos en Python?




9 Answers

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

El código a continuación proviene de un artículo / publicación de blog que definitivamente debe 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 levantado 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.

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 envoltorio alrededor del módulo de subprocesamiento.

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 user136036 por el comentario útil)

thread compartir

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?




NOTA : para la paralelización real en Python, debe usar el módulo de pool para dividir varios procesos que se ejecutan en paralelo (debido al bloqueo 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 intercalado (o está realizando operaciones de E / S que pueden paralelizarse a pesar del bloqueo global del intérprete), entonces el módulo de threading es el lugar para comenzar. Como un ejemplo realmente simple, consideremos el problema de sumar un rango grande sumando 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 bloqueo global del intérprete.




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

Este es el ejemplo más simple que puedo 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"



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



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)




Aquí está el ejemplo muy simple de 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) 



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



import threading
import requests

def send():

  r = requests.get('https://www.stackoverlow.com')

thread = []
t = threading.Thread(target=send())
thread.append(t)
t.start()



Related

python multithreading