Asynchroniczne skrobanie w Pythonie z httpx, aiohttp i proxy

Asynchroniczność zamienia wolny skrobak w szybki — a szybki skrobak w zablokowany, chyba że dobrze ustawisz proxy, semafory i limity czasowe. Oto cały wzorzec dla httpx i aiohttp, z kodem.

Jeśli twój skrobak większość czasu oczekiwania spędza na sieci, asynchroniczność jest największym możliwym przyspieszeniem — a proxy to to, co chroni cię przed zablokowaniem. Ten przewodnik obejmuje asynchroniczne skrobanie w Pythonie z httpx i aiohttp oraz proxy od początku do końca: jak każda biblioteka ustawia proxy, jak ograniczyć współbieżność za pomocą semafora, jak ustawić limity czasowe, które faktycznie działają, i jak rotować adresy IP bez niszczenia sesji. Wszystko z działającym kodem i przykładowymi danymi uwierzytelniającymi, które zamieniasz na własne.

Dlaczego asynchroniczność i gdzie pasują proxy

Standardowe requests jest blokujące: każde wywołanie czeka na odpowiedź przed rozpoczęciem następnego. Pobierz 500 stron i płacisz za 500 podróży tam i z powrotem. Asynchroniczność pobiera je równocześnie na jednej pętli zdarzeń, więc całkowity czas zbliża się do najwolniejszego pojedynczego żądania zamiast sumy. Haczyk: wybuch jednoczesnych żądań z jednego IP to dokładnie sygnatura, na którą polują systemy antybotowe. Rozwiązanie to nie zwalniać do pełzania — to rozłożyć ruch na rotującą pulę proxy i celowo ograniczać za pomocą semafora.

Ustaw proxy w httpx (async)

httpx jest pragmatycznym domyślnym wyborem, ponieważ jeden model klienta obsługuje zarówno synchronizację, jak i asynchroniczność, a także obsługuje HTTP/2. Zwróć uwagę na nowoczesne API: to pojedynczy argument proxy= na kliencie, a nie stara lista proxies= — częste źródło zamieszania "dlaczego moje proxy jest ignorowane" po aktualizacji. Dane uwierzytelniające trafiają bezpośrednio do URL proxy.

import asyncio, httpx

PROXY = "http://USER:PASS@gate.quantumproxies.io:8000"

async def fetch(client, url):
    r = await client.get(url, timeout=httpx.Timeout(20.0))
    return url, r.status_code, r.text

async def main(urls):
    async with httpx.AsyncClient(proxy=PROXY, http2=True) as client:
        tasks = [fetch(client, u) for u in urls]
        return await asyncio.gather(*tasks)

urls = ["https://httpbin.org/ip"] * 5
print(asyncio.run(main(urls)))

Jeden klient, używany ponownie przy każdym żądaniu, to klucz — utrzymuje ciepłą pulę połączeń, dzięki czemu omijasz powtarzające się uściski TLS przez proxy. Tworzenie nowego klienta przy każdym żądaniu to najczęstszy błąd wydajności asynchronicznej: wyrzuca ponowne użycie połączenia i stan ciasteczek przy każdym wywołaniu.

Ustaw proxy w aiohttp (na żądanie)

aiohttp jest natywny dla asyncio i daje najdokładniejszą kontrolę nad współbieżnością, ale jego konwencja proxy różni się: proxy jest przekazywane na żądanie w session.get(), a nie na sesji. To faktycznie wygodne dla rotacji. Dwie rzeczy gryzą tu ludzi — domyślny User-Agent to dosłownie Python/3.x aiohttp/3.x, co jest oczywistym sygnałem bota, który musisz nadpisać, i powinieneś jawnie określić rozmiar puli połączeń.

import aiohttp, asyncio

PROXY = "http://USER:PASS@gate.quantumproxies.io:8000"
HEADERS = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
           "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/132.0.0.0 Safari/537.36"}

async def fetch(session, url):
    timeout = aiohttp.ClientTimeout(total=30, connect=10, sock_read=20)
    async with session.get(url, proxy=PROXY, timeout=timeout) as resp:
        return url, resp.status, await resp.text()

async def main(urls):
    conn = aiohttp.TCPConnector(limit=20, limit_per_host=8)
    async with aiohttp.ClientSession(connector=conn, headers=HEADERS) as session:
        return await asyncio.gather(*(fetch(session, u) for u in urls))

print(asyncio.run(main(["https://httpbin.org/ip"] * 5)))

