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

Estoy abriendo un archivo que tiene 100,000 URL. Necesito enviar una solicitud HTTP a cada URL e imprimir el código de estado. Estoy usando Python 2.6, y hasta ahora examiné las muchas formas confusas en que Python implementa subprocesamiento / concurrencia. Incluso he mirado la biblioteca de concurrencia de Python , pero no puedo entender cómo escribir este programa correctamente. ¿Alguien ha encontrado un problema similar? Supongo que generalmente necesito saber cómo realizar miles de tareas en Python lo más rápido posible, supongo que eso significa 'concurrentemente'.

Solución retorcida:

from urlparse import urlparse
from threading import Thread
import httplib, sys
from Queue import Queue

concurrent = 200

def doWork():
    while True:
        url = q.get()
        status, url = getStatus(url)
        doSomethingWithResult(status, url)
        q.task_done()

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

def doSomethingWithResult(status, url):
    print status, url

q = Queue(concurrent * 2)
for i in range(concurrent):
    t = Thread(target=doWork)
    t.daemon = True
    t.start()
try:
    for url in open('urllist.txt'):
        q.put(url.strip())
    q.join()
except KeyboardInterrupt:
    sys.exit(1)

Esta es ligeramente más rápida que la solución retorcida y usa menos CPU.

Una solución que utiliza la biblioteca de red asincrónica de tornado

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

Las cosas han cambiado bastante desde 2010 cuando esto se publicó y no he probado todas las otras respuestas, pero he intentado algunas, y descubrí 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())

Los hilos no son absolutamente la respuesta aquí. Proporcionarán cuellos de botella en el proceso y en el 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 twistedy su HTTPcliente asincrónico le daría resultados mucho mejores.

Sé que esta es una vieja pregunta, pero en Python 3.7 puedes hacer esto usando asyncioy aiohttp.

import asyncio
import aiohttp
from aiohttp import ClientSession, ClientConnectorError

async def fetch_html(url: str, session: ClientSession, **kwargs) -> tuple:
    try:
        resp = await session.request(method="GET", url=url, **kwargs)
    except ClientConnectorError:
        return (url, 404)
    return (url, resp.status)

async def make_requests(urls: set, **kwargs) -> None:
    async with ClientSession() as session:
        tasks = []
        for url in urls:
            tasks.append(
                fetch_html(url=url, session=session, **kwargs)
            )
        results = await asyncio.gather(*tasks)

    for result in results:
        print(f'{result[1]} - {str(result[0])}')

if __name__ == "__main__":
    import pathlib
    import sys

    assert sys.version_info >= (3, 7), "Script requires Python 3.7+."
    here = pathlib.Path(__file__).parent

    with open(here.joinpath("urls.txt")) as infile:
        urls = set(map(str.strip, infile))

    asyncio.run(make_requests(urls=urls))

Puede leer más al respecto y ver un ejemplo aquí .

Usa grequests , es una combinación de solicitudes + módulo Gevent.

GRequests le permite usar Solicitudes con Gevent para hacer Solicitudes HTTP asincrónicas 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'
]

Cree un conjunto de solicitudes no enviadas:

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

Envíelos todos al mismo tiempo:

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

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

En un mundo perfecto, esto simplemente significaría iniciar simultáneamente 100,000 hilos que generan sus resultados en un diccionario o lista para su posterior procesamiento, pero en la práctica está limitado en la cantidad de solicitudes HTTP paralelas que 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 le 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 cualquier momento (100, como se menciona en otro póster, es probablemente un tamaño de grupo de subprocesos decente, aunque puede encontrar que puede implementar con éxito muchos más).

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

