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

287

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”.

Igor Ganapolski
źródło
47
Upewnij się, że wykonujesz tylko żądanie HEAD (aby nie pobrać całego dokumentu). Zobacz: stackoverflow.com/questions/107405/…
Tarnay Kálmán
5
Doskonały punkt, Kalmi. Jeśli wszystko, czego chce Igor, to status żądania, te 100 000 żądań pójdzie znacznie, znacznie, znacznie szybciej. Znacznie szybciej.
Adam Crossland
1
Nie potrzebujesz do tego nici; najskuteczniejszym sposobem jest użycie biblioteki asynchronicznej, takiej jak Twisted.
jemfinch
3
Oto przykłady kodu opartego na gevent, twisted i asyncio (testowane na 1000000 żądaniach)
jfs
4
@ TarnayKálmán możliwe jest requests.geti requests.head(tj. Żądanie strony kontra żądanie główne) zwrócenie różnych kodów stanu, więc nie jest to najlepsza rada
AlexG

Odpowiedzi:

200

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.

Tarnay Kálmán
źródło
10
@Kalmi, dlaczego ustawiasz kolejkę concurrent*2?
Marcel Wilson,
8
Nie zapomnij zamknąć połączenia conn.close() . Otwarcie zbyt wielu połączeń HTTP może w pewnym momencie zatrzymać skrypt i zjadać pamięć.
Aamir Adnan
4
@hyh, nazwa Queuemodułu została zmieniona na queuePython 3. To jest kod Python 2.
Tarnay Kálmán
3
O ile szybciej możesz iść, jeśli chcesz rozmawiać z SAMYM serwerem za każdym razem, utrzymując połączenie? Czy można tego dokonać nawet w wątkach lub za pomocą jednego trwałego połączenia na wątek?
mdurant
2
@mptevsion, jeśli używasz CPython, możesz (na przykład) po prostu zamienić „print status, url” na „my_global_list.append ((status, url))”. Listy (większość operacji na) są domyślnie bezpieczne dla wątków w CPython (i niektórych innych implementacjach Pythona) ze względu na GIL, więc jest to bezpieczne.
Tarnay Kálmán
54

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()
mher
źródło
7
Ten kod używa nieblokujących sieciowych we / wy i nie ma żadnych ograniczeń. Może być skalowany do dziesiątek tysięcy otwartych połączeń. Będzie działał w jednym wątku, ale będzie znacznie szybszy niż każde rozwiązanie do wątkowania. Zamówienie nieblokujące we / wy en.wikipedia.org/wiki/Asynchronous_I/O
mher
1
Czy możesz wyjaśnić, co się tutaj dzieje z globalną zmienną i? Jakiś rodzaj sprawdzania błędów?
LittleBobbyTables
4
Jest to licznik określający, kiedy wyjść z `` ioloop '' - a więc kiedy skończysz.
Michael Dorner,
1
@AndrewScottEvans przyjęto, że używasz Pythona 2.7 i serwerów proxy
Dejell,
5
@Guy Avraham Powodzenia w uzyskiwaniu pomocy w planie ddos.
Walter
51

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())
Glen Thompson
źródło
1
Pytam tylko dlatego, że nie wiem, ale czy te przyszłe rzeczy można zastąpić asynchronicznie / czekać?
TankorSmash
1
Mógłby, ale uważam, że powyższe działa lepiej. możesz użyć aiohttp, ale nie jest to część standardowej biblioteki lib i bardzo się zmienia. To działa, ale po prostu nie znalazłem, żeby to działało. Kiedy go używam, otrzymuję wyższy poziom błędów i przez całe życie nie mogę sprawić, by działał tak dobrze, jak równoczesna przyszłość, chociaż teoretycznie wydaje się, że powinien działać lepiej, patrz: stackoverflow.com/questions/45800857/... jeśli dobrze działa, proszę opublikować swoją odpowiedź, abym mógł ją przetestować.
Glen Thompson,
1
To jest nitpick, ale myślę, że dużo czystsze jest umieszczanie time1 = time.time()na górze pętli for i time2 = time.time()zaraz po pętli for.
Matt M.
Przetestowałem Twój fragment, który w jakiś sposób wykonuje się dwa razy. czy robię coś źle? A może ma działać dwa razy? Jeśli tak jest w tym drugim przypadku, czy możesz mi pomóc zrozumieć, w jaki sposób uruchamia się dwukrotnie?
Ronnie
1
Nie powinien działać dwa razy. Nie jestem pewien, dlaczego to widzisz.
Glen Thompson
40

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.

