Was ist der schnellste Weg, um 100.000 HTTP-Anfragen in Python zu senden?

Question 1

Ich öffne eine Datei mit 100.000 URLs. Ich muss eine HTTP-Anforderung an jede URL senden und den Statuscode drucken. Ich verwende Python 2.6 und habe mir bisher die vielen verwirrenden Möglichkeiten angesehen, wie Python Threading/Parallelität implementiert. Ich habe mir sogar die Python angesehen Übereinstimmung Bibliothek, kann aber nicht herausfinden, wie man dieses Programm richtig schreibt. Ist jemand auf ein ähnliches Problem gestoßen? Ich denke, im Allgemeinen muss ich wissen, wie ich Tausende von Aufgaben in Python so schnell wie möglich ausführen kann – ich nehme an, das bedeutet “gleichzeitig”.

Question 2

Twistedless-Lösung:

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)

Diese ist etwas schneller als die verdrehte Lösung und verbraucht weniger CPU.

Question 3

Die Dinge haben sich seit 2010, als dies gepostet wurde, ziemlich geändert, und ich habe nicht alle anderen Antworten ausprobiert, aber ich habe einige ausprobiert, und ich fand, dass dies mit Python 3.6 am besten für mich funktioniert.

Ich konnte etwa 150 eindeutige Domains pro Sekunde abrufen, die auf AWS ausgeführt wurden.

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

Question 4

Eine Lösung mit Tornado asynchrone Netzwerkbibliothek

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

Dieser Code verwendet nicht blockierende Netzwerk-E/A und hat keine Einschränkung. Es kann auf Zehntausende offener Verbindungen skaliert werden. Es wird in einem einzigen Thread ausgeführt, ist aber viel schneller als jede Threading-Lösung. Kasse nicht blockierende E/A

Question 5

Ich weiß, dass dies eine alte Frage ist, aber in Python 3.7 können Sie dies mit tun asyncio und 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))

Sie können mehr darüber lesen und ein Beispiel sehen hier.

Question 6

Threads sind hier absolut nicht die Antwort. Sie bieten sowohl Prozess- als auch Kernel-Engpässe sowie Durchsatzgrenzen, die nicht akzeptabel sind, wenn das Gesamtziel “der schnellste Weg” ist.

Ein kleines Bisschen von twisted und es ist asynchron HTTP Client würde Ihnen viel bessere Ergebnisse liefern.

Question 7

Verwenden wünsche es ist eine Kombination aus Anfragen + Gevent-Modul .

GRequests ermöglicht es Ihnen, Anfragen mit Gevent zu verwenden, um asynchrone HTTP-Anfragen einfach zu stellen.

Die Verwendung ist einfach:

import grequests

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

Erstellen Sie eine Reihe nicht gesendeter Anfragen:

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

Senden Sie sie alle gleichzeitig:

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

Question 8

(Notiz an mich selbst für das nächste Projekt)

Nur Python 3-Lösung verwenden requests. Es ist am einfachsten und es ist schnell, keine Notwendigkeit für Multiprocessing oder komplizierte asynchrone Bibliotheken.

Der wichtigste Aspekt ist die Wiederverwendung von Verbindungen, insbesondere für HTTPS (TLS erfordert einen zusätzlichen Roundtrip zum Öffnen). Beachten Sie, dass eine Verbindung spezifisch für eine Subdomain ist. Wenn Sie viele Seiten auf vielen Domains kratzen, können Sie die Liste der URLs sortieren, um die Wiederverwendung von Verbindungen zu maximieren (es wird effektiv nach Domain sortiert).

Es ist so schnell wie jeder asynchrone Code, wenn genügend Threads vorhanden sind. (Anfragen geben die Python-GIL frei, wenn sie auf die Antwort warten).

[Production grade code with some logging and error handling]

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

# source: https://stackoverflow.com/a/68583332/5994461

THREAD_POOL = 16

# This is how to create a reusable connection pool with python requests.
session = requests.Session()
session.mount(
    'https://',
    requests.adapters.HTTPAdapter(pool_maxsize=THREAD_POOL,
                                  max_retries=3,
                                  pool_block=True)
)

def get(url):
    response = session.get(url)
    logging.info("request was completed in %s seconds [%s]", response.elapsed.total_seconds(), response.url)
    if response.status_code != 200:
        logging.error("request failed, error code %s [%s]", response.status_code, response.url)
    if 500 <= response.status_code < 600:
        # server is overloaded? give it a break
        time.sleep(5)
    return response

def download(urls):
    with ThreadPoolExecutor(max_workers=THREAD_POOL) as executor:
        # wrap in a list() to wait for all requests to complete
        for response in list(executor.map(get, urls)):
            if response.status_code == 200:
                print(response.content)

def main():
    logging.basicConfig(
        format="%(asctime)s.%(msecs)03d %(levelname)-8s %(message)s",
        level=logging.INFO,
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    urls = [
        "https://httpstat.us/200",
        "https://httpstat.us/200",
        "https://httpstat.us/200",
        "https://httpstat.us/404",
        "https://httpstat.us/503"
    ]

    download(urls)

if __name__ == "__main__":
    main()