在数据驱动的时代，爬虫是获取信息的重要工具。传统的同步爬虫（如requests）在遇到大量URL时，会因I/O等待而效率低下。Python的异步编程（asyncio + aiohttp）能让我们在单线程内并发处理网络请求，显著提升抓取速度。本文将带你从零构建一个可用的异步爬虫，并解决实际中的限速、错误处理等难题。
异步编程的核心是事件循环（Event Loop）。它不断检查待执行的任务，当遇到I/O操作（如网络请求）时，不阻塞线程，而是注册回调并继续执行其他任务。协程（Coroutine）是可暂停和恢复的函数，用async def定义，用await等待另一个协程或异步操作。
import asyncio
async def fetch_url(url):
print(f"开始请求: {url}")
await asyncio.sleep(1) # 模拟网络延迟
print(f"完成请求: {url}")
return f"数据 from {url}"
async def main():
tasks = [fetch_url(f"https://example.com/page/{i}") for i in range(5)]
results = await asyncio.gather(*tasks)
print(results)
asyncio.run(main())aiohttp是Python最流行的异步HTTP库。它提供客户端会话（ClientSession），自动管理连接池和请求复用。
import aiohttp
import asyncio
async def fetch(session, url):
async with session.get(url) as response:
return await response.text()
async def main():
async with aiohttp.ClientSession() as session:
html = await fetch(session, "https://httpbin.org/get")
print(html[:200])
asyncio.run(main())假设我们需要抓取一个分页API（如JSONPlaceholder），共10页，每页10条帖子。同步方式需要等待每个请求完成，而异步可以同时发起所有请求。
import aiohttp
import asyncio
BASE_URL = "https://jsonplaceholder.typicode.com/posts"
async def fetch_page(session, page):
params = {"_page": page, "_limit": 10}
async with session.get(BASE_URL, params=params) as resp:
return await resp.json()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, page) for page in range(1, 11)]
results = await asyncio.gather(*tasks)
for page_data in results:
print(f"获取到 {len(page_data)} 条帖子")
asyncio.run(main())这段代码在几秒内就能完成10个请求，而同步版可能需要等待每个请求的延迟。
如果目标服务器有反爬或限流，并发过多可能导致IP被封。我们可以使用asyncio.Semaphore控制同时进行的请求数。
import aiohttp
import asyncio
BASE_URL = "https://jsonplaceholder.typicode.com/posts"
semaphore = asyncio.Semaphore(3) # 最多同时3个请求
async def fetch_page(session, page):
async with semaphore:
params = {"_page": page, "_limit": 10}
async with session.get(BASE_URL, params=params) as resp:
return await resp.json()
async def main():
async with aiohttp.ClientSession() as session:
tasks = [fetch_page(session, page) for page in range(1, 11)]
results = await asyncio.gather(*tasks)
for page_data in results:
print(f"获取到 {len(page_data)} 条帖子")
asyncio.run(main())网络请求可能失败，我们需要优雅地处理。可以包装一个重试装饰器：
import asyncio
import aiohttp
from functools import wraps
def retry(max_retries=3, delay=1):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
for attempt in range(max_retries):
try:
return await func(*args, **kwargs)
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
if attempt == max_retries - 1:
raise
print(f"请求失败，重试 {attempt+1}/{max_retries}: {e}")
await asyncio.sleep(delay * (attempt + 1))
return None
return wrapper
return decorator
@retry(max_retries=3, delay=1)
async def fetch_page(session, page):
params = {"_page": page, "_limit": 10}
async with session.get(BASE_URL, params=params, timeout=aiohttp.ClientTimeout(total=10)) as resp:
resp.raise_for_status()
return await resp.json()抓取的数据通常需要解析（如HTML/JSON）并持久化。以下示例将结果保存到CSV文件：
import csv
import asyncio
import aiohttp
async def fetch_and_parse(session, page, writer):
params = {"_page": page, "_limit": 10}
async with session.get(BASE_URL, params=params) as resp:
data = await resp.json()
for item in data:
writer.writerow([item["id"], item["title"], item["body"]])
async def main():
async with aiohttp.ClientSession() as session:
with open("posts.csv", "w", newline="", encoding="utf-8") as f:
writer = csv.writer(f)
writer.writerow(["ID", "Title", "Body"])
tasks = [fetch_and_parse(session, page, writer) for page in range(1, 11)]
await asyncio.gather(*tasks)
asyncio.run(main())我们做一个简单测试：抓取20个URL（模拟延迟500ms），同步版需要约10秒（20*0.5），而异步版（无并发限制）只需约0.5秒。实际中受网络带宽和服务器限制，但提升依然显著。
ClientSession作为上下文管理器，它维护连接池，避免每次请求都创建新连接。ClientTimeout防止某个请求挂起。ClientSession正确关闭（使用async with）。本文从异步编程基础出发，逐步构建了一个完整的异步爬虫，涵盖了并发控制、错误处理和结果存储。aiohttp结合asyncio是Python爬虫提速的利器，但需要谨慎处理资源限制和服务器负载。希望你能将这些技术应用到实际项目中，高效获取所需数据。
专注技术分享与实战