异步爬虫 aiohttp：百万级数据高效采集方案-洪萨配资

前言

在当今数据驱动决策的时代，无论是企业数据分析、商业情报监测、行业研究还是个人项目开发，对数据量级的要求都在不断提升。从过去的万级、十万级数据采集，逐步过渡到如今百万级甚至千万级数据的常态化需求。传统的同步单线程爬虫，甚至多线程、多进程爬虫，在面对大规模 URL 批量采集时，往往暴露出效率低、耗时长、资源占用高、稳定性差等问题。

而基于asyncio+aiohttp实现的异步爬虫，凭借非阻塞 IO、超高并发、轻量级协程等特性，成为当前 Python 生态中实现百万级数据高效采集的主流方案。它可以在单线程内同时发起数百甚至上千个网络请求，极大提升爬取速度，同时降低服务器与本地资源开销。

本文将从异步原理、环境搭建、基础写法、高并发控制、百万级批量采集、异常处理、反爬策略、数据存储、性能调优、实战案例等方面，进行系统性、深度、可落地的讲解，帮助你从零掌握 aiohttp 异步爬虫，并真正具备百万级数据采集的工程化能力。

一、为什么要使用异步爬虫

1.1 传统爬虫的瓶颈

在没有接触异步之前，大多数开发者使用的是requests+ 循环的同步爬虫。这种方式逻辑简单、易于上手，但存在致命缺陷：

阻塞 IO发起一次 HTTP 请求后，程序必须等待服务器返回结果才能继续下一个请求。在等待的这段时间内，CPU 完全闲置，造成巨大的时间浪费。
并发能力极低即使使用多线程，受限于 GIL（全局解释器锁）以及线程切换开销，真实并发量依旧有限，且容易出现线程崩溃、内存暴涨等问题。
百万级数据几乎不可用若使用同步爬虫爬取 100 万条数据，按照每个请求 1 秒计算，理论耗时超过 270 小时，现实中完全不可接受。
资源利用率低下网络 IO 等待时 CPU 空闲，CPU 运行时网络又闲置，整体资源利用率极低。

1.2 异步爬虫的优势

异步爬虫基于协程（coroutine）实现，在单线程内实现 “伪并行”，其优势非常明显：

非阻塞网络请求：发起请求后立刻切换执行其他任务，不等待响应。
超高并发：单机轻松实现数百并发，合理配置可达上千并发。
低资源消耗：协程极轻量，无线程 / 进程切换开销，内存占用极低。
百万级数据可快速落地：同等条件下，速度比同步爬虫提升30～100 倍。
适合大规模接口爬取、列表页翻页、全站抓取。

可以说，不会异步爬虫，就无法胜任现代大规模数据采集工作。

1.3 aiohttp 框架简介

aiohttp是基于 asyncio 的异步 HTTP 框架，同时支持客户端与服务端。在爬虫领域，我们主要使用它的客户端功能，用于异步发送 GET/POST 请求。

它的核心特点：

完全异步非阻塞
支持长连接、连接池复用
支持超时、代理、请求头、Cookie 等完整配置
生态完善，可配合aiomysql、motor、aiofiles实现异步存储
社区活跃，文档完善，是 Python 异步爬虫事实标准

二、异步编程核心基础

在正式写爬虫前，必须理解 asyncio 的几个核心概念，否则代码只会复制粘贴，无法真正掌握。

2.1 协程（Coroutine）

协程是一种轻量级的执行单元，可以在函数内部暂停和恢复执行。通过async def定义协程函数，通过await挂起当前协程，让事件循环去执行其他任务。

2.2 事件循环（Event Loop）

事件循环是异步程序的 “大脑”，负责调度所有协程。它不断检测哪些协程可以执行、哪些协程在等待 IO，并在合适的时机切换任务。

2.3 Task（任务）

Task 是对协程的进一步封装，用于将协程注册到事件循环中。使用asyncio.create_task()可以创建任务并实现并发。

2.4 await 关键字

await后面只能跟可等待对象（协程、Task、Future）。当程序执行到 await 时，当前协程会被挂起，事件循环可以去执行其他任务。

2.5 信号量（Semaphore）

信号量用于控制最大并发数，防止并发过高导致目标网站封禁 IP，或导致本地报错 “Too many open files”。

这是百万级爬虫稳定运行的关键。

三、环境搭建与基础依赖安装

3.1 Python 版本要求

建议使用Python 3.7+，最好是 3.8 及以上，对 asyncio 支持更完善。

3.2 核心依赖安装

bash

运行

pip install aiohttp pip install asyncio pip install beautifulsoup4 pip install lxml pip install aiofiles pip install aiomysql pip install motor pip install fake-useragent pip install python-dotenv

简单说明：

