Jaki jest najszybszy sposób na wysłanie 100 000 żądań HTTP w Pythonie?

RozwiÄ…zanie wykorzystujÄ…ce tornado asynchronicznÄ… bibliotekÄ™ sieciowÄ…

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

Wątki absolutnie nie są tutaj odpowiedzią. Zapewnią one zarówno wąskie gardła procesu, jak i jądra, a także limity przepustowości, które nie są akceptowalne, jeśli ogólnym celem jest "najszybsza droga".

TrochÄ™ twisted i jego asynchroniczny HTTP klient daÅ‚by znacznie lepsze wyniki.

Użyj grequests , jest to kombinacja żądań + moduł Gevent .

GRequests pozwala używać żądań z Gevent do asynchronicznego żądania HTTP łatwo.

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 nie wysłanych żądań:

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

Wyślij je wszystkie w tym samym czasie:

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

Wiele się zmieniło od 2010 roku, kiedy to zostało opublikowane i nie próbowałem wszystkich innych odpowiedzi, ale próbowałem kilku, i okazało się, że to 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())

Jeśli chcesz uzyskać najlepszą możliwą wydajność, warto rozważyć użycie asynchronicznych We/Wy zamiast wątków. Narzut związany z tysiącami wątków OS jest nietrywialny, a przełączanie kontekstu w interpreterze Pythona dodaje jeszcze więcej. Threading z pewnością da radę, ale podejrzewam, że asynchroniczna trasa zapewni lepszą ogólną wydajność.

W szczególności proponuję async web client w Twisted library ( http://www.twistedmatrix.com). ma wprawdzie stromą krzywą uczenia się, ale jest dość łatwy w użyciu, gdy już opanujesz styl asynchronicznego programowania Twisted.

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

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 do równoległej 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 wydać w ten sposób. Lokalnie masz limity liczby gniazd, które możesz otworzyć jednocześnie, ile wątków wykonania Twój interpreter Pythona pozwoli. Zdalnie możesz mieć ograniczoną liczbę jednoczesnych połączeń, jeśli wszystkie żądania są skierowane przeciwko jednemu serwerowi lub wielu. Ograniczenia te prawdopodobnie będą wymagały, że piszesz skrypt w taki sposób, aby przeszukiwać tylko niewielką część adresów URL w tym samym czasie (100, jak wspomniano w innym plakacie, jest prawdopodobnie przyzwoitym rozmiarem puli wątków, chociaż może się okazać, że możesz z powodzeniem wdrożyć wiele więcej).

Możesz postępować zgodnie z tym wzorcem projektowym, aby rozwiązać powyższy problem:

1. Uruchom wątek, który uruchamia nowe wątki żądania, aż do liczby aktualnie uruchomionych wątków (możesz je śledzić poprzez threading.active_count () lub poprzez wciśnięcie obiektów wątku do struktury danych) wynosi > = maksymalna liczba jednoczesnych żądań (powiedzmy 100), a następnie uśpi na krótki czas. Ten wątek powinien zakończyć się, gdy nie ma więcej adresów URL do przetworzenia. W ten sposób wątek zachowa budzisz się, uruchamiasz nowe wątki i śpisz do końca.
2. niech wÄ…tki żądania przechowujÄ… swoje wyniki w jakiejÅ› strukturze danych, aby później je pobrać i wydrukować. JeÅ›li struktura, w której przechowujesz wyniki, jest list LUB dict w CPython, możesz bezpiecznie dodawać lub wstawiać unikalne elementy z wÄ…tków bez blokad, ale jeÅ›li piszesz do pliku lub wymagasz bardziej zÅ‚ożonej interakcji danych miÄ™dzy wÄ…tkami , powinieneÅ› użyć blokady wzajemnego wykluczenia, aby chronić ten wÄ…tek. PaÅ„stwo od korupcji .

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

Wreszcie, jeśli chcesz zobaczyć dość prostą aplikację równoległej aplikacji sieciowej napisanej w Pythonie, sprawdź ssh.py . jest to mała biblioteka, która używa Pythona threading w celu równoległego połączenia wielu SSH. Projekt jest na tyle zbliżony do Twoich wymagań, że może okazać się dobrym zasobem.

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

Testtime:

[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 dobrym rozwiązaniem i ułatwi to sprawę. Niestety, python nie ma standardowej biblioteki, która sprawia, że pule wątków są bardzo łatwe. Ale tutaj jest przyzwoita biblioteka, która powinna zacząć: http://www.chrisarndt.de/projects/threadpool/

Przykład kodu z ich strony:

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

W Twoim przypadku, threading prawdopodobnie załatwi sprawę, ponieważ prawdopodobnie będziesz spędzać większość czasu czekając na odpowiedź. Istnieją pomocne moduły, takie jak Queue w bibliotece standardowej, które mogą pomóc.

Zrobiłem podobną rzecz z równoległym pobieraniem plików wcześniej i było to dla mnie wystarczająco dobre, ale nie było to w skali, o której mówisz.

Jeśli Twoje zadanie było bardziej związane z procesorem, warto przyjrzeć się modułowi multiprocessing , który pozwoli Ci aby wykorzystać więcej procesorów / rdzeni/wątków (więcej procesów, które nie blokują się nawzajem, ponieważ blokowanie jest dla każdego procesu)

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

Można to zrobić ręcznie zwijanym skryptem Pythona na 5 maszynach, z których każda łączy wychodzące za pomocą portów 40000-60000, otwierając 100 000 połączeń portowych.

Może również pomóc zrobić przykładowy test z ładnie gwintowaną aplikacją QA, taką jak OpenSTA, aby zorientować się, ile każdy serwer może obsłużyć.

Spróbuj również przyjrzeć się prostemu Perlowi z LWP:: Klasa ConnCache. W ten sposób prawdopodobnie uzyskasz większą wydajność (więcej połączeń).

Ten Pokręcony klient www 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)

Najprostszym sposobem byłoby użycie wbudowanej biblioteki wątków Pythona. nie są "prawdziwymi" / wątkami jądra mają problemy (jak serializacja), ale są wystarczająco dobre. Chcesz mieć pulę kolejek i wątków. Jedną z opcji jest tutaj , ale pisanie własnych jest trywialne. Nie można równoległe wszystkie 100,000 połączeń, ale można odpalić 100 (lub tak) z nich w tym samym czasie.