¿Cómo puedo usar threading en Python?

Estoy tratando de entender el enhebrado en Python. He visto la documentación y los ejemplos, pero, francamente, muchos ejemplos son demasiado sofisticados y tengo problemas para comprenderlos.

¿Cómo muestra claramente las tareas divididas para subprocesos múltiples?

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

El siguiente código proviene de un artículo / blog que definitivamente debe consultar (sin afiliación) - Paralelismo en una línea: un mejor modelo para las tareas de subprocesamiento día a día . 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 paralelismo fácilmente en su código Python. Para aquellos que no están familiarizados, el mapa es algo sacado de lenguajes funcionales como Lisp. Es una función que mapea otra función sobre una secuencia.

Map 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

Dos bibliotecas proporcionan las versiones paralelas de la función de mapa: multiprocesamiento, y también su hijastro poco conocido pero igualmente fantástico: multiprocesamiento.dummy.

multiprocessing.dummyes exactamente lo mismo que el módulo de multiprocesamiento, pero usa subprocesos en su lugar ( una distinción importante : use múltiples procesos para tareas intensivas de CPU; subprocesos para (y durante) E / S ):

multiprocessing.dummy replica la API de multiprocesamiento, 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 momento:

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 para 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 alternativa ).

(Gracias al usuario 136036 por el útil comentario).

Aquí hay un ejemplo simple: debe 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 el subproceso se utiliza como una optimización simple: cada subproceso está esperando que una URL se resuelva y responda, para poner su contenido en la cola; cada hilo es un demonio (no mantendrá el proceso si el hilo principal termina, eso es más común que no); el subproceso principal inicia todos los subprocesos, hace un geten la cola para esperar hasta que uno de ellos haya hecho un a put, luego emite los resultados y finaliza (lo que elimina todos los subprocesos que aún puedan estar en ejecución, ya que son subprocesos de daemon).

El uso adecuado de los subprocesos en Python está invariablemente conectado a las operaciones de E / S (dado que CPython no usa múltiples núcleos para ejecutar tareas vinculadas a la CPU de todos modos, la única razón para el subproceso no es bloquear el proceso mientras hay que esperar algunas E / S ) Las colas son casi invariablemente la mejor manera de agilizar el trabajo en subprocesos y / o recopilar los resultados del trabajo, por cierto, y son intrínsecamente seguros, por lo que evitan que se preocupe por bloqueos, condiciones, eventos, semáforos y otros inter -proceso de coordinación / conceptos de comunicación.

NOTA : Para la paralelización real en Python, debe usar el módulo de multiprocesamiento para bifurcar múltiples procesos que se ejecutan en paralelo (debido al bloqueo global del intérprete, los hilos de Python proporcionan intercalado, pero de hecho se ejecutan en serie, no en paralelo, y solo son útil cuando se intercalan operaciones de E / S).

Sin embargo, si simplemente está buscando entrelazado (o está haciendo operaciones de E / S que pueden ser paralelas a pesar del bloqueo global del intérprete), entonces el módulo de subprocesos es el lugar para comenzar. Como un ejemplo realmente simple, consideremos el problema de sumar un amplio rango sumando subranges 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 nada de E / S y se ejecutará en serie aunque intercalado (con la sobrecarga adicional de cambio de contexto) en CPython debido al bloqueo global del intérprete.

Como otros mencionaron, CPython puede usar hilos solo para las esperas de E / S debido a GIL .

Si desea beneficiarse de múltiples núcleos para tareas vinculadas a la CPU, use 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: no se requiere una cola para enhebrar.

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í hay una versión modificada que pensé que era más útil (al menos para mí).

Actualizado: funciona tanto en Python 2 como en Python 3

try:
    # For Python 3
    import queue
    from urllib.request import urlopen
except:
    # For Python 2 
    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()

Dada una función, fenhebra así:

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

Para pasar argumentos a f

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