aiohttp：异步请求核心
asyncio：异步调度
beautifulsoup4 / lxml：HTML 解析
aiofiles：异步写入文件
aiomysql：异步写入 MySQL
motor：异步写入 MongoDB
fake-useragent：随机 UA 防反爬

四、aiohttp 异步爬虫基础写法

4.1 最简单的异步请求示例

python

运行

import asyncio import aiohttp async def fetch(url): async with aiohttp.ClientSession() as session: async with session.get(url) as resp: print("状态码:", resp.status) html = await resp.text() return html async def main(): url = "https://www.baidu.com" result = await fetch(url) print(result[:500]) if __name__ == '__main__': asyncio.run(main())

要点：

ClientSession是 aiohttp 推荐的会话对象，必须复用，不能每次请求新建。
async with用于自动管理上下文，避免资源泄漏。
await resp.text()等待响应内容返回。

4.2 多 URL 并发爬取

python

运行

import asyncio import aiohttp async def fetch(session, url): try: async with session.get(url, timeout=10) as resp: if resp.status == 200: return await resp.text() except Exception as e: print(f"请求失败 {url}: {e}") return None async def main(): urls = [ "https://www.baidu.com", "https://www.qq.com", "https://www.163.com" ] async with aiohttp.ClientSession() as session: tasks = [asyncio.create_task(fetch(session, url)) for url in urls] results = await asyncio.gather(*tasks) for res in results: print(res[:100] if res else "None") if __name__ == '__main__': asyncio.run(main())

asyncio.gather(*tasks)可以批量运行多个任务，并统一收集结果。

五、高并发控制：Semaphore 信号量

不加限制的并发会直接导致：

目标网站触发反爬，直接封 IP
本地出现 “Too many open files”
大量超时、连接重置错误

因此必须使用信号量控制并发。

5.1 信号量使用示例

python

运行

import asyncio import aiohttp async def fetch(session, sem, url): async with sem: # 限制并发 try: async with session.get(url, timeout=15) as resp: return await resp.text() except Exception as e: return None async def main(): urls = [f"https://www.baidu.com/s?wd={i}" for i in range(1000)] sem = asyncio.Semaphore(50) # 最大并发 50 async with aiohttp.ClientSession() as session: tasks = [fetch(session, sem, url) for url in urls] results = await asyncio.gather(*tasks) if __name__ == '__main__': asyncio.run(main())

并发数建议：

普通小网站：10～30
中型网站：30～80
大型高防网站：5～10
企业级采集：根据代理池质量动态调整

六、百万级 URL 批量采集方案设计

百万级数据不能简单把所有 URL 一次性塞进内存，必须采用工程化方案。

6.1 百万级 URL 来源

从数据库读取
从 txt/csv 文件读取
规则生成（如自增 ID）
从列表页分页提取

6.2 分批爬取策略

一次性加载 100 万 URL 会占用大量内存，正确做法是分批处理：

每批 1000 / 2000 / 5000 条
一批爬完再爬下一批
结合断点续爬

示例思路：

python

运行

BATCH_SIZE = 2000 for i in range(0, len(url_list), BATCH_SIZE): batch = url_list[i:i+BATCH_SIZE] await crawl_batch(batch)

6.3 生产 - 消费模型

更稳定的方式是使用异步队列：

生产者不断往队列放入 URL
消费者并发从队列取出并爬取
适合长时间运行、断点续爬

七、HTML 解析与数据提取

爬取到网页后，需要解析数据，常用方案：

BeautifulSoup
lxml
正则表达式
pyquery（类 jQuery 语法）

基础解析示例：

python

运行

from bs4 import BeautifulSoup async def parse(html): soup = BeautifulSoup(html, 'lxml') title = soup.title.string if soup.title else "" contents = soup.find_all('div', class_='content') data = [c.get_text(strip=True) for c in contents] return { "title": title, "data": data }

在异步爬虫中，解析本身是 CPU 密集型，不会阻塞事件循环，可直接同步写。

八、异步数据存储方案

异步爬虫速度极快，如果使用同步存储（如 pymysql、pymongo）会成为瓶颈，导致爬取速度被存储拖慢。

因此必须使用异步存储。

8.1 异步写入 MySQL（aiomysql）

python

运行

import aiomysql async def save_mysql(pool, item): async with pool.acquire() as conn: async with conn.cursor() as cur: sql = """ INSERT INTO data(title, content, url) VALUES(%s, %s, %s) ON DUPLICATE KEY UPDATE content=VALUES(content) """ await cur.execute(sql, (item['title'], item['content'], item['url'])) await conn.commit() async def create_pool(): pool = await aiomysql.create_pool( host="localhost", user="root", password="123456", db="spider", charset="utf8mb4", autocommit=False ) return pool

8.2 异步写入 MongoDB（motor）

python

运行

