¿Cuál es la forma más rápida de enviar 100.000 solicitudes HTTP en Python?

Una solución usando tornado biblioteca de redes asíncronas

from tornado import ioloop, httpclient

i = 0

def handle_request(response):
    print(response.code)
    global i
    i -= 1
    if i == 0:
        ioloop.IOLoop.instance().stop()

http_client = httpclient.AsyncHTTPClient()
for url in open('urls.txt'):
    i += 1
    http_client.fetch(url.strip(), handle_request, method='HEAD')
ioloop.IOLoop.instance().start()

Los hilos no son absolutamente la respuesta aquí. Proporcionarán cuellos de botella tanto del proceso como del núcleo, así como límites de rendimiento que no son aceptables si el objetivo general es "la forma más rápida".

Un poco de twisted y su cliente asíncrono HTTP le daría resultados mucho mejores.

Use grequests, es una combinación de peticiones + módulo Gevent .

GRequests le permite usar Peticiones con Gevent para hacer peticiones HTTP asíncronas fácilmente.

El uso es simple:

import grequests

urls = [
   'http://www.heroku.com',
   'http://tablib.org',
   'http://httpbin.org',
   'http://python-requests.org',
   'http://kennethreitz.com'
]

Crear un conjunto de solicitudes no enviadas:

>>> rs = (grequests.get(u) for u in urls)

Envíalos todos al mismo tiempo:

>>> grequests.map(rs)
[<Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>, <Response [200]>]

Las cosas han cambiado bastante desde 2010 cuando se publicó esto y no he probado todas las otras respuestas, pero he intentado algunas, y encontré que esto funciona mejor para mí usando python3.6.

Pude obtener alrededor de ~150 dominios únicos por segundo que se ejecutan en AWS.

import pandas as pd
import concurrent.futures
import requests
import time

out = []
CONNECTIONS = 100
TIMEOUT = 5

tlds = open('../data/sample_1k.txt').read().splitlines()
urls = ['http://{}'.format(x) for x in tlds[1:]]

def load_url(url, timeout):
    ans = requests.head(url, timeout=timeout)
    return ans.status_code

with concurrent.futures.ThreadPoolExecutor(max_workers=CONNECTIONS) as executor:
    future_to_url = (executor.submit(load_url, url, TIMEOUT) for url in urls)
    time1 = time.time()
    for future in concurrent.futures.as_completed(future_to_url):
        try:
            data = future.result()
        except Exception as exc:
            data = str(type(exc))
        finally:
            out.append(data)

            print(str(len(out)),end="\r")

    time2 = time.time()

print(f'Took {time2-time1:.2f} s')
print(pd.Series(out).value_counts())

Si está buscando obtener el mejor rendimiento posible, es posible que desee considerar el uso de E/S asincrónicas en lugar de hilos. La sobrecarga asociada con miles de subprocesos del sistema operativo no es trivial y el cambio de contexto dentro del intérprete de Python agrega aún más. El enhebrado sin duda hará el trabajo, pero sospecho que una ruta asíncrona proporcionará un mejor rendimiento general.

