Jaki jest najszybszy sposób wysyłania 100 000 żądań HTTP w Pythonie?

Otwieram plik, który ma 100 000 adresów URL. Muszę wysłać żądanie HTTP do każdego adresu URL i wydrukować kod stanu. Używam Pythona 2.6 i do tej pory przyglądałem się wielu mylącym sposobom, w jakie Python implementuje wątkowanie / współbieżność. Zajrzałem nawet do biblioteki współbieżności w Pythonie , ale nie mogę dowiedzieć się, jak poprawnie napisać ten program. Czy ktoś napotkał podobny problem? Chyba na ogół muszę wiedzieć, jak wykonywać tysiące zadań w Pythonie tak szybko, jak to możliwe - przypuszczam, że oznacza to „jednocześnie”.

Rozwiązanie Twistedless:

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)

Ten jest nieco szybszy niż skręcone rozwiązanie i zużywa mniej procesora.

Rozwiązanie wykorzystujące asynchroniczną bibliotekę sieciową 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()

Wszystko zmieniło się nieco od 2010 roku, kiedy to zostało opublikowane i nie wypróbowałem wszystkich innych odpowiedzi, ale wypróbowałem kilka, i znalazłem, że działa najlepiej dla mnie za pomocą python3.6.

Udało mi się pobrać około ~ 150 unikalnych domen na sekundę działających na 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())

Wątki absolutnie nie są tutaj odpowiedzią. Zapewnią zarówno wąskie gardła procesu, jak i jądra, a także ograniczenia przepustowości, które są nie do przyjęcia, jeśli ogólnym celem jest „najszybszy sposób”.

Trochę twistedi jego asynchroniczny HTTPklient dałby znacznie lepsze wyniki.

Wiem, że to stare pytanie, ale w Pythonie 3.7 możesz to zrobić za pomocą asyncioi 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))

Możesz przeczytać więcej na ten temat i zobaczyć tutaj przykład .

Używaj grequestów , jest to kombinacja żądań + moduł Gevent.

GRequests pozwala na używanie żądań z Geventem do łatwego tworzenia asynchronicznych żądań HTTP.

Użycie jest proste:

import grequests

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

Utwórz zestaw niewysłanych żądań:

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

Wyślij je wszystkie jednocześnie:

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

Dobrym podejściem do rozwiązania tego problemu jest najpierw napisanie kodu wymaganego do uzyskania jednego wyniku, a następnie włączenie kodu wątkowego w celu zrównoleglenia aplikacji.

W idealnym świecie oznaczałoby to po prostu jednoczesne uruchomienie 100 000 wątków, które wysyłają swoje wyniki do słownika lub listy do późniejszego przetworzenia, ale w praktyce masz ograniczoną liczbę równoległych żądań HTTP, które możesz wysłać w ten sposób. Lokalnie istnieją ograniczenia dotyczące liczby gniazd, które można jednocześnie otworzyć, liczby wątków wykonania, na które zezwoli interpreter języka Python. Zdalnie możesz mieć ograniczoną liczbę jednoczesnych połączeń, jeśli wszystkie żądania dotyczą jednego serwera lub wielu. Ograniczenia te prawdopodobnie będą wymagać napisania skryptu w taki sposób, aby w dowolnym momencie sondować tylko niewielką część adresów URL (100, jak wspomniano w innym plakacie, to prawdopodobnie przyzwoity rozmiar puli wątków, chociaż może się okazać, że może pomyślnie wdrożyć wiele innych).

Możesz wykonać ten wzór, aby rozwiązać powyższy problem:

1. Rozpocznij wątek, który uruchamia nowe wątki żądań, aż liczba aktualnie działających wątków (możesz je śledzić za pomocą threading.active_count () lub poprzez pchanie obiektów wątku do struktury danych) wynosi> = maksymalna liczba jednoczesnych żądań (powiedzmy 100) , potem śpi przez krótki czas. Wątek powinien zakończyć się, gdy nie będzie już adresów URL do przetworzenia. W ten sposób wątek będzie się budził, uruchamiał nowe wątki i spał, dopóki nie skończysz.
2. Niech wątki żądań przechowują swoje wyniki w jakiejś strukturze danych do późniejszego pobrania i wygenerowania. Jeśli struktura, w której zapisujesz wyniki, to a listlubdict CPython, możesz bezpiecznie dołączyć lub wstawić unikalne elementy z wątków bez blokad , ale jeśli piszesz do pliku lub wymagasz bardziej złożonej interakcji między wątkami , powinieneś użyć blokada wzajemnego wykluczenia w celu ochrony tego stanu przed korupcją .

Sugerowałbym użycie modułu wątków . Możesz go użyć do uruchamiania i śledzenia uruchomionych wątków. Obsługa wątków w Pythonie jest pusta, ale opis problemu sugeruje, że jest całkowicie wystarczający dla twoich potrzeb.

