当前位置：首页 > news >正文

asyncio+queue实现生产者消费者爬虫模型

news 2026/3/26 20:00:19

在网络爬虫开发中，生产者 - 消费者模型是经典且高效的架构模式。它将 “任务生产（URL 采集）” 和 “任务消费（页面爬取）” 解耦，能有效控制并发、避免资源浪费。而 Python 的asyncio（异步 I/O）结合asyncio.Queue，可以打造出高性能的异步爬虫，相比传统多线程 / 多进程爬虫，异步模型在 IO 密集型的爬虫场景下效率提升显著。

一、核心原理

1. 生产者 - 消费者模型

生产者：负责生成待爬取的 URL，将其放入队列中，是爬虫的 “任务源”；
消费者：从队列中取出 URL，执行页面请求、数据解析等操作，是爬虫的 “执行端”；
队列（asyncio.Queue）：作为生产者和消费者之间的缓冲，解耦两者的执行节奏，控制并发数量，避免瞬间请求量过大导致的被封 IP 或服务器压力过高。

2. asyncio 的优势

asyncio是 Python 内置的异步编程框架，基于事件循环实现，无需创建大量线程 / 进程，仅通过单线程的协程切换就能处理大量 IO 操作（如网络请求），资源开销远低于多线程，适合爬虫这类高 IO 场景。

二、完整实现代码

以下是一个可直接运行的异步爬虫示例，以爬取某测试站点的页面标题为例，完整实现生产者 - 消费者模型：

python

运行

import asyncio import aiohttp from aiohttp import ClientTimeout from typing import List # 全局配置 MAX_CONCURRENT = 5 # 最大并发消费者数量 QUEUE_MAXSIZE = 10 # 队列最大容量，控制缓冲大小 BASE_URL = "https://httpbin.org/get?page={}" # 测试URL PAGE_RANGE = range(1, 21) # 待爬取的页面范围（1-20页） class AsyncCrawler: def __init__(self): # 初始化异步队列，设置最大容量 self.queue = asyncio.Queue(maxsize=QUEUE_MAXSIZE) # 存储爬取结果 self.results = [] async def producer(self, urls: List[str]): """生产者：将待爬取的URL放入队列""" for url in urls: await self.queue.put(url) print(f"生产者：已放入URL -> {url}") # 放入结束标记（数量等于消费者数量），通知消费者退出 for _ in range(MAX_CONCURRENT): await self.queue.put(None) async def consumer(self, session: aiohttp.ClientSession, consumer_id: int): """消费者：从队列取出URL并爬取""" while True: # 从队列获取URL（异步阻塞，直到有数据） url = await self.queue.get() # 检测结束标记 if url is None: print(f"消费者{consumer_id}：收到结束标记，退出") self.queue.task_done() break try: # 异步请求页面 async with session.get(url, timeout=ClientTimeout(total=10)) as response: if response.status == 200: # 解析响应数据（此处仅示例，可替换为实际解析逻辑） data = await response.json() page = data["args"]["page"] self.results.append(f"页面{page}爬取成功") print(f"消费者{consumer_id}：成功爬取 -> {url}") else: print(f"消费者{consumer_id}：爬取失败，状态码 -> {response.status}") except Exception as e: print(f"消费者{consumer_id}：爬取异常 -> {url}，错误：{str(e)}") finally: # 标记任务完成（用于队列的join()方法） self.queue.task_done() async def run(self): """启动爬虫主流程""" # 1. 生成待爬取的URL列表 urls = [BASE_URL.format(page) for page in PAGE_RANGE] # 2. 创建异步HTTP会话（复用连接，提升性能） async with aiohttp.ClientSession() as session: # 3. 创建并启动生产者任务 producer_task = asyncio.create_task(self.producer(urls)) # 4. 创建并启动多个消费者任务 consumer_tasks = [] for i in range(MAX_CONCURRENT): task = asyncio.create_task(self.consumer(session, i+1)) consumer_tasks.append(task) # 5. 等待生产者任务完成 await producer_task # 6. 等待队列中所有任务处理完成 await self.queue.join() # 7. 等待所有消费者任务退出 await asyncio.gather(*consumer_tasks) # 8. 输出爬取结果 print("\n===== 爬取完成 =====") for res in self.results: print(res) print(f"总计成功爬取：{len(self.results)} 条数据") if __name__ == "__main__": # 运行异步爬虫 crawler = AsyncCrawler() asyncio.run(crawler.run())