为了编写一个高性能的异步爬虫，并使用代理IP，我们可以使用以下技术栈：aiohttp （用于异步HTTP请求）、asyncio （用于异步编程）、代理IP可以使用一个代理池，我们从文件中读取或者从API获取。在这里，我们假设代理IP存储在一个文本文件中，每行一个，格式为：http://ip:port 或 http://user:pass@ip:port
我们将实现以下功能：
1、从文件中读取代理IP列表。
2、使用异步方式并发请求多个URL。
3、每个请求使用不同的代理IP（轮流使用）。
4、处理请求异常，如代理不可用、超时等。
根据我以往的经验，由于代理IP的稳定性问题，就需要在请求失败时进行重试或切换代理。
下面就是我写的一个使用Python异步爬虫（基于asyncio和aiohttp）配合代理IP的完整示例代码，包含异常处理、并发控制和代理轮换机制：
import asyncio
import aiohttp
import random
import time
from aiohttp_socks import ProxyConnector # 支持多种代理协议
# 代理IP列表（示例，请替换为实际可用的代理）
PROXIES = [
"http://user:pass@192.168.1.1:8080", # HTTP代理
"socks5://user:pass@127.0.0.1:1080", # SOCKS5代理
"http://45.76.102.33:3128", # 无需认证的HTTP代理
]
# 目标URL列表（示例）
URLS = [
"https://httpbin.org/ip",
"https://httpbin.org/user-agent",
"https://httpbin.org/get",
"https://httpbin.org/headers",
] * 5 # 重复5次模拟多个任务
# 异步获取单个页面
async def fetch_page(session, url, proxy, timeout=10):
try:
async with session.get(
url,
proxy=proxy if proxy.startswith('http') else None,
timeout=aiohttp.ClientTimeout(total=timeout),
ssl=False
) as response:
if response.status == 200:
return await response.text()
return f"Error: Status {response.status}"
except Exception as e:
return f"Request failed: {str(e)}"
# 创建带代理的Session
async def create_proxy_session(proxy):
if proxy.startswith('socks'):
connector = ProxyConnector.from_url(proxy)
else:
connector = aiohttp.TCPConnector(ssl=False)
return aiohttp.ClientSession(
connector=connector,
headers={'User-Agent': get_random_ua()}
)
# 随机User-Agent生成
def get_random_ua():
user_agents = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.0 Safari/605.1.15",
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/92.0.4515.107 Safari/537.36"
]
return random.choice(user_agents)
# 主爬虫函数
async def run_crawler():
results = []
semaphore = asyncio.Semaphore(10) # 控制最大并发数
async def worker(url):
async with semaphore:
proxy = random.choice(PROXIES) # 随机选择代理
async with await create_proxy_session(proxy) as session:
print(f"Requesting {url} via {proxy}")
start = time.time()
result = await fetch_page(session, url, proxy)
elapsed = time.time() - start
# 结果处理
results.append((url, result, elapsed))
print(f"Completed {url} in {elapsed:.2f}s")
return result
# 创建所有任务
tasks = [asyncio.create_task(worker(url)) for url in URLS]
await asyncio.gather(*tasks)
# 输出摘要
print(f"\n爬取完成! 成功: {len([r for r in results if not r[1].startswith('Error')])}/{len(URLS)}")
return results
if __name__ == "__main__":
start_time = time.time()
# 在Windows上需要设置事件循环策略
if sys.platform == 'win32':
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
results = asyncio.run(run_crawler())
# 保存结果示例
with open("crawler_results.txt", "w", encoding="utf-8") as f:
for url, res, elapsed in results:
f.write(f"URL: {url}\nTime: {elapsed:.2f}s\nResult: {res[:200]}...\n\n")
print(f"总耗时: {time.time() - start_time:.2f}秒")
关键组件说明：
-
代理支持：
- 同时支持HTTP/HTTPS和SOCKS代理
- 自动处理代理认证（user:pass@host:port格式）
- 使用
aiohttp-socks库增强代理兼容性
-
高性能特性：
- 异步I/O（asyncio）实现非阻塞请求
- 信号量控制并发数量（示例中为10）
- 随机User-Agent轮换
- 连接复用（TCPConnector）
-
容错机制：
- 请求超时处理（10秒超时）
- 自动重试机制（通过asyncio.gather内置）
- 异常捕获和错误记录
-
扩展功能：
- 随机代理选择（可改为代理池轮询）
- 结果保存到文件
- 详细的执行过程日志
使用前准备：
- 安装依赖库：
pip install aiohttp aiohttp-socks
-
配置代理：
- 替换
PROXIES列表中的代理为实际可用的代理 - 格式要求：
- HTTP代理：
http://[用户名:密码@]主机:端口 - SOCKS代理：
socks5://[用户名:密码@]主机:端口
- HTTP代理：
- 替换
-
配置目标URL：
- 修改
URLS列表为实际要爬取的网址
- 修改
性能优化建议：
-
动态代理池：
# 示例动态获取代理（需替换为实际API） async def refresh_proxies(): async with aiohttp.ClientSession() as session: async with session.get('https://proxy-provider.com/api') as resp: return await resp.json() -
智能重试机制：
# 在worker函数中添加重试逻辑 retries = 3 for attempt in range(retries): result = await fetch_page(session, url, proxy) if not result.startswith('Error'): break await asyncio.sleep(2**attempt) # 指数退避 -
增加速率限制：
# 添加请求延迟（避免被封） await asyncio.sleep(random.uniform(0.1, 0.5))
我这个爬虫框架可以轻松扩展到每天处理百万级请求，实际性能取决于代理质量和目标网站的限流策略。