żelazna żaba
źródło
ironfroggy: Pochylam się w stronę twoich sentymentów. Próbowałem wdrożyć moje rozwiązanie z wątkami i kolejkami (dla automatycznych muteksów), ale czy możesz sobie wyobrazić, ile czasu zajmuje wypełnienie kolejki 100 000 rzeczy? Nadal bawię się różnymi opcjami i sugestiami wszystkich w tym temacie i być może Twisted będzie dobrym rozwiązaniem.
IgorGanapolsky
2
Możesz uniknąć zapełniania w kolejce 100 000 rzeczy. Po prostu przetwarzaj elementy pojedynczo na podstawie danych wejściowych, a następnie uruchom wątek, aby przetworzyć żądanie odpowiadające każdemu elementowi. (Jak opiszę poniżej, użyj wątku uruchamiającego, aby rozpocząć wątki żądania HTTP, gdy liczba wątków spadnie poniżej pewnego progu. Spraw, by wątki zapisały wyniki w adresie URL odwzorowania nagrania do odpowiedzi lub dodały krotki do listy.)
Erik Garrison
ironfroggy: Ciekawe też, jakie wąskie gardła znalazłeś używając wątków Pythona? Jak wątki Pythona współdziałają z jądrem systemu operacyjnego?
Erik Garrison,
Upewnij się, że zainstalowałeś reaktor epoll; w przeciwnym razie będziesz używał opcji select / poll, a to będzie bardzo wolne. Ponadto, jeśli rzeczywiście chcesz jednocześnie otwierać 100 000 połączeń (zakładając, że twój program jest napisany w ten sposób, a adresy URL znajdują się na różnych serwerach), musisz dostroić system operacyjny, aby nie zabrakło deskryptorów plików, efemerycznych portów itp. (prawdopodobnie łatwiej jest po prostu upewnić się, że nie masz więcej niż, powiedzmy, 10 000 zaległych połączeń jednocześnie).
Mark Nottingham,
erikg: poleciłeś świetny pomysł. Jednak najlepszy wynik, jaki udało mi się osiągnąć przy 200 wątkach, to ok. 6 minut. Jestem pewien, że są sposoby na osiągnięcie tego w krótszym czasie ... Mark N: jeśli wybrałem Twisted, to z pewnością przydatny będzie reaktor epoll. Jeśli jednak mój skrypt będzie uruchamiany z wielu komputerów, czy nie wymagałoby to instalacji Twisted na KAŻDEJ maszynie? Nie wiem, czy uda mi się przekonać mojego szefa, by wybrał tę drogę ...
IgorGanapolsky
21

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 .

Marius Stănescu
źródło
Czy jest to podobne do C # async / await i Kotlin Coroutines?
IgorGanapolsky
@IgorGanapolsky, tak, jest bardzo podobny do C # async / czekaj. Nie znam Kotlin Coroutines.
Marius Stănescu
@sandyp, nie jestem pewien, czy to działa, ale jeśli chcesz spróbować, będziesz musiał użyć UnixConnector do aiohttp. Przeczytaj więcej tutaj: docs.aiohttp.org/en/stable/client_reference.html#connectors .
Marius Stănescu
Dzięki @ MariusStănescu. Właśnie tego użyłem.
sandyp
+1 za pokazanie asyncio.gather (* zadania). oto jeden taki fragment, którego użyłem: urls= [fetch(construct_fetch_url(u),idx) for idx, u in enumerate(some_URI_list)] results = await asyncio.gather(*urls)
Ashwini Kumar
19

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]>]
Akshay Pratap Singh
źródło
7
gevent obsługuje teraz python 3
Benjamin Toueg
14
grequests nie jest częścią normalnych próśb i wydaje się być w dużej mierze nieobsługiwane
Thom,
8

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.

Erik Garrison
źródło
1
erikg: czy wrzucenie do twojego równania kolejki byłoby rozsądne (dla blokowania wzajemnego wykluczenia)? Podejrzewam, że GIL Pythona nie jest nastawiony na zabawę z tysiącami wątków.
IgorGanapolsky
Dlaczego potrzebujesz blokady wzajemnego wykluczania, aby zapobiec generowaniu zbyt wielu wątków? Podejrzewam, że źle zrozumiałem ten termin. Możesz śledzić działające wątki w kolejce wątków, usuwając je po ich zakończeniu i dodając więcej do wspomnianego limitu wątków. Ale w prostym przypadku, takim jak ten, można również po prostu obserwować liczbę aktywnych wątków w bieżącym procesie Pythona, czekać, aż spadnie poniżej progu, i uruchamiać więcej wątków do progu zgodnie z opisem. Wydaje mi się, że można uznać to za ukrytą blokadę, ale nie są wymagane żadne wyraźne blokady.
Erik Garrison,
erikg: czy wiele wątków nie udostępnia stanu? Na stronie 305 w książce O'Reilly „Python for Unix and Linux System Administration” napisano: „… używanie wątków bez kolejek sprawia, że ​​jest ono bardziej skomplikowane, niż wiele osób może sobie z tym poradzić realistycznie. O wiele lepszym pomysłem jest zawsze używanie kolejkowania moduł, jeśli uznasz, że musisz użyć wątków. Dlaczego? Ponieważ moduł kolejki również eliminuje potrzebę jawnej ochrony danych za pomocą muteksów, ponieważ sama kolejka jest już wewnętrznie chroniona przez muteks. ” Ponownie cieszę się z twojego punktu widzenia w tej sprawie.
IgorGanapolsky
Igor: Masz absolutną rację, że powinieneś użyć blokady. Zredagowałem post, aby to odzwierciedlić. To powiedziawszy, praktyczne doświadczenie z pythonem sugeruje, że nie musisz blokować struktur danych, które modyfikujesz atomowo ze swoich wątków, takich jak list.append lub przez dodanie klucza skrótu. Uważam, że powodem jest GIL, który zapewnia operacje takie jak lista. Dołącza się z pewnym stopniem atomowości. Obecnie prowadzę test, aby to zweryfikować (użyj wątków 10k, aby dołączyć numery 0-9999 do listy, sprawdź, czy wszystkie załączniki działały). Po prawie 100 iteracjach test nie powiódł się.
Erik Garrison
Igor: Zadano mi kolejne pytanie na ten temat: stackoverflow.com/questions/2740435/…
Erik Garrison
7

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

