核心内容摘要
AI驱动的正向+反向测试用例自动生成
在数据爬取场景中传统同步爬虫受限于 IO 等待如网络请求、数据库写入效率往往难以满足大规模数据采集需求。
而异步编程能最大化利用 CPU 资源结合 MongoDB 的异步驱动则可实现 “爬取 - 存储” 全流程异步化大幅提升爬虫整体性能。
本文将详细讲解如何基于 Python 的 aiohttp 实现异步爬虫并结合 pymongo 的异步特性Motor完成数据的高效存储。
核心技术栈说明在开始编码前先明确核心依赖库的作用aiohttpPython 异步 HTTP 客户端 / 服务器框架用于发起异步网络请求替代同步的 requests 库MotorMongoDB 官方推荐的异步驱动是 pymongo 的异步版本支持异步 IO 操作 MongoDBasyncioPython 内置的异步编程框架用于管理异步任务、事件循环pymongoMongoDB 的同步驱动本文仅作为对比参考核心使用 Motor。
环境准备
安装依赖bash运行pip install aiohttp motor python-dotenv
前提条件已安装并启动 MongoDB 服务本地或远程了解基本的异步编程概念如协程、async/await目标网站允许爬虫访问遵守 robots 协议避免法律风险。
异步爬虫 MongoDB 异步存储实现
核心思路初始化 Motor 客户端建立与 MongoDB 的异步连接定义异步爬虫函数通过 aiohttp 发起异步请求解析目标数据定义异步存储函数将解析后的数据异步写入 MongoDB利用 asyncio 创建任务列表批量执行异步爬取 存储任务统一管理事件循环确保所有异步任务完成后关闭连接。
完整代码实现python运行import asyncio import aiohttp from motor.motor_asyncio import AsyncIOMotorClient from dotenv import load_dotenv import os from typing import Dict, List # 加载环境变量建议将敏感信息放在.env文件中 load_dotenv() # MongoDB配置 MONGO_URI os.getenv(MONGO_URI, mongodb://localhost:27017/) MONGO_DB_NAME os.getenv(MONGO_DB_NAME, spider_db) MONGO_COLLECTION_NAME os.getenv(MONGO_COLLECTION_NAME, async_spider_data) # 爬虫配置 TARGET_URLS [ https://jsonplaceholder.typicode.com/posts/1, https://jsonplaceholder.typicode.com/posts/2, https://jsonplaceholder.typicode.com/posts/3, https://jsonplaceholder.typicode.com/posts/4, https://jsonplaceholder.typicode.com/posts/5, ] HEADERS { User-Agent: Mozilla/
0 (Windows NT
1
0; Win64; x
AppleWebKit/
5
36 (KHTML, like Gecko) Chrome/
120.
0.
0 Safari/
5
36 } # 初始化MongoDB异步客户端 async def init_mongo() - AsyncIOMotorClient: 初始化MongoDB异步连接 try: client AsyncIOMotorClient(MONGO_URI) # 测试连接 await client.admin.command(ping) print(MongoDB异步连接成功) return client except Exception as e: print(fMongoDB连接失败{e}) raise # 异步爬取单条数据 async def crawl_single_url(session: aiohttp.ClientSession, url: str) - Dict | None: 异步爬取单个URL的内容 try: async with session.get(url, headersHEADERS, timeoutaiohttp.ClientTimeout(
) as response: if response.status 200: data await response.json() print(f成功爬取{url}) return data else: print(f爬取失败状态码{response.status}URL{url}) return None except Exception as e: print(f爬取异常{e}URL{url}) return None # 异步存储数据到MongoDB async def save_to_mongo(db, data: Dict) - None: 异步将数据写入MongoDB if not data: return try: collection db[MONGO_COLLECTION_NAME] # 异步插入单条数据 result await collection.insert_one(data) print(f数据存储成功ID{result.inserted_id}) except Exception as e: print(f数据存储失败{e}) # 核心异步任务爬取存储 async def crawl_and_save(session: aiohttp.ClientSession, db, url: str) - None: 单个URL的爬取存储一体化异步任务 data await crawl_single_url(session, url) await save_to_mongo(db, data) # 主函数批量执行异步任务 async def main(): # 初始化MongoDB连接 client await init_mongo() db client[MONGO_DB_NAME] # 创建aiohttp会话复用连接池提升效率 async with aiohttp.ClientSession() as session: # 创建异步任务列表 tasks [] for url in TARGET_URLS: task asyncio.create_task(crawl_and_save(session, db, url)) tasks.append(task) # 等待所有任务完成 await asyncio.gather(*tasks) # 关闭MongoDB连接 client.close() print(所有爬取和存储任务完成) if __name__ __main__: # 适配Python
7的事件循环运行方式 if sys.version_info (3,
: asyncio.run(main()) else: loop asyncio.get_event_loop() loop.run_until_complete(main())
代码关键解析1MongoDB 异步初始化init_mongo函数通过AsyncIOMotorClient创建异步客户端替代 pymongo 的同步MongoClient并通过await client.admin.command(ping)测试连接确保连接成功后再执行后续操作。
2异步 HTTP 请求使用aiohttp.ClientSession创建会话复用连接池通过async with session.get()发起异步请求await response.json()异步解析响应数据避免同步请求的 IO 阻塞。
3异步数据存储save_to_mongo函数中await collection.insert_one(data)是核心异步操作替代 pymongo 同步的insert_one无需等待数据库写入完成即可继续处理下一个爬取任务。
4批量任务管理通过asyncio.create_task创建多个异步任务再用asyncio.gather批量执行实现多 URL 并行爬取和存储最大化利用资源。
性能对比与优化建议
同步 vs 异步性能差异以爬取 100 个 URL 并写入 MongoDB 为例同步爬虫requestspymongo约需
秒受网络延迟、数据库写入等待影响异步爬虫aiohttpMotor约需
秒并行处理 IO 操作无等待时间。
优化方向1控制并发数避免无限制并发导致目标网站封禁 IP 或 MongoDB 压力过大可使用asyncio.Semaphore限制并发数python运行# 在main函数中添加信号量限制最大并发为5 semaphore asyncio.Semaphore(
# 修改crawl_and_save函数增加信号量控制 async def crawl_and_save(session: aiohttp.ClientSession, db, url: str) - None: async with semaphore: # 限制并发 data await crawl_single_url(session, url) await save_to_mongo(db, data)2数据批量写入若爬取数据量极大可将多条数据缓存后批量插入减少数据库交互次数python运行async def save_batch_to_mongo(db, data_list: List[Dict]) - None: 批量异步插入数据 if not data_list: return try: collection db[MONGO_COLLECTION_NAME] result await collection.insert_many(data_list) print(f批量存储成功插入ID数量{len(result.inserted_ids)}) except Exception as e: print(f批量存储失败{e})3异常重试机制对失败的爬取任务添加重试逻辑提升稳定性python运行async def crawl_single_url(session: aiohttp.ClientSession, url: str, retry
- Dict | None: 带重试机制的异步爬取 for i in range(retry): try: async with session.get(url, headersHEADERS, timeoutaiohttp.ClientTimeout(
) as response: if response.status 200: return await response.json() else: print(f第{i1}次重试失败状态码{response.status}) await asyncio.sleep(
# 重试前休眠1秒 except Exception as e: print(f第{i1}次重试异常{e}) await asyncio.sleep(
return None
五、
注意事项遵守网站规则异步爬虫效率高需控制爬取频率避免对目标网站造成压力必要时添加延迟MongoDB 索引针对查询频繁的字段如 url、id创建索引提升后续数据查询效率资源释放确保异步任务完成后关闭 MongoDB 客户端和 aiohttp 会话避免资源泄漏数据去重可通过 MongoDB 的唯一索引如create_index(id, uniqueTrue)避免重复存储数据。
总结异步爬虫结合 MongoDB 异步驱动Motor可实现 “爬取 - 存储” 全流程异步化相比同步方案能提升
倍效率核心依赖为 aiohttp异步请求和 MotorMongoDB 异步驱动需通过async/await管理异步任务实际应用中需控制并发数、添加重试机制、批量写入数据平衡效率与稳定性同时遵守网站爬取规则。
通过本文的实践你可以快速搭建高效的异步爬虫系统满足大规模数据采集与存储的需求同时兼顾代码的可维护性和扩展性。