TCPConnector(limit=20, limit_per_host=8) ogranicza całkowitą liczbę otwartych połączeń i połączeń do jednego hosta — to pierwsza linia grzeczności, która zapobiega pochłonięciu całej puli przez jeden cel.

Ogranicz współbieżność za pomocą semafora

Wysyłanie nieograniczonej liczby jednoczesnych żądań to najszybszy sposób na spalenie puli proxy i przekroczenie limitów. asyncio.Semaphore to przepustnica: ogranicza liczbę żądań w locie jednocześnie, niezależnie od liczby zadań w kolejce. Ustaw globalny limit, a dla agresywnych zadań także limit na domenę.

sem = asyncio.Semaphore(15)  # never more than 15 requests in flight

async def guarded_fetch(client, url):
    async with sem:
        return await fetch(client, url)

async def main(urls):
    async with httpx.AsyncClient(proxy=PROXY) as client:
        return await asyncio.gather(*(guarded_fetch(client, u) for u in urls))

Odpowiednia liczba zależy od tolerancji celu i rozmiaru twojej puli, a nie od tego, jak szybko może działać twoja maszyna. Zacznij konserwatywnie (10–20), obserwuj wskaźnik blokad i zwiększaj go tylko wtedy, gdy sukces pozostaje wysoki. Na jednym lepkim IP trzymaj się jednocyfrowych wartości.

Limity czasowe: modeluj każdą fazę

Żądanie z proxy zawodzi w większej liczbie miejsc niż bezpośrednie — rozwiązywanie DNS, łączenie się z proxy, konfiguracja tunelu, łączenie się z celem, oczekiwanie na nagłówki i odczyt ciała to wszystkie odrębne zastoje. Jeden ogólny limit czasowy ukrywa, która faza się zawiesiła. aiohttp's ClientTimeout(total=, connect=, sock_read=) i httpx's Timeout() pozwalają ograniczyć je osobno. Jedna zasada bez wyjątków: nigdy nie wysyłaj żądania bez limitu czasowego, bo jedno martwe wyjście zawiesi korutynę na zawsze i cicho zagłodzi twoją pętlę zdarzeń.

Rotuj proxy bez przerywania sesji

Istnieją dwie strategie rotacji, a wybór niewłaściwej niszczy twoje dane. Losowa rotacja na żądanie jest idealna dla bezstanowych pobrań stron. Ale niszczy każdy przepływ, który zależy od ciasteczek, logowania lub lokalizacji, ponieważ drugie żądanie trafia na inne IP niż pierwsze. Czysty podział: rotuj na granicy jednostki logicznej — jedno IP na segment skrobania lub na konto — i użyj rotującej bramy, która automatycznie dostarcza ci świeże wyjście, dzięki czemu twój kod nigdy nie zarządza listą.

# rotating gateway: one endpoint, new exit IP per request
ROT = "http://USER:PASS@rotating.quantumproxies.io:8000"

# sticky session: same IP for a multi-step flow, tag the session id
STICKY = "http://USER-session-a1b2:PASS@gate.quantumproxies.io:8000"

async def crawl_segment(urls):
    async with httpx.AsyncClient(proxy=ROT) as client:  # rotates per call
        return await asyncio.gather(*(fetch(client, u) for u in urls))

Rotująca brama rezydencyjna jest pragmatycznym domyślnym wyborem dla skrobania o wysokiej współbieżności: rotacja na żądanie wśród 90M+ IP w 200+ krajach, z lepkimi sesjami, gdy koszyk lub logowanie wymaga tego samego wyjścia przez kilka minut. Jeśli zastanawiasz się nad wyborem między rezydencyjnymi a ISP lub centrami danych dla zadania, nasz przewodnik który typ proxy użyć przedstawia kompromisy.

Zdobądź rotującą bramę rezydencyjną

Ponowne próby, wycofanie i jitter

Martwe wyjścia i przejściowe blokady są normalne na dużą skalę, a nie wyjątkowe. Ale naiwne ponowne próby pogarszają sytuację: gdy 50 asynchronicznych zadań zawiedzie w tym samym momencie i wszystkie natychmiast próbują ponownie, wywołujesz zsynchronizowany wybuch, który uderza w cel mocniej niż pierwotny przebieg. Dodaj wykładnicze wycofanie plus losowy jitter, aby ponowne próby się rozproszyły, ogranicz próby i rotuj IP w przypadku niepowodzenia zamiast ponownie używać spalonego.

