Acelera tus aplicaciones en python usando multi threading

Muchas veces terminamos escribiendo código en Python que realiza solicitudes remotas o lee múltiples archivos o procesa algunos datos. Y en muchos de esos casos, he visto programadores que usan un bucle simple que tarda una eternidad en terminar de ejecutarse. Por ejemplo:

import requests
from time import time

lista_urls = [
    "https://ejemplo.com/400",
    "https://ejemplo.com/410",
    "https://ejemplo.com/420",
    "https://ejemplo.com/430",
    "https://ejemplo.com/440",
    "https://ejemplo.com/450",
    "https://ejemplo.com/460",
    "https://ejemplo.com/470",
    "https://ejemplo.com/480",
    "https://ejemplo.com/490",
    "https://ejemplo.com/500",
    "https://ejemplo.com/510",
    "https://ejemplo.com/520",
    "https://ejemplo.com/530",
]

def descargar_archivo(url):
    html = requests.get(url, stream=True)
    return html.status_code

start = time()

for url in lista_urls:
    print(descargar_archivo(url))

Esto tomaria basante tiempo en realizar ya que son varias urls, probablemente muy pesadas.

Este es un ejemplo sensato y el código abrirá cada URL, esperará a que se cargue, imprimirá su código de estado y solo luego pasará a la siguiente URL. Este tipo de código es un muy buen candidato para subprocesos múltiples.

Los sistemas modernos pueden ejecutar muchos subprocesos y eso significa que puede realizar múltiples tareas a la vez con una sobrecarga muy baja. ¿Por qué no intentamos usarlo para que el código anterior procese estas URL más rápido?

Haremos uso de ThreadPoolExecutor de la biblioteca concurrent.futures. Es super fácil de usar. Déjame mostrarte un poco de código y luego explicarte cómo funciona.

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

lista_urls = [
    "https://ejemplo.com/400",
    "https://ejemplo.com/410",
    "https://ejemplo.com/420",
    "https://ejemplo.com/430",
    "https://ejemplo.com/440",
    "https://ejemplo.com/450",
    "https://ejemplo.com/460",
    "https://ejemplo.com/470",
    "https://ejemplo.com/480",
    "https://ejemplo.com/490",
    "https://ejemplo.com/500",
    "https://ejemplo.com/510",
    "https://ejemplo.com/520",
    "https://ejemplo.com/530",
]

def descargar_archivo(url):
    html = requests.get(url, stream=True)
    return html.status_code

start = time()

procesos = []
with ThreadPoolExecutor(max_workers=10) as executor:
    for url in url_list:
        procesos.append(executor.submit(descargar_archivo, url))

for tarea in as_completed(procesos):
    print(tarea.result())

¡Acabamos de acelerar nuestro código por un factor de casi 9! Y ni siquiera hicimos nada súper complicado. Los beneficios de rendimiento habrían sido aún mayores si hubiera más URL.

Entonces, ¿qué está pasando? Cuando llamamos a executor.submit estamos agregando una nueva tarea al grupo de subprocesos. Almacenamos esa tarea en la lista de procesos. Más adelante iteramos sobre los procesos e imprimimos el resultado.

El método as_completed produce los elementos (tareas) de la lista de procesos tan pronto como se completan. Hay dos razones por las que una tarea puede pasar al estado completado. Ha terminado de ejecutarse o se canceló. También podríamos haber pasado un parámetro de tiempo de espera a as_completed y si una tarea tomó más tiempo que ese período de tiempo, incluso entonces as_completed producirá esa tarea.