from motor.motor_asyncio import AsyncIOMotorClient async def save_mongo(item): client = AsyncIOMotorClient("mongodb://localhost:27017") db = client.spider_db await db.data_collection.insert_one(item)

8.3 异步写入文件（aiofiles）

python

运行

import aiofiles async def save_file(item): async with aiofiles.open("result.txt", "a", encoding="utf-8") as f: line = f"{item['title']}\t{item['url']}\n" await f.write(line)

九、异常处理与重试机制

百万级爬取中，网络波动、超时、5xx 错误、DNS 失败非常常见，必须做异常捕获与自动重试。

9.1 通用重试装饰器

python

运行

def retry(max_retry=3): def decorator(func): async def wrapper(*args, **kwargs): for i in range(max_retry): try: return await func(*args, **kwargs) except Exception as e: if i == max_retry - 1: print(f"最终失败: {e}") await asyncio.sleep(1) return None return wrapper return decorator @retry(max_retry=3) async def fetch(session, sem, url): async with sem: async with session.get(url, timeout=15) as resp: return await resp.text()

9.2 常见异常类型

asyncio.TimeoutError：超时
aiohttp.ClientError：客户端错误
ConnectionResetError：连接被重置
TooManyRedirects：重定向次数过多
反爬 403、404、500 等状态码

对不同状态码做不同处理：

404：标记失效，不再重试
403：IP 被限制，降低并发或换代理
500/502：服务器错误，延迟重试

十、反爬应对策略

异步爬虫并发高，更容易触发反爬，必须做好防护。

10.1 请求头伪装

User-Agent 随机切换
Referer 合理设置
Accept-Language、Accept-Encoding 模拟浏览器

python

运行

from fake_useragent import UserAgent ua = UserAgent() headers = { "User-Agent": ua.random, "Referer": "https://www.baidu.com", "Accept-Language": "zh-CN,zh;q=0.9", }

10.2 随机延迟

python

运行

await asyncio.sleep(random.uniform(0.2, 1.2))

10.3 代理池使用

aiohttp 支持 http/https/socks 代理：

python

运行

proxy = "http://ip:port" async with session.get(url, proxy=proxy) as resp: ...

10.4 限流与风控

并发不要过高
同一 IP 不要长时间高频访问
关键接口增加间隔时间

十一、百万级爬虫完整工程化实战

下面给出可直接运行、面向百万级采集的完整结构代码，包含：

并发控制
异常重试
随机 UA
分批爬取
异步 MySQL 存储
日志统计
优雅退出

python

运行

import asyncio import aiohttp import aiomysql from bs4 import BeautifulSoup from fake_useragent import UserAgent import random import time ua = UserAgent() class MillionSpider: def __init__(self): self.max_concurrent = 50 self.batch_size = 2000 self.sem = asyncio.Semaphore(self.max_concurrent) self.db_config = { "host": "localhost", "user": "root", "password": "123456", "db": "spider", "charset": "utf8mb4" } self.success = 0 self.failed = 0 def get_headers(self): return { "User-Agent": ua.random, "Referer": "https://www.baidu.com", } async def save(self, pool, item): try: async with pool.acquire() as conn: async with conn.cursor() as cur: sql = """INSERT IGNORE INTO data(title, url, content) VALUES(%s, %s, %s)""" await cur.execute(sql, (item["title"], item["url"], item["content"])) await conn.commit() self.success += 1 except Exception as e: self.failed += 1 async def parse(self, html, url): soup = BeautifulSoup(html, "lxml") title = soup.title.string if soup.title else "" content = soup.find("div", class_="content") content_text = content.get_text(strip=True) if content else "" return { "title": title, "url": url, "content": content_text } async def fetch(self, session, pool, url): try: async with self.sem: await asyncio.sleep(random.uniform(0.1, 0.8)) async with session.get( url, headers=self.get_headers(), timeout=20 ) as resp: if resp.status != 200: self.failed +=1 return html = await resp.text() item = await self.parse(html, url) await self.save(pool, item) except Exception as e: self.failed +=1 async def crawl_batch(self, pool, session, batch): tasks = [self.fetch(session, pool, url) for url in batch] await asyncio.gather(*tasks) async def run(self, url_list): pool = await aiomysql.create_pool(**self.db_config) async with aiohttp.ClientSession() as session: total = len(url_list) start_time = time.time() print(f"开始爬取，总数量：{total}") for i in range(0, total, self.batch_size): batch = url_list[i:i+self.batch_size] await self.crawl_batch(pool, session, batch) speed = (i + len(batch)) / (time.time() - start_time) print(f"已完成 {i+len(batch)}/{total} 速度: {speed:.1f}条/秒") pool.close() await pool.wait_closed() end_time = time.time() print("="*50) print(f"总耗时: {end_time - start_time:.2f}s") print(f"成功: {self.success} 失败: {self.failed}") if __name__ == '__main__': # 模拟百万 URL url_list = [f"https://example.com/item/{i}" for i in range(1000000)] spider = MillionSpider() asyncio.run(spider.run(url_list))