import random
from httpx import HTTPError

async def robust_fetch(client, url, tries=3):
    for attempt in range(tries):
        try:
            r = await client.get(url, timeout=httpx.Timeout(20.0))
            if r.status_code < 400 and looks_real(r.text):
                return r
        except HTTPError:
            pass
        # exponential backoff + jitter before the next attempt
        await asyncio.sleep((2 ** attempt) + random.uniform(0, 1))
    return None

200 to nie sukces

Najsubtelniejszy błąd asynchronicznego skrobania to traktowanie HTTP 200 jako zakończonego. Systemy antybotowe zwracają 200 z CAPTCHA, powiadomieniem o odmowie dostępu, pustym zestawem wyników lub wyzwaniem JS — więc proxy, które ocenia "sukces" tylko na podstawie kodu statusu, cicho dostarcza ci zablokowane strony. Zweryfikuj zawartość: sprawdź znany element, minimalną długość lub brak znaczników wyzwań, zanim zaufasz odpowiedzi. To bramka looks_real() w pętli ponownych prób powyżej.

Kiedy przestać samodzielnie budować stos

Powyższy wzorzec — asynchroniczny klient, semafor, limity czasowe, rotacja, walidacja zawartości — obsługuje większość celów czysto. Ale gdy strona nakłada Cloudflare, odciski palców TLS lub ciężkie renderowanie po stronie klienta, surowy asynchroniczny HTTP zaczyna przegrywać niezależnie od twojego proxy, ponieważ uścisk TLS w Pythonie nie wygląda jak w Chrome. Na tej granicy Scraper API, który przenosi prawdziwy odcisk palca przeglądarki, rotuje IP i renderuje JavaScript na żądanie, to mniej kodu i wyższy wskaźnik sukcesu niż utrzymywanie tego wszystkiego ręcznie. Nasz post na temat kosztów headless vs HTTP obejmuje, gdzie ta eskalacja się opłaca.

Często zadawane pytania

Jak używać proxy z aiohttp?

Przekaż URL proxy na żądanie: session.get(url, proxy="http://user:pass@host:port"). W przeciwieństwie do requests, aiohttp nie przyjmuje listy proxy na sesji. Zawsze nadpisuj domyślny User-Agent (Python/3.x aiohttp/3.x to oczywisty sygnał bota) i ustaw ClientTimeout, aby martwe wyjście nie mogło zawiesić korutyny.

Czy httpx czy aiohttp jest lepszy do asynchronicznego skrobania?

Sięgnij po httpx jako domyślny wybór: jeden klient działa synchronicznie i asynchronicznie, HTTP/2 jest wbudowany, a proxy to pojedynczy argument. Wybierz aiohttp, gdy chcesz maksymalnej kontroli nad współbieżnością — jawne limity puli połączeń i proxy na żądanie pasują do dużych, szybkich skrobanek. Oba są w porządku; strategia proxy ma większe znaczenie niż biblioteka.

Ile jednoczesnych żądań powinienem uruchomić?

Nie tyle, ile pozwala twoja maszyna — tyle, ile toleruje cel i twoja pula. Zacznij od semafora z 10–20 w locie, obserwuj wskaźnik blokad i błędów, i zwiększaj go tylko wtedy, gdy sukces pozostaje wysoki. Na jednym lepkim IP trzymaj się jednocyfrowych wartości. Szersza rotacja IP pozwala na bezpieczne uruchamianie większej całkowitej współbieżności.

Dlaczego mój asynchroniczny skrobak jest blokowany, gdy synchronizowany nie był?

Ponieważ współbieżność koncentruje sygnał: wiele jednoczesnych żądań z jednego IP to klasyczny wzorzec bota. Rozłóż obciążenie na rotującą pulę, ograniczaj za pomocą semafora, dodaj jitterowane wycofanie przy ponownych próbach i zweryfikuj zawartość odpowiedzi — 200 może nadal być stroną z wyzwaniem. Prędkość bez rotacji to to, co cię zablokowało.

To pełny wzorzec asynchroniczny: wybierz klienta, ustaw proxy w odpowiedni sposób, ogranicz współbieżność za pomocą semafora, ogranicz każdą fazę limitu czasowego, rotuj na granicy logicznej i nigdy nie ufaj gołemu 200. Najpierw popraw warstwę proxy, a większość listy blokad zniknie, zanim ją osiągniesz.

Wypróbuj QuantumProxies Scraper API