Rakis
źródło
Rakis: Obecnie szukam asynchronicznych i nieblokujących wejść / wyjść. Muszę się tego lepiej nauczyć przed wdrożeniem. Jednym komentarzem, który chciałbym zamieścić w twoim poście, jest to, że nie jest możliwe (przynajmniej w mojej dystrybucji Linuksa) odrodzenie „tysięcy wątków systemu operacyjnego”. Istnieje maksymalna liczba wątków, które Python pozwoli na odrodzenie przed zerwaniem programu. I w moim przypadku (w CentOS 5) maksymalna liczba wątków wynosi 303.
IgorGanapolsky
Dobrze wiedzieć. Nigdy nie próbowałem odradzać więcej niż kilku naraz w Pythonie naraz, ale spodziewałbym się, że będę w stanie stworzyć więcej niż to, zanim zostanie zbombardowane.
Rakis
6

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
Tarnay Kálmán
źródło
6
Używanie Twisted jako puli wątków ignoruje większość korzyści, jakie możesz z niego uzyskać. Zamiast tego powinieneś używać asynchronicznego klienta HTTP.
Jean-Paul Calderone
1

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.

Kevin Wiskia
źródło
Sugeruję, abyś określił q_size dla ThreadPool w następujący sposób: ThreadPool (wielkość puli, q_size = 1000) Abyś nie miał w pamięci 100000 obiektów WorkRequest. „Jeśli q_size> 0 rozmiar kolejki żądań pracy jest ograniczony, a pula wątków blokuje się, gdy kolejka jest pełna i próbuje umieścić w niej więcej żądań pracy (patrz putRequestmetoda), chyba że użyjesz również timeoutwartości dodatniej putRequest.”
Tarnay Kálmán
Do tej pory próbuję wdrożyć rozwiązanie Threadpool - zgodnie z sugestią. Jednak nie rozumiem listy parametrów w funkcji makeRequests. Co to jest jakiś_nazwalacz, lista_arg_s, oddzwonienie? Może gdybym zobaczył prawdziwy fragment kodu, który by pomógł. Dziwi mnie, że autor tej biblioteki nie opublikował ŻADNYCH przykładów.
IgorGanapolsky
some_callable to twoja funkcja, w której cała praca jest wykonywana (połączenie z serwerem http). list_of_args to argumenty, które zostaną przekazane do some_callabe. callback to funkcja, która zostanie wywołana po zakończeniu wątku roboczego. Wymaga dwóch argumentów, obiektu roboczego (nie musisz się tym naprawdę przejmować) i wyników, które pracownik odzyskał.
Kevin Wiskia
1

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.

George Sovetov
źródło
@IgorGanapolsky Must be. Inaczej byłbym zaskoczony. Ale z pewnością wymaga eksperymentów.
George Sovetov,
0

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)

Mattias Nilsson
źródło
Jedyną rzeczą, o której chciałbym wspomnieć, jest to, że spawnowanie wielu procesów może być droższe niż spawnowanie wielu wątków. Ponadto nie ma wyraźnego wzrostu wydajności w wysyłaniu 100 000 żądań HTTP z wieloma procesami w porównaniu z wieloma wątkami.
IgorGanapolsky
0

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ń).

djangofan
źródło
0

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)
Robᵩ
źródło
0

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])
RDRR
źródło
-2

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.

pestilence669
źródło
7
Wątki Pythona są całkiem realne, w przeciwieństwie do na przykład Ruby. Pod maską są one implementowane jako natywne wątki systemu operacyjnego, przynajmniej w systemach Unix / Linux i Windows. Może masz na myśli GIL, ale to nie czyni wątków mniej realnymi ...
Eli Bendersky
2
Eli ma rację co do wątków Pythona, ale argument Pestilence, że chcesz użyć puli wątków, jest również poprawny. Ostatnią rzeczą, którą chcesz zrobić w tym przypadku, jest uruchomienie osobnego wątku dla każdego z 100 000 żądań jednocześnie.
Adam Crossland
1
Igor, nie możesz rozsądnie zamieszczać fragmentów kodu w komentarzach, ale możesz edytować swoje pytanie i tam je dodać.
Adam Crossland
Zarazy: ile kolejek i wątków na kolejkę poleciłbyś dla mojego rozwiązania?
IgorGanapolsky
Plus jest to I / O związany zadanie nie CPU związana, w dużej mierze wpływa na zadania GIL CPU związany
PirateApp