最近在做一个舆情监控小工具，需要从几个新闻网站抓取标题和正文。一开始用 requests 写同步爬虫，一趟跑下来要将近十分钟，实在忍不了。后来改用 aiohttp 做异步并发，时间直接缩到 30 秒。今天我把这个异步爬虫的完整思路写出来，代码拿过去改改就能用。
你可能会想，aiohttp 不是经常容易踩坑吗？其实只要处理好连接池、重试和信号量，它比想象中要稳得多。我会详细讲到这几个方面，并且附上完整的项目结构。
爬虫目标
假设我们要抓取某个新闻站点的列表页，然后进入每条新闻的详情页提取正文。这里以我本地起的服务为例，接口返回 HTML 片段。重点不在目标网站，而是异步爬虫的写法。
一个典型的列表页包含若干链接，我们把它简化成下面这样：
<div class="news-list">
<a href="/news/1" rel="external nofollow" >标题1</a>
<a href="/news/2" rel="external nofollow" >标题2</a>
<a href="/news/3" rel="external nofollow" >标题3</a>
</div>
详情页则是一段 HTML，里面包含 <h1> 和 <div class="content">。
最终我们需要的数据结构是：[{url, title, content}, ...]，并存成 JSON 文件。
为什么用 aiohttp 而不是 requests
requests 是同步阻塞的，发一个请求必须等回应才能继续。而 aiohttp 基于 asyncio，可以在等待网络响应期间切换到其他协程，从而实现高并发。
但要注意，aiohttp 默认的并发控制需要自己加信号量（semaphore），否则会一口气创建几千个连接，把目标网站打死，也可能自己撑不住。
完整代码：异步新闻爬虫
我习惯把爬虫分成三个部分：一个是负责请求的模块，一个是负责解析的模块，最后是控制入口。为了简化，下面全部写在一个文件里。
import asyncio
import json
import logging
from typing import List, Tuple
import aiohttp
from bs4 import BeautifulSoup
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# 配置区
BASE_URL = "https://example-news.com"
LIST_API = f"{BASE_URL}/news"
CONCURRENCY = 5 # 同时最多 5 个请求
MAX_RETRIES = 3
TIMEOUT_SECONDS = 10
OUTPUT_FILE = "news.json"
# 公共请求头
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html,application/xhtml+xml",
}
这里设置了一个信号量，控制并发数量。然后定义一个通用的请求函数，内置重试机制。
async def fetch_html(session: aiohttp.ClientSession, url: str, semaphore: asyncio.Semaphore) -> str:
"""
发起GET请求并返回HTML文本。失败时按指数退避重试。
"""
async with semaphore:
for attempt in range(1, MAX_RETRIES + 1):
try:
async with session.get(url, headers=HEADERS, timeout=aiohttp.ClientTimeout(total=TIMEOUT_SECONDS)) as resp:
resp.raise_for_status()
# 有些网站的编码是GBK，如果乱码可以改成 resp.charset 或手动指定
return await resp.text(encoding='utf-8', errors='ignore')
except (aiohttp.ClientError, asyncio.TimeoutError) as e:
logger.warning("请求失败 %s，第 %s/次 错误：%s", url, attempt, e)
if attempt == MAX_RETRIES:
raise
await asyncio.sleep(attempt * 0.5) # 退避 0.5s, 1s, 1.5s
# 不会走到这里
解析列表页，提取所有详情页的路径：
def parse_list(html: str, base_url: str) -> List[str]:
soup = BeautifulSoup(html, 'html.parser')
links = []
for a in soup.select('.news-list a[href]'):
href = a['href']
if href.startswith('http'):
links.append(href)
else:
links.append(base_url.rstrip('/') + href)
return links
解析详情页，返回标题和正文：
def parse_detail(html: str) -> Tuple[str, str]:
soup = BeautifulSoup(html, 'html.parser')
title = soup.select_one('h1').get_text(strip=True)
content_div = soup.select_one('div.content')
content = content_div.get_text('n', strip=True) if content_div else ''
return title, content
处理单条新闻，获取详情并解析：
async def fetch_news_detail(session, url, semaphore):
try:
html = await fetch_html(session, url, semaphore)
title, content = parse_detail(html)
return {"url": url, "title": title, "content": content}
except Exception as e:
logger.error("解析失败 %s: %s", url, e)
return None
主函数：先抓列表页，再并发抓取详情页：
async def main():
async with aiohttp.ClientSession() as session:
# 信号量控制并发数
semaphore = asyncio.Semaphore(CONCURRENCY)
# 1. 抓取列表页
logger.info("开始抓取列表页: %s", LIST_API)
try:
list_html = await fetch_html(session, LIST_API, semaphore)
except Exception as e:
logger.error("列表页最终失败: %s", e)
return
news_urls = parse_list(list_html, BASE_URL)
logger.info("发现 %d 条新闻", len(news_urls))
# 2. 并发抓取详情页
tasks = [fetch_news_detail(session, url, semaphore) for url in news_urls]
results = await asyncio.gather(*tasks)
# 过滤掉失败任务
valid_results = [res for res in results if res is not None]
logger.info("成功抓取 %d 条新闻", len(valid_results))
# 3. 保存到JSON
with open(OUTPUT_FILE, 'w', encoding='utf-8') as f:
json.dump(valid_results, f, ensure_ascii=False, indent=2)
logger.info("数据已保存到 %s", OUTPUT_FILE)
if __name__ == "__main__":
asyncio.run(main())
这个爬虫执行流程很清晰：请求列表页 → 解析出所有URL → 创建协程任务 → 信号量限制并发 → gather等待全部完成。如果你需要跑几万个URL，建议用 asyncio.Queue 来做生产者消费者模型，这里简单场景已经够了。
实战中的几个关键问题
1. 连接池大小与并发数的取舍
aiohttp 默认的 TCPConnector 并发限制是 100，但如果你设置信号量为 10，实际并发不会超过 10。另一种做法是直接在 ClientSession 上设置 connector=TCPConnector(limit=10)，但那样信号的灵活性会差一些。我习惯在请求函数内部用 Semaphore，这样不同请求类型可以共用同一个session但拥有不同并发度。
2. 编码问题
我一开始抓某个网站时全是乱码，后来发现是GBK编码。aiohttp 返回的 resp.text(encoding='gbk') 能解决，但如果你不指定，它会尝试从响应头里找charset。我建议统一在请求函数里加一个encoding参数，根据实际网站调整。
3. HTTP 状态码伪装
有些网站会检查 source/Referer 头。最简单的办法是增加 Referer 和 User-Agent。我这里只加了 User-Agent，实际使用中可能还要添加别的。
4. 重试时的幂等性
如果目标接口只允许 GET，重试是安全的。但如果 POST，必须确保请求体可以被重复发送，否则可能产生重复数据。好在我们这里是 GET，所以直接重试没问题。
5. 定时限速
信号量控制的是同时进行的数量，但如果想让每次请求之间隔一个固定时间，可以再包一层 asyncio.sleep。比如平均 100ms 一个请求，就在 async with semaphore 之后加一句 await asyncio.sleep(0.1)。不过我这个示例没有加，因为信号量已经足够。
测试时我踩过的一个大坑
我一开始把信号量放在函数内部创建，结果每个任务都创建了一个新的信号量，并发控制完全失效。正确的做法是在 main 里创建一次，然后传进请求函数。这一点我在代码里已经做了。
另外，asyncio.gather(*tasks) 默认不会取消其他任务，如果其中一个任务抛出异常，异常会在 gather 等待所有任务完成后才抛出。所以我把每个任务内部的异常都 catch 掉并返回 None，这样 gather 永远不会失败。这个方法虽然损失了错误详情，但保证了整体稳定性。
进一步优化：将爬虫重构为可复用的模块
这段代码改一改就可以变成异步爬虫框架。比如把 fetch_html、parse_list、parse_detail 抽成单独的函数，然后通过参数注入不同的解析器。我自己的项目里就是这么干的，现在已经可以用一个 YAML 文件配置多个站点。
class AsyncCrawler:
def __init__(self, concurrency=5, retries=3):
self.semaphore = asyncio.Semaphore(concurrency)
self.retries = retries
async def fetch(self, session, url):
# 通用请求
...
async def run(self, urls):
async with aiohttp.ClientSession() as session:
tasks = [self.fetch(session, url) for url in urls]
return await asyncio.gather(*tasks)
不过对于小任务，上面那个脚本已经够用。你可以把它直接扔到 crontab 里定时运行。
最后：性能对比
我用同一个测试站（100条新闻）分别跑了同步requests版本和异步aiohttp版本。同步版本耗时约 105 秒（平均每条1秒，串行），异步版本并发 5 个，耗时约 22 秒。如果你把并发提高到 10，可以压到 12 秒左右。但并发太高有可能触发反爬，所以实际使用时需要权衡。
这篇文章的所有代码都是可以直接运行的。你只要把 BASE_URL 换成自己的目标站点，再调整一下选择器，就能开始抓取。如果你想抓取更复杂的网站，比如需要登录或者有JS渲染，那就得上 Playwright 了，那是另一个主题。
希望这篇能帮到你。如果你在调试异步爬虫时遇到其他问题，欢迎在评论区留下你遇到的问题，我们一起讨论。