Esto me pareció muy útil: crear tantos subprocesos como núcleos y permitirles ejecutar una (gran) cantidad de tareas (en este caso, llamar a un programa de 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

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

Tiene agrupación de subprocesos y agrupación de procesos .

Lo siguiente da una idea:

Ejemplo de ThreadPoolExecutor ( fuente )

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

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

Usando el nuevo y sorprendente 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 al margen: para mantener el universo cuerdo, no olvide cerrar sus grupos / ejecutores si no usa el withcontexto (lo cual es tan increíble que lo hace por usted)

Para mí, el ejemplo perfecto para enhebrar es monitorear eventos asincrónicos. 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

La mayoría de la documentación y los tutoriales usan Python Threadingy el Queuemódulo, y pueden parecer abrumadores para los principiantes.

Quizás considere el concurrent.futures.ThreadPoolExecutormódulo de Python 3.

Combinado con la withcláusula y la comprensión de 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()

Vi muchos ejemplos aquí donde no se realizaba trabajo real, y en su mayoría estaban vinculados a la CPU. Aquí hay 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 cuatro 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 four workers,
    1000 - 1250 to worker 1
    1250 - 1500 to worker 2
    1500 - 1750 to worker 3
    1750 - 2000 to worker 4
    so let’s split the minimum and maximum 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 thread with the minimum and maximum split up to compute
        # Parallel computation will not work here due to the GIL since this is a CPU-bound task
        t = threading.Thread(target = find_primes_in, args = (start, end))
        threads.append(t)
        t.start()

    # Don’t forget to wait for the threads to finish
    for t in threads:
        t.join()

@time_stuff
def processing_prime_finder(nmin, nmax):
    """
    Split the minimum, maximum 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 a 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 de cuatro núcleos Mac OS X

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

Aquí está el ejemplo muy simple de importación CSV utilizando subprocesos. (La inclusión de 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)

Me gustaría contribuir con un ejemplo simple y las explicaciones que he encontrado útiles cuando tuve que abordar este problema yo mismo.

En esta respuesta, encontrará información sobre el GIL de Python (bloqueo global del intérprete) y un ejemplo simple del día a día escrito usando multiprocessing.dummy más algunos puntos de referencia simples.

Bloqueo global de intérpretes (GIL)

Python no permite múltiples subprocesos en el verdadero sentido de la palabra. Tiene un paquete de subprocesos múltiples, pero si desea varios subprocesos para acelerar su código, generalmente no es una buena idea usarlo.

Python tiene una construcción llamada bloqueo global del intérprete (GIL). El GIL se asegura de que solo uno de sus 'hilos' pueda ejecutarse a la vez. Un hilo adquiere el GIL, hace un poco de trabajo, luego pasa el GIL al siguiente hilo.

Esto sucede muy rápido, por lo que para el ojo humano puede parecer que sus hilos se están ejecutando en paralelo, pero en realidad solo se turnan para usar el mismo núcleo de CPU.

Todo este paso de GIL agrega gastos generales a la ejecución. Esto significa que si desea que su código se ejecute más rápido, usar el paquete de subprocesos a menudo no es una buena idea.

Hay razones para usar el paquete de subprocesos de Python. Si desea ejecutar algunas cosas simultáneamente, y la eficiencia no es una preocupación, entonces está totalmente bien y conveniente. O si está ejecutando código que necesita esperar algo (como algunas E / S), entonces podría tener mucho sentido. Pero la biblioteca de subprocesos no le permitirá usar núcleos de CPU adicionales.

El subprocesamiento múltiple se puede subcontratar al sistema operativo (mediante el procesamiento múltiple) y a alguna aplicación externa que llame a su código de Python (por ejemplo, Spark o Hadoop ), o algún código al que llame su código de Python (por ejemplo: podría haga que su código de Python llame a una función C que hace el costoso material de múltiples subprocesos

Por qué esto importa

Porque muchas personas pasan mucho tiempo tratando de encontrar cuellos de botella en su elegante código multiproceso de Python antes de aprender qué es el GIL.

Una vez que esta información es clara, aquí está mi código:

#!/bin/python
from multiprocessing.dummy import Pool
from subprocess import PIPE,Popen
import time
import os

# In the variable pool_size we define the "parallelness".
# For CPU-bound tasks, it doesn't make sense to create more Pool processes
# than you have cores to run them on.
#
# On the other hand, if you are using I/O-bound tasks, it may make sense
# to create a quite a few more Pool processes than cores, since the processes
# will probably spend most their time blocked (waiting for I/O to complete).
pool_size = 8

def do_ping(ip):
    if os.name == 'nt':
        print ("Using Windows Ping to " + ip)
        proc = Popen(['ping', ip], stdout=PIPE)
        return proc.communicate()[0]
    else:
        print ("Using Linux / Unix Ping to " + ip)
        proc = Popen(['ping', ip, '-c', '4'], stdout=PIPE)
        return proc.communicate()[0]


os.system('cls' if os.name=='nt' else 'clear')
print ("Running using threads\n")
start_time = time.time()
pool = Pool(pool_size)
website_names = ["www.google.com","www.facebook.com","www.pinterest.com","www.microsoft.com"]
result = {}
for website_name in website_names:
    result[website_name] = pool.apply_async(do_ping, args=(website_name,))
pool.close()
pool.join()
print ("\n--- Execution took {} seconds ---".format((time.time() - start_time)))

# Now we do the same without threading, just to compare time
print ("\nRunning NOT using threads\n")
start_time = time.time()
for website_name in website_names:
    do_ping(website_name)
print ("\n--- Execution took {} seconds ---".format((time.time() - start_time)))

# Here's one way to print the final output from the threads
output = {}
for key, value in result.items():
    output[key] = value.get()
print ("\nOutput aggregated in a Dictionary:")
print (output)
print ("\n")

print ("\nPretty printed output: ")
for key, value in output.items():
    print (key + "\n")
    print (value)

Aquí hay múltiples subprocesos con un ejemplo simple que será útil. Puede ejecutarlo y comprender fácilmente cómo funciona el subprocesamiento múltiple en Python. Usé un candado para evitar el acceso a otros hilos hasta que los hilos anteriores terminaron su trabajo. Mediante el uso de esta línea de código,

tLock = threading.BoundedSemaphore (valor = 4)

puede permitir una serie de procesos a la vez y mantener el resto de los subprocesos que se ejecutarán más tarde o después de haber terminado 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()

Con el préstamo de esta publicación , sabemos cómo elegir entre el subprocesamiento múltiple, el multiprocesamiento y el asíncrono / asyncioy su uso.

Python 3 tiene una nueva biblioteca integrada para concurrencia y paralelismo: concurrent.futures

Entonces demostraré a través de un experimento ejecutar cuatro tareas (es decir, .sleep()método) de la siguiente Threading-Poolmanera:

from concurrent.futures import ThreadPoolExecutor, as_completed
from time import sleep, time

def concurrent(max_worker=1):
    futures = []

    tick = time()
    with ThreadPoolExecutor(max_workers=max_worker) as executor:
        futures.append(executor.submit(sleep, 2))  # Two seconds sleep
        futures.append(executor.submit(sleep, 1))
        futures.append(executor.submit(sleep, 7))
        futures.append(executor.submit(sleep, 3))

        for future in as_completed(futures):
            if future.result() is not None:
                print(future.result())

    print('Total elapsed time by {} workers:'.format(max_worker), time()-tick)

concurrent(5)
concurrent(4)
concurrent(3)
concurrent(2)
concurrent(1)

Salida:

Total elapsed time by 5 workers: 7.007831811904907
Total elapsed time by 4 workers: 7.007944107055664
Total elapsed time by 3 workers: 7.003149509429932
Total elapsed time by 2 workers: 8.004627466201782
Total elapsed time by 1 workers: 13.013478994369507

[ NOTA ]:

• Como puede ver en los resultados anteriores, el mejor caso fue 3 trabajadores para esas cuatro tareas.
• Si tiene una tarea de proceso en lugar de E / S unida o bloqueada ( multiprocessingvs threading), puede cambiarla ThreadPoolExecutora ProcessPoolExecutor.

Ninguna de las soluciones anteriores usaba múltiples núcleos en mi servidor GNU / Linux (donde no tengo derechos de administrador). Simplemente corrieron en un solo núcleo.

Usé la os.forkinterfaz de nivel inferior para generar múltiples procesos. Este es el código que funcionó para mí:

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