Wreszcie, jeśli chcesz zobaczyć całkiem proste stosowanie równoległej aplikacji sieciowej napisany w Pythonie, sprawdź ssh.py . To mała biblioteka, która wykorzystuje wątki w języku Python do równoległego łączenia wielu połączeń SSH. Projekt jest na tyle zbliżony do twoich wymagań, że możesz uznać go za dobry zasób.

Jeśli chcesz uzyskać najlepszą możliwą wydajność, możesz rozważyć użycie asynchronicznego we / wy zamiast wątków. Narzut związany z tysiącami wątków systemu operacyjnego nie jest trywialny, a przełączanie kontekstu w interprecie Pythona dodaje jeszcze więcej. Wątkowanie z pewnością wykona zadanie, ale podejrzewam, że trasa asynchroniczna zapewni lepszą ogólną wydajność.

W szczególności sugerowałbym asynchronicznego klienta WWW w bibliotece Twisted ( http://www.twistedmatrix.com ). Ma wprawdzie stromą krzywą uczenia się, ale jest dość łatwy w użyciu, gdy dobrze opanujesz styl programowania asynchronicznego Twisted.

Poradnik na temat asynchronicznego interfejsu API klienta WWW Twisted jest dostępny pod adresem:

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

Rozwiązanie:

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

Czas na test:

[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

Korzystanie z puli wątków jest dobrą opcją i znacznie to ułatwi. Niestety, python nie ma standardowej biblioteki, która sprawia, że ​​pule wątków są wyjątkowo łatwe. Ale tutaj jest przyzwoita biblioteka, która powinna na początek: http://www.chrisarndt.de/projects/threadpool/

Przykładowy kod z ich strony:

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

Mam nadzieję że to pomoże.

Tworzenie epollobiektu,
otworzyć wiele gniazd TCP klienta,
dostosować swoje bufory Prześlij być nieco więcej niż nagłówka żądania,
wysłać nagłówek żądania - powinna być natychmiastowa, po prostu umieszczając w buforze, zarejestruj gniazdo w epollobiekcie,
zrobić .pollna epollobect,
czytać pierwszy 3 bajty z każdego gniazda z .poll,
zapisz je, sys.stdouta następnie\n (nie opróżnij), zamknij gniazdo klienta.

Ogranicz liczbę gniazd otwieranych jednocześnie - radzić sobie z błędami podczas tworzenia gniazd. Utwórz nowe gniazdo tylko wtedy, gdy inne jest zamknięte.
Dostosuj limity systemu operacyjnego.
Spróbuj rozwidlić kilka (nielicznych) procesów: może to pomóc nieco bardziej efektywnie wykorzystać procesor.

W twoim przypadku wątki prawdopodobnie załatwią sprawę, ponieważ prawdopodobnie będziesz spędzać najwięcej czasu czekając na odpowiedź. W standardowej bibliotece znajdują się pomocne moduły, takie jak Kolejka, które mogą pomóc.

Zrobiłem podobnie z równoległym pobieraniem plików i było to dla mnie wystarczająco dobre, ale nie na skalę, o której mówisz.

Jeśli twoje zadanie było bardziej związane z procesorem, możesz spojrzeć na moduł wieloprocesowy , który pozwoli ci wykorzystać więcej procesorów / rdzeni / wątków (więcej procesów, które nie będą się wzajemnie blokować, ponieważ blokowanie jest na proces)

Rozważ użycie wiatraka , chociaż Windmill prawdopodobnie nie może zrobić tak wielu wątków.

Można to zrobić za pomocą ręcznie toczonego skryptu Python na 5 komputerach, z których każdy łączy dane wychodzące za pomocą portów 40000-60000, otwierając 100 000 połączeń portów.

Może także pomóc w przeprowadzeniu przykładowego testu z ładnie napisaną aplikacją QA, taką jak OpenSTA , aby dowiedzieć się, ile każdy serwer może obsłużyć.

Spróbuj także po prostu użyć prostego Perla z klasą LWP :: ConnCache. Prawdopodobnie uzyskasz w ten sposób większą wydajność (więcej połączeń).

Ten pokręcony klient sieciowy asynchroniczny działa dość szybko.

#!/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)

Przekonałem się, że użycie tego tornadopakietu jest najszybszym i najprostszym sposobem na osiągnięcie tego:

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

Najprostszym sposobem byłoby użycie wbudowanej biblioteki wątków Pythona. Nie są „prawdziwymi” wątkami jądra. Mają problemy (takie jak serializacja), ale są wystarczająco dobre. Chcesz kolejki i puli wątków. Jedną z opcji jest tutaj , ale to jest trywialne napisać własną. Nie możesz zrównoważyć wszystkich 100 000 połączeń, ale możesz zwolnić 100 (lub więcej) z nich jednocześnie.