该结构可以直接用于真实项目，只需替换解析规则与数据库配置。

十二、性能测试与对比

12.1 测试环境

CPU：4 核
网络：家庭宽带
目标：静态页面 + 简单解析
数据量：10000 条

12.2 结果对比

表格

方案	耗时	并发	稳定性	内存占用
requests 同步	约 1200s	1	一般	低
多线程 requests	约 180s	32	较差	中
aiohttp 异步	约 25s	50	优秀	极低

效率提升接近 50 倍。

若是百万级数据，差距会更加恐怖。

十三、常见问题与避坑指南

13.1 Too many open files

原因：并发过高，超出系统文件句柄限制
解决：降低信号量、修改系统 ulimit

13.2 大量超时

网络波动
目标网站限流
代理质量差解决：增加超时时间、降低并发、使用代理池。

13.3 数据重复

解决：

数据库建唯一索引
使用INSERT IGNORE
布隆过滤器去重

13.4 内存持续上涨

一次性加载太多 URL
结果未及时释放
未复用 ClientSession解决：分批爬取、及时清理变量、复用 session。

13.5 事件循环异常崩溃

不要在异步函数中调用同步阻塞函数。

十四、扩展方向与进阶

掌握基础异步爬虫后，可继续进阶：

aiohttp + 代理池：实现高匿稳定爬取
aiohttp + 布隆过滤器：海量 URL 去重
aiohttp + Redis 队列：分布式爬虫
aiohttp + 定时任务：增量采集
aiohttp + AI 验证码识别：突破复杂反爬
aiohttp + 日志系统 + 监控：企业级爬虫服务

十五、总结

本文从理论到实践，完整讲解了aiohttp 异步爬虫实现百万级数据高效采集的全流程，包括异步原理、环境配置、基础写法、高并发控制、数据解析、异步存储、异常重试、反爬策略、工程化代码、性能优化等内容。

aiohttp 凭借其高效、轻量、稳定的特性，已经成为 Python 大规模爬虫的首选方案。只要掌握本文内容，你完全可以独立完成：

百万级商品数据采集
舆情与文章批量爬取
行业数据全站采集
接口数据高并发拉取
企业级数据中台爬虫服务

异步爬虫不仅是一项技术，更是现代数据工程师必备的核心能力。在数据量爆炸式增长的今天，高效、稳定、低成本地获取数据，将成为你在工作与项目中的巨大优势。

前言

一、为什么要使用异步爬虫

1.1 传统爬虫的瓶颈

1.2 异步爬虫的优势

1.3 aiohttp 框架简介

二、异步编程核心基础

2.1 协程（Coroutine）

2.2 事件循环（Event Loop）

2.3 Task（任务）

2.4 await 关键字

2.5 信号量（Semaphore）

三、环境搭建与基础依赖安装

3.1 Python 版本要求

3.2 核心依赖安装

四、aiohttp 异步爬虫基础写法

4.1 最简单的异步请求示例

4.2 多 URL 并发爬取

五、高并发控制：Semaphore 信号量

5.1 信号量使用示例

六、百万级 URL 批量采集方案设计

6.1 百万级 URL 来源

6.2 分批爬取策略

6.3 生产 - 消费模型

七、HTML 解析与数据提取

八、异步数据存储方案

8.1 异步写入 MySQL（aiomysql）

8.2 异步写入 MongoDB（motor）

8.3 异步写入文件（aiofiles）

九、异常处理与重试机制

9.1 通用重试装饰器

9.2 常见异常类型

十、反爬应对策略

10.1 请求头伪装

10.2 随机延迟

10.3 代理池使用

10.4 限流与风控

十一、百万级爬虫完整工程化实战

十二、性能测试与对比

12.1 测试环境

12.2 结果对比

十三、常见问题与避坑指南

13.1 Too many open files

13.2 大量超时

13.3 数据重复

13.4 内存持续上涨

13.5 事件循环异常崩溃

十四、扩展方向与进阶

十五、总结

暗黑3终极自动化指南：D3KeyHelper图形化宏工具完整配置教程

别再手动穷举了！用Matlab的BPSO算法搞定背包问题，附完整代码和避坑指南

Gemini 3 Flash：效率革命，如何重塑AI应用的“不可能三角”

C++ 多态与虚函数入门：从概念到规则

手把手教你用MATLAB解析TI毫米波雷达原始bin文件（附完整代码与数据）

HBuilderX里uni-app项目老报caniuse-lite过期？别慌，手把手教你两种修复方法（含手动更新npm包）