在 Python 中使用 aiohttp 配置代理

在 Python 中使用 aiohttp 配置代理

aiohttp 与代理

aiohttp 是 Python 最流行的异步 HTTP 库。它可以发送数千个并发请求，非常适合数据抓取与自动化。aiohttp 的代理支持是大规模任务的关键特性。

基本代理用法

HTTP 代理

在 aiohttp 中使用 HTTP 代理最简单的方式，是在请求方法中传入 proxy 参数：

import aiohttp
import asyncio

async def fetch_with_proxy():
    proxy = "http://proxy_ip:port"
    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://httpbin.org/ip",
            proxy=proxy
        ) as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_proxy())

带认证的代理

对于带用户名和密码的代理，使用 BasicAuth：

import aiohttp
import asyncio

async def fetch_with_auth_proxy():
    proxy = "http://proxy_ip:port"
    proxy_auth = aiohttp.BasicAuth("username", "password")

    async with aiohttp.ClientSession() as session:
        async with session.get(
            "https://httpbin.org/ip",
            proxy=proxy,
            proxy_auth=proxy_auth
        ) as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_auth_proxy())

另一种写法是把凭据放进 URL：

proxy = "http://username:password@proxy_ip:port"

SOCKS5 代理

安装 aiohttp-socks

aiohttp 原生不支持 SOCKS。请安装 aiohttp-socks：

pip install aiohttp-socks

用法

import aiohttp
from aiohttp_socks import ProxyConnector

async def fetch_with_socks():
    connector = ProxyConnector.from_url("socks5://user:pass@proxy_ip:port")

    async with aiohttp.ClientSession(connector=connector) as session:
        async with session.get("https://httpbin.org/ip") as response:
            data = await response.json()
            print(data)

asyncio.run(fetch_with_socks())

支持的协议

ProxyConnector 支持：
- socks5://
- socks4://
- http://
- https://

代理轮换

简单轮换

import aiohttp
import asyncio
import random

PROXIES = [
    "http://user:pass@proxy1:port",
    "http://user:pass@proxy2:port",
    "http://user:pass@proxy3:port",
]

async def fetch_with_rotation(url):
    proxy = random.choice(PROXIES)
    async with aiohttp.ClientSession() as session:
        async with session.get(url, proxy=proxy) as response:
            return await response.text()

async def main():
    urls = ["https://example.com"] * 10
    tasks = [fetch_with_rotation(url) for url in urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for r in results:
        if isinstance(r, Exception):
            print(f"Error: {r}")
        else:
            print(f"OK: {len(r)} bytes")

asyncio.run(main())

剔除失效代理的轮换

import aiohttp
import asyncio
from collections import deque

class ProxyRotator:
    def __init__(self, proxies):
        self.proxies = deque(proxies)
        self.failed = set()

    def get_proxy(self):
        for _ in range(len(self.proxies)):
            proxy = self.proxies[0]
            self.proxies.rotate(-1)
            if proxy not in self.failed:
                return proxy
        raise Exception("All proxies failed")

    def mark_failed(self, proxy):
        self.failed.add(proxy)

    def mark_success(self, proxy):
        self.failed.discard(proxy)

async def fetch(session, url, rotator, retries=3):
    for attempt in range(retries):
        proxy = rotator.get_proxy()
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=10)) as resp:
                if resp.status == 200:
                    rotator.mark_success(proxy)
                    return await resp.text()
                elif resp.status == 403:
                    rotator.mark_failed(proxy)
        except Exception:
            rotator.mark_failed(proxy)
    return None

带限制的并发请求

用信号量控制并发

import aiohttp
import asyncio

async def fetch(session, url, proxy, semaphore):
    async with semaphore:
        try:
            async with session.get(url, proxy=proxy, timeout=aiohttp.ClientTimeout(total=15)) as resp:
                return await resp.text()
        except Exception as e:
            return None

async def main():
    urls = [f"https://example.com/page/{i}" for i in range(100)]
    proxy = "http://user:pass@proxy:port"
    semaphore = asyncio.Semaphore(10)  # 最多 10 个并发

    async with aiohttp.ClientSession() as session:
        tasks = [fetch(session, url, proxy, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)
        success = sum(1 for r in results if r)
        print(f"Success: {success}/{len(urls)}")

asyncio.run(main())

会话配置

超时

timeout = aiohttp.ClientTimeout(
    total=30,        # 总超时
    connect=10,      # 连接超时
    sock_read=10     # socket 读取超时
)
session = aiohttp.ClientSession(timeout=timeout)

请求头

headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "Accept": "text/html,application/xhtml+xml",
    "Accept-Language": "en-US,en;q=0.9",
}
session = aiohttp.ClientSession(headers=headers)

SSL

# 关闭 SSL 校验（仅用于测试）
import ssl
ssl_context = ssl.create_default_context()
ssl_context.check_hostname = False
ssl_context.verify_mode = ssl.CERT_NONE

session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=ssl_context))

错误处理

import aiohttp

async def safe_fetch(session, url, proxy):
    try:
        async with session.get(url, proxy=proxy) as resp:
            if resp.status == 200:
                return await resp.text()
            elif resp.status == 407:
                print("Proxy authentication required")
            elif resp.status == 403:
                print("Forbidden - proxy might be banned")
            elif resp.status == 429:
                print("Rate limited - slow down")
            return None
    except aiohttp.ClientProxyConnectionError:
        print("Cannot connect to proxy")
    except aiohttp.ClientConnectorError:
        print("Connection error")
    except asyncio.TimeoutError:
        print("Request timed out")
    except Exception as e:
        print(f"Unexpected error: {e}")
    return None

aiohttp 与 requests 对比

结论

在 Python 中进行异步代理请求，aiohttp 是最佳选择。它对 HTTP 和 SOCKS5（通过 aiohttp-socks）的支持、代理轮换、并发控制和错误处理，使其成为大规模抓取与自动化的强大工具。