Específicamente, sugeriría el cliente web async en la biblioteca Twisted ( http://www.twistedmatrix.com ). Tiene una curva de aprendizaje ciertamente empinada, pero es bastante fácil de usar una vez que tienes un buen manejo del estilo de programación asíncrona de Twisted.

Un HowTo sobre la API de cliente web asíncrona de Twisted está disponible en:

Http://twistedmatrix.com/documents/current/web/howto/client.html

Un buen enfoque para resolver este problema es primero escribir el código requerido para obtener un resultado, luego incorporar código de subproceso para paralelizar la aplicación.

En un mundo perfecto esto simplemente significaría iniciar simultáneamente 100,000 subprocesos que arrojan sus resultados en un diccionario o lista para su procesamiento posterior, pero en la práctica está limitado en cuántas solicitudes HTTP paralelas puede emitir de esta manera. Localmente, tiene límites en la cantidad de sockets que puede abrir simultáneamente, cuántos hilos de ejecución permitirá su intérprete de Python. De forma remota, puede estar limitado en el número de conexiones simultáneas si todas las solicitudes son contra un servidor o varios. Estas limitaciones probablemente requerirán que escriba el script de tal manera que solo sondee una pequeña fracción de las URL en un momento dado (100, como mencionó otro póster, es probablemente un tamaño decente de grupo de subprocesos, aunque puede encontrar que puede implementar con éxito muchos mas).

Puede seguir este patrón de diseño para resolver el problema anterior:

1. Iniciar un subproceso que inicia nuevos subprocesos de solicitud hasta el número de subprocesos actualmente en ejecución (puede rastrearlos a través de subprocesos.active_count() o empujando los objetos thread en una estructura de datos) es >= su número máximo de solicitudes simultáneas (digamos 100), luego duerme por un corto tiempo de espera. Este hilo debe terminar cuando no hay más URLs para procesar. Por lo tanto, el hilo se mantendrá despertarse, lanzar nuevos hilos y dormir hasta que haya terminado.
2. Haga que los subprocesos de solicitud almacenen sus resultados en alguna estructura de datos para su posterior recuperación y salida. Si la estructura en la que está almacenando los resultados es un list o dict en CPython, puede agregar o insertar elementos únicos de sus subprocesos sin bloqueos, pero si escribe en un archivo o requiere una interacción de datos entre subprocesos más compleja debe usar un bloqueo de exclusión mutua para proteger esto estado de corrupción .

Te sugiero que uses el módulo threading. Puede usarlo para iniciar y rastrear subprocesos en ejecución. El soporte de subprocesos de Python es simple, pero la descripción de su problema sugiere que es completamente suficiente para sus necesidades.

Finalmente, si quieres ver una aplicación bastante sencilla de una aplicación de red paralela escrita en Python, echa un vistazo a ssh.py . Es una pequeña biblioteca que usa Python enhebrado para paralelizar muchas conexiones SSH. El diseño es lo suficientemente cerca de sus necesidades que usted puede encontrar que es un buen recurso.

Una solución:

from twisted.internet import reactor, threads
from urlparse import urlparse
import httplib
import itertools


concurrent = 200
finished=itertools.count(1)
reactor.suggestThreadPoolSize(concurrent)

def getStatus(ourl):
    url = urlparse(ourl)
    conn = httplib.HTTPConnection(url.netloc)   
    conn.request("HEAD", url.path)
    res = conn.getresponse()
    return res.status

def processResponse(response,url):
    print response, url
    processedOne()

def processError(error,url):
    print "error", url#, error
    processedOne()

def processedOne():
    if finished.next()==added:
        reactor.stop()

def addTask(url):
    req = threads.deferToThread(getStatus, url)
    req.addCallback(processResponse, url)
    req.addErrback(processError, url)   

added=0
for url in open('urllist.txt'):
    added+=1
    addTask(url.strip())

try:
    reactor.run()
except KeyboardInterrupt:
    reactor.stop()

Tiempo de prueba:

[kalmi@ubi1:~] wc -l urllist.txt
10000 urllist.txt
[kalmi@ubi1:~] time python f.py > /dev/null 

real    1m10.682s
user    0m16.020s
sys 0m10.330s
[kalmi@ubi1:~] head -n 6 urllist.txt
http://www.google.com
http://www.bix.hu
http://www.godaddy.com
http://www.google.com
http://www.bix.hu
http://www.godaddy.com
[kalmi@ubi1:~] python f.py | head -n 6
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu
200 http://www.bix.hu

Pingtime:

bix.hu is ~10 ms away from me
godaddy.com: ~170 ms
google.com: ~30 ms

Usar un grupo de subprocesos es una buena opción, y lo hará bastante fácil. Desafortunadamente, python no tiene una biblioteca estándar que haga que los grupos de subprocesos sean ultra fáciles. Pero aquí hay una biblioteca decente que debería ayudarte a empezar: http://www.chrisarndt.de/projects/threadpool /

Ejemplo de código de su sitio:

pool = ThreadPool(poolsize)
requests = makeRequests(some_callable, list_of_args, callback)
[pool.putRequest(req) for req in requests]
pool.wait()

Espero que esto ayude.

Para su caso, enhebrar probablemente hará el truco, ya que probablemente pasará la mayor parte del tiempo esperando una respuesta. Hay módulos útiles como Queue en la biblioteca estándar que podrían ayudar.

Hice algo similar con la descarga paralela de archivos antes y fue lo suficientemente bueno para mí, pero no estaba en la escala de la que estás hablando.

Si su tarea estaba más vinculada a la CPU, es posible que desee mirar el módulo multiprocesamiento, que le permitirá para utilizar más CPU / núcleos / hilos (más procesos que no se bloquearán entre sí ya que el bloqueo es por proceso)

Considere usar Windmill, aunque Windmill probablemente no pueda hacer tantos hilos.

Podría hacerlo con un script Python enrollado a mano en 5 máquinas, cada una conectando outbound usando los puertos 40000-60000, abriendo 100,000 conexiones de puerto.

También, podría ayudar hacer una prueba de ejemplo con una aplicación de control de calidad bien roscada como OpenSTA para tener una idea de cuánto puede manejar cada servidor.

También, trate de buscar solo usando Perl simple con la clase LWP:: ConnCache. Probablemente obtendrás más rendimiento (más conexiones) de esa manera.

Este cliente web asincrónico retorcido va bastante rápido.

#!/usr/bin/python2.7

from twisted.internet import reactor
from twisted.internet.defer import Deferred, DeferredList, DeferredLock
from twisted.internet.defer import inlineCallbacks
from twisted.web.client import Agent, HTTPConnectionPool
from twisted.web.http_headers import Headers
from pprint import pprint
from collections import defaultdict
from urlparse import urlparse
from random import randrange
import fileinput

pool = HTTPConnectionPool(reactor)
pool.maxPersistentPerHost = 16
agent = Agent(reactor, pool)
locks = defaultdict(DeferredLock)
codes = {}

def getLock(url, simultaneous = 1):
    return locks[urlparse(url).netloc, randrange(simultaneous)]

@inlineCallbacks
def getMapping(url):
    # Limit ourselves to 4 simultaneous connections per host
    # Tweak this number, but it should be no larger than pool.maxPersistentPerHost 
    lock = getLock(url,4)
    yield lock.acquire()
    try:
        resp = yield agent.request('HEAD', url)
        codes[url] = resp.code
    except Exception as e:
        codes[url] = str(e)
    finally:
        lock.release()


dl = DeferredList(getMapping(url.strip()) for url in fileinput.input())
dl.addCallback(lambda _: reactor.stop())

reactor.run()
pprint(codes)

La forma más fácil sería usar la biblioteca de subprocesos integrada en Python. No son hilos "reales" / kernel Tienen problemas (como serialización), pero son lo suficientemente buenos. Querrías una cola y un grupo de subprocesos. Una opción es aquí, pero es trivial escribir la tuya propia. No puedes paralelizar todas las 100,000 llamadas, pero puedes disparar 100 (o más) de ellas al mismo tiempo.