1. Inicie un subproceso que inicie nuevos subprocesos de solicitud hasta que el número de subprocesos actualmente en ejecución (puede rastrearlos a través de threading.active_count () o insertando los objetos del subproceso 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 debería terminar cuando no haya más URL para procesar. Por lo tanto, el hilo seguirá despertando, lanzando nuevos hilos y durmiendo hasta que haya terminado.
2. Haga que los hilos 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 a listo dicten CPython, puede agregar o insertar de forma segura elementos únicos de sus hilos sin bloqueos , pero si escribe en un archivo o requiere una interacción de datos entre hilos más compleja , debe usar un bloqueo de exclusión mutua para proteger este estado de la corrupción .

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

Por último, si desea ver una aplicación directa bonita de una aplicación de red en paralelo escrito en Python, echa un vistazo a ssh.py . Es una pequeña biblioteca que utiliza subprocesos de Python para paralelizar muchas conexiones SSH. El diseño está lo suficientemente cerca de sus requisitos como para que sea un buen recurso.

Si está buscando obtener el mejor rendimiento posible, es posible que desee considerar el uso de E / S asincrónicas en lugar de subprocesos. 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 ciertamente hará el trabajo, pero sospecho que una ruta asincrónica proporcionará un mejor rendimiento general.

Específicamente, sugeriría el cliente web asíncrono 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 dominas el estilo de programación asincrónica de Twisted.

Un tutorial sobre API de cliente web asincrónico de Twisted está disponible en:

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

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 extremadamente fáciles. Pero aquí hay una biblioteca decente que debería comenzar: 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.

Cree un epollobjeto,
abra muchos sockets TCP del cliente,
ajuste sus búferes de envío para que sean un poco más que el encabezado de la solicitud,
envíe un encabezado de solicitud: debe ser inmediato, simplemente colóquelo en un búfer, registre el zócalo en el epollobjeto,
hágalo .pollal epollobedecer,
lea primero 3 bytes de cada socket de .poll,
escríbalos a sys.stdoutseguido de\n (no vaciar), cierre el socket del cliente.

Limite el número de sockets abiertos simultáneamente: maneje los errores cuando se crean sockets. Cree un nuevo socket solo si otro está cerrado.
Ajusta los límites del sistema operativo.
Intente bifurcarse en unos pocos (no muchos) procesos: esto puede ayudar a usar la CPU un poco más efectivamente.

Para su caso, el enhebrado probablemente servirá, 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 ver el módulo de multiprocesamiento , que le permitirá utilizar más CPU / núcleos / subprocesos (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.

Puede hacerlo con un script Python enrollado a mano en 5 máquinas, cada una de las cuales se conecta a la salida utilizando los puertos 40000-60000, abriendo 100,000 conexiones de puerto.

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

Además, intente buscar simplemente usando Perl simple con la clase LWP :: ConnCache. Probablemente obtendrá más rendimiento (más conexiones) de esa manera.

Este retorcido cliente web asíncrono 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)

Descubrí que usar el tornadopaquete es la forma más rápida y sencilla de lograr esto:

from tornado import ioloop, httpclient, gen


def main(urls):
    """
    Asynchronously download the HTML contents of a list of URLs.
    :param urls: A list of URLs to download.
    :return: List of response objects, one for each URL.
    """

    @gen.coroutine
    def fetch_and_handle():
        httpclient.AsyncHTTPClient.configure(None, defaults=dict(user_agent='MyUserAgent'))
        http_client = httpclient.AsyncHTTPClient()
        waiter = gen.WaitIterator(*[http_client.fetch(url, raise_error=False, method='HEAD')
                                    for url in urls])
        results = []
        # Wait for the jobs to complete
        while not waiter.done():
            try:
                response = yield waiter.next()
            except httpclient.HTTPError as e:
                print(f'Non-200 HTTP response returned: {e}')
                continue
            except Exception as e:
                print(f'An unexpected error occurred querying: {e}')
                continue
            else:
                print(f'URL \'{response.request.url}\' has status code <{response.code}>')
                results.append(response)
        return results

    loop = ioloop.IOLoop.current()
    web_pages = loop.run_sync(fetch_and_handle)

    return web_pages

my_urls = ['url1.com', 'url2.com', 'url100000.com']
responses = main(my_urls)
print(responses[0])

La forma más fácil sería utilizar la biblioteca de subprocesos incorporada de Python. No son hilos "reales" / del núcleo. Tienen problemas (como la serialización), pero son lo suficientemente buenos. Desea una cola y un grupo de subprocesos. Una opción está aquí , pero es trivial escribir la suya. No puede hacer paralelo a todas las 100,000 llamadas, pero puede disparar 100 (más o menos) al mismo tiempo.