Python 爬虫实战：asyncio 异步爬虫任务调度

前言

在基于 aiohttp 实现的异步爬虫中，单纯依靠asyncio.gather批量执行协程虽能实现高并发，但面对复杂场景（如任务优先级调度、动态任务添加、任务失败重试、资源限流）时，缺乏灵活的任务调度能力。asyncio 作为 Python 异步编程的核心库，提供了丰富的任务调度原语（如事件循环、任务队列、Future、信号量），可实现精细化的异步任务管控。本文从 asyncio 任务调度核心原理入手，结合实战案例实现优先级任务调度、动态任务分发、分布式任务协同等高级功能，解决异步爬虫在复杂场景下的调度难题。

摘要

本文聚焦 asyncio 异步爬虫的任务调度实战，首先剖析 asyncio 任务调度的核心组件（事件循环、Task、Future、Queue）及调度模式；其次以 B 站热门视频榜单 为爬取目标，依次实现基础任务队列调度、优先级任务调度、动态任务添加与失败重试、基于多进程的分布式任务调度；最后给出任务调度性能调优与监控方案。通过本文，读者可掌握 asyncio 任务调度的全场景开发能力，实现异步爬虫从 “高并发” 到 “智能调度” 的升级，适配企业级复杂爬虫场景。

一、asyncio 任务调度核心原理

1.1 核心组件与作用

1.2 调度模式对比

1.3 任务调度核心流程

1. 初始化事件循环与任务队列；
2. 将初始爬虫任务（如首页 URL）加入队列；
3. 启动消费者协程，从队列中循环取出任务执行；
4. 任务执行过程中生成的新 URL（如分页、详情页）动态加入队列；
5. 监控任务状态，对失败任务进行重试，对完成任务进行结果处理；
6. 队列为空且所有任务执行完成后，关闭事件循环。

二、环境搭建

2.1 基础环境要求

2.2 环境安装

bash

运行

pip install aiohttp==3.8.5 aiofiles==23.1.0 beautifulsoup4==4.12.2 aiomultiprocess==0.9.0

三、asyncio 任务调度实战开发

3.1 实战 1：基础队列调度（动态任务分发）

3.1.1 核心代码实现

python

运行

import asyncio import aiohttp import aiofiles from bs4 import BeautifulSoup from typing import Dict, List import json class BilibiliRankingSpider: """B站热门榜单异步爬虫（队列调度版）""" def __init__(self): # 初始化异步队列（限长 1000，避免内存溢出） self.task_queue = asyncio.Queue(maxsize=1000) # 并发控制信号量（最大并发 30） self.semaphore = asyncio.Semaphore(30) # 已爬取 URL 集合（去重） self.crawled_urls = set() # 存储爬取结果 self.result_list = [] # 请求头 self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "Referer": "https://www.bilibili.com/" } async def fetch(self, url: str) -> Dict: """异步请求并解析数据（带并发控制）""" async with self.semaphore: # 去重校验 if url in self.crawled_urls: return {"url": url, "status": "skipped", "reason": "已爬取"} self.crawled_urls.add(url) try: async with aiohttp.ClientSession(headers=self.headers) as session: async with session.get( url, timeout=aiohttp.ClientTimeout(total=10) ) as response: if response.status != 200: return {"url": url, "status": "failed", "reason": f"状态码 {response.status}"} html = await response.text() soup = BeautifulSoup(html, 'html.parser') # 区分榜单页和视频详情页 if "ranking" in url: # 解析榜单页，提取视频 URL 并加入任务队列 video_list = soup.find_all('a', class_='title') for video in video_list: video_url = video['href'] if not video_url.startswith('http'): video_url = f"https://www.bilibili.com{video_url}" # 避免重复加入队列 if video_url not in self.crawled_urls: await self.task_queue.put(video_url) return {"url": url, "status": "success", "type": "ranking", "count": len(video_list)} else: # 解析视频详情页 title = soup.find('h1', class_='video-title').text.strip() play_count = soup.find('div', class_='view').text.strip() author = soup.find('a', class_='up-name').text.strip() data = { "title": title, "play_count": play_count, "author": author, "url": url } self.result_list.append(data) return {"url": url, "status": "success", "type": "video", "data": data} except Exception as e: return {"url": url, "status": "failed", "reason": str(e)} async def consumer(self): """任务消费者：循环从队列取任务执行""" while True: try: # 非阻塞获取任务，队列为空时抛出 QueueEmpty url = self.task_queue.get_nowait() # 执行爬取任务 result = await self.fetch(url) # 打印任务执行日志 print(f"任务执行结果：{result['url']} - {result['status']}") # 标记任务完成 self.task_queue.task_done() except asyncio.QueueEmpty: # 队列为空时退出循环 break except Exception as e: print(f"消费者执行异常：{e}") async def run(self, start_url: str): """启动爬虫""" # 添加初始任务 await self.task_queue.put(start_url) print(f"初始任务已添加：{start_url}") # 启动多个消费者协程（提升队列消费速度） consumer_tasks = [asyncio.create_task(self.consumer()) for _ in range(5)] # 等待队列所有任务完成 await self.task_queue.join() # 取消所有消费者任务 for task in consumer_tasks: task.cancel() await asyncio.gather(*consumer_tasks, return_exceptions=True) # 异步保存结果 await self.save_result() print(f"爬虫执行完成，共爬取 {len(self.result_list)} 个视频数据") async def save_result(self): """异步保存结果到 JSON 文件""" async with aiofiles.open("bilibili_ranking.json", "w", encoding="utf-8") as f: await f.write(json.dumps(self.result_list, ensure_ascii=False, indent=2)) # 主函数 async def main(): spider = BilibiliRankingSpider() # 初始 URL：B站全站热门榜单 start_url = "https://www.bilibili.com/ranking/all" await spider.run(start_url) if __name__ == "__main__": # Windows 事件循环适配 import platform if platform.system() == "Windows": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(main())

3.1.2 输出结果与原理

JSON 文件输出示例（bilibili_ranking.json）：

json

[ { "title": "2025 年最新 Python 全栈教程 | 从入门到精通", "play_count": "1000.2万", "author": "Python 教程君", "url": "https://www.bilibili.com/video/BV1234567890/" }, { "title": "B站热门游戏盘点 2025", "play_count": "850.5万", "author": "游戏菌", "url": "https://www.bilibili.com/video/BV0987654321/" } ]

控制台输出示例：

plaintext

初始任务已添加：https://www.bilibili.com/ranking/all 任务执行结果：https://www.bilibili.com/ranking/all - success 任务执行结果：https://www.bilibili.com/video/BV1234567890/ - success 任务执行结果：https://www.bilibili.com/video/BV0987654321/ - success ... 爬虫执行完成，共爬取 100 个视频数据

核心原理：

1. 队列调度：asyncio.Queue作为任务载体，初始仅加入榜单页 URL，爬取过程中动态将视频详情页 URL 加入队列；
2. 多消费者模式：启动 5 个消费者协程并行消费队列任务，提升任务处理效率；
3. 队列阻塞控制：task_queue.join()等待所有任务完成，task_done()标记单个任务完成，实现精准的任务生命周期管理；
4. 并发与去重：Semaphore限制并发数，crawled_urls集合实现 URL 去重，避免重复爬取。

3.2 实战 2：优先级任务调度（核心数据优先爬取）

3.2.1 核心代码实现

python

运行

import asyncio import aiohttp import aiofiles from bs4 import BeautifulSoup import json from typing import Tuple, Dict class PriorityQueue(asyncio.Queue): """自定义优先级队列（继承 asyncio.Queue）""" def _put(self, item: Tuple[int, str]) -> None: """重写入队逻辑，按优先级排序（数字越小优先级越高）""" # item 格式：(优先级, URL) # 插入到合适的位置，保持队列升序排列 for i, (priority, _) in enumerate(self._queue): if item[0] < priority: self._queue.insert(i, item) return self._queue.append(item) class BilibiliPrioritySpider: """B站热门榜单异步爬虫（优先级调度版）""" def __init__(self): # 初始化优先级队列 self.task_queue = PriorityQueue(maxsize=1000) self.semaphore = asyncio.Semaphore(30) self.crawled_urls = set() self.result_list = [] self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36", "Referer": "https://www.bilibili.com/" } async def fetch(self, priority: int, url: str) -> Dict: """带优先级的异步爬取""" async with self.semaphore: if url in self.crawled_urls: return {"url": url, "priority": priority, "status": "skipped"} self.crawled_urls.add(url) try: async with aiohttp.ClientSession(headers=self.headers) as session: async with session.get(url, timeout=10) as response: if response.status != 200: return {"url": url, "priority": priority, "status": "failed", "reason": f"status {response.status}"} html = await response.text() soup = BeautifulSoup(html, 'html.parser') if "ranking" in url: # 解析榜单页，为 Top10 视频设置高优先级（1），其余为低优先级（2） video_list = soup.find_all('a', class_='title')[:20] # 取前20个视频 for idx, video in enumerate(video_list): video_url = video['href'] if not video_url.startswith('http'): video_url = f"https://www.bilibili.com{video_url}" # Top10 优先级 1，其余 2 pri = 1 if idx < 10 else 2 await self.task_queue.put((pri, video_url)) return {"url": url, "priority": priority, "status": "success", "type": "ranking"} else: # 解析视频详情页 title = soup.find('h1', class_='video-title').text.strip() play_count = soup.find('div', class_='view').text.strip() author = soup.find('a', class_='up-name').text.strip() data = { "title": title, "play_count": play_count, "author": author, "url": url, "priority": priority } self.result_list.append(data) print(f"优先级 {priority} 任务完成：{title}") return {"url": url, "priority": priority, "status": "success", "type": "video"} except Exception as e: return {"url": url, "priority": priority, "status": "failed", "reason": str(e)} async def consumer(self): """优先级任务消费者""" while True: try: priority, url = self.task_queue.get_nowait() result = await self.fetch(priority, url) self.task_queue.task_done() except asyncio.QueueEmpty: break except Exception as e: print(f"消费者异常：{e}") async def run(self, start_url: str): """启动爬虫（初始任务优先级 0，最高）""" await self.task_queue.put((0, start_url)) print(f"初始优先级任务已添加：{start_url} (优先级 0)") # 启动 5 个消费者 consumer_tasks = [asyncio.create_task(self.consumer()) for _ in range(5)] await self.task_queue.join() # 取消消费者任务 for task in consumer_tasks: task.cancel() await asyncio.gather(*consumer_tasks, return_exceptions=True) # 按优先级排序保存结果 self.result_list.sort(key=lambda x: x['priority']) await self.save_result() print(f"爬虫完成，共爬取 {len(self.result_list)} 条数据（优先级 1：{len([x for x in self.result_list if x['priority']==1])} 条）") async def save_result(self): """保存结果""" async with aiofiles.open("bilibili_priority_ranking.json", "w", encoding="utf-8") as f: await f.write(json.dumps(self.result_list, ensure_ascii=False, indent=2)) async def main(): spider = BilibiliPrioritySpider() await spider.run("https://www.bilibili.com/ranking/all") if __name__ == "__main__": if platform.system() == "Windows": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(main())

3.2.2 输出结果与原理

控制台输出示例：

plaintext

初始优先级任务已添加：https://www.bilibili.com/ranking/all (优先级 0) 优先级 1 任务完成：2025 年最新 Python 全栈教程 | 从入门到精通 优先级 1 任务完成：B站热门游戏盘点 2025 ... 优先级 2 任务完成：2025 年数码产品开箱 ... 爬虫完成，共爬取 20 条数据（优先级 1：10 条）

核心原理：

1. 自定义优先级队列：重写_put方法，按优先级数字升序排列任务（数字越小优先级越高）；
2. 分级优先级：初始榜单页优先级 0（最高），Top10 视频优先级 1，其余视频优先级 2；
3. 优先级消费：消费者从队列头部取任务，确保高优先级任务优先执行，核心数据优先爬取；
4. 结果排序：最终结果按优先级排序存储，便于后续数据处理。

3.3 实战 3：任务失败重试与动态监控

3.3.1 核心代码实现

python

运行

import asyncio import aiohttp import aiofiles from bs4 import BeautifulSoup import json import time from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type class BilibiliRetrySpider: """带失败重试与监控的异步爬虫""" def __init__(self): self.task_queue = asyncio.Queue(maxsize=1000) self.semaphore = asyncio.Semaphore(30) self.crawled_urls = set() self.result_list = [] self.failed_tasks = [] # 失败任务列表 self.start_time = None # 爬虫启动时间 self.headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36" } @retry( stop=stop_after_attempt(3), # 最大重试 3 次 wait=wait_exponential(multiplier=1, min=2, max=10), # 指数退避等待 retry=retry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError)), # 仅重试网络异常 reraise=True ) async def fetch_with_retry(self, url: str) -> Dict: """带重试的爬取方法""" async with self.semaphore: if url in self.crawled_urls: return {"url": url, "status": "skipped"} self.crawled_urls.add(url) async with aiohttp.ClientSession(headers=self.headers) as session: async with session.get(url, timeout=10) as response: if response.status != 200: raise aiohttp.ClientResponseError( response.request_info, response.history, status=response.status ) html = await response.text() soup = BeautifulSoup(html, 'html.parser') if "ranking" in url: video_list = soup.find_all('a', class_='title')[:10] for video in video_list: video_url = video['href'] if not video_url.startswith('http'): video_url = f"https://www.bilibili.com{video_url}" await self.task_queue.put(video_url) return {"url": url, "status": "success", "type": "ranking"} else: title = soup.find('h1', class_='video-title').text.strip() play_count = soup.find('div', class_='view').text.strip() self.result_list.append({"title": title, "play_count": play_count, "url": url}) return {"url": url, "status": "success", "type": "video"} async def fetch(self, url: str) -> Dict: """封装重试逻辑，捕获最终失败的任务""" try: return await self.fetch_with_retry(url) except Exception as e: self.failed_tasks.append({"url": url, "reason": str(e)}) return {"url": url, "status": "failed", "reason": str(e)} async def consumer(self): """任务消费者""" while True: try: url = self.task_queue.get_nowait() result = await self.fetch(url) self.task_queue.task_done() except asyncio.QueueEmpty: break async def monitor(self): """实时监控爬虫状态""" while True: # 每 5 秒输出一次监控数据 await asyncio.sleep(5) if self.task_queue.empty() and all(task.done() for task in asyncio.all_tasks() if task != asyncio.current_task()): break elapsed_time = time.time() - self.start_time queue_size = self.task_queue.qsize() crawled_count = len(self.crawled_urls) success_count = len(self.result_list) failed_count = len(self.failed_tasks) print(f"\n监控数据 [耗时 {elapsed_time:.1f}s]：") print(f"待执行任务数：{queue_size} | 已爬取 URL 数：{crawled_count}") print(f"成功数：{success_count} | 失败数：{failed_count}") async def run(self, start_url: str): """启动爬虫（含监控）""" self.start_time = time.time() await self.task_queue.put(start_url) # 启动监控任务 monitor_task = asyncio.create_task(self.monitor()) # 启动 5 个消费者 consumer_tasks = [asyncio.create_task(self.consumer()) for _ in range(5)] await self.task_queue.join() # 取消任务 for task in consumer_tasks: task.cancel() monitor_task.cancel() await asyncio.gather(*consumer_tasks, monitor_task, return_exceptions=True) # 保存结果与失败任务 await self.save_result() await self.save_failed_tasks() # 输出最终统计 total_time = time.time() - self.start_time print(f"\n===== 爬虫执行完成 =====") print(f"总耗时：{total_time:.1f}s | 成功爬取：{len(self.result_list)} 条") print(f"失败任务：{len(self.failed_tasks)} 条 | 已爬取 URL：{len(self.crawled_urls)} 个") async def save_result(self): """保存成功结果""" async with aiofiles.open("bilibili_retry_success.json", "w", encoding="utf-8") as f: await f.write(json.dumps(self.result_list, ensure_ascii=False, indent=2)) async def save_failed_tasks(self): """保存失败任务（便于后续重爬）""" if self.failed_tasks: async with aiofiles.open("bilibili_retry_failed.json", "w", encoding="utf-8") as f: await f.write(json.dumps(self.failed_tasks, ensure_ascii=False, indent=2)) print(f"失败任务已保存至 bilibili_retry_failed.json") async def main(): spider = BilibiliRetrySpider() await spider.run("https://www.bilibili.com/ranking/all") if __name__ == "__main__": if platform.system() == "Windows": asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(main())

3.3.2 输出结果与原理

监控输出示例：

plaintext

监控数据 [耗时 5.0s]： 待执行任务数：8 | 已爬取 URL 数：3 成功数：2 | 失败数：0 监控数据 [耗时 10.0s]： 待执行任务数：0 | 已爬取 URL 数：10 成功数：9 | 失败数：1 ===== 爬虫执行完成 ===== 总耗时：12.5s | 成功爬取：9 条 失败任务：1 条 | 已爬取 URL：10 个 失败任务已保存至 bilibili_retry_failed.json

核心原理：

1. 失败重试：使用tenacity装饰器实现网络异常重试，指数退避等待避免短时间重复请求；
2. 实时监控：独立的监控协程每隔 5 秒输出队列大小、爬取统计等数据，便于实时掌握爬虫状态；
3. 失败任务留存：将最终失败的任务保存至文件，便于后续分析原因或重爬；
4. 耗时统计：记录爬虫启动 / 结束时间，输出总耗时，便于性能分析。

3.4 实战 4：分布式任务调度（多进程 + 共享队列）

3.4.1 核心代码实现

python

运行

import asyncio import aiohttp import aiofiles from bs4 import BeautifulSoup import json from aiomultiprocess import Pool import multiprocessing from typing import List # 全局配置 MAX_CONCURRENT = 20 MAX_PROCESSES = 4 # 进程数 START_URL = "https://www.bilibili.com/ranking/all" async def fetch_url(url: str, semaphore: asyncio.Semaphore) -> Dict: """单个 URL 爬取函数（进程内复用）""" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Safari/537.36" } async with semaphore: try: async with aiohttp.ClientSession(headers=headers) as session: async with session.get(url, timeout=10) as response: if response.status != 200: return {"url": url, "status": "failed", "reason": f"status {response.status}"} html = await response.text() soup = BeautifulSoup(html, 'html.parser') if "ranking" in url: # 解析榜单页，返回视频 URL 列表 video_list = soup.find_all('a', class_='title')[:20] video_urls = [] for video in video_list: video_url = video['href'] if not video_url.startswith('http'): video_url = f"https://www.bilibili.com{video_url}" video_urls.append(video_url) return {"url": url, "status": "success", "type": "ranking", "data": video_urls} else: # 解析视频详情页 title = soup.find('h1', class_='video-title').text.strip() play_count = soup.find('div', class_='view').text.strip() return {"url": url, "status": "success", "type": "video", "data": {"title": title, "play_count": play_count}} except Exception as e: return {"url": url, "status": "failed", "reason": str(e)} async def process_task(urls: List[str]) -> List[Dict]: """单个进程的任务处理函数""" semaphore = asyncio.Semaphore(MAX_CONCURRENT) tasks = [fetch_url(url, semaphore) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) return results def split_urls(urls: List[str], num_processes: int) -> List[List[str]]: """将 URL 列表均分至多个进程""" avg = len(urls) // num_processes remainder = len(urls) % num_processes chunks = [] start = 0 for i in range(num_processes): end = start + avg + (1 if i < remainder else 0) chunks.append(urls[start:end]) start = end return chunks async def main(): """分布式任务调度主函数""" # 第一步：爬取榜单页，获取所有视频 URL print("第一步：爬取榜单页，获取视频 URL 列表...") ranking_result = await fetch_url(START_URL, asyncio.Semaphore(1)) if ranking_result['status'] != 'success': print(f"榜单页爬取失败：{ranking_result['reason']}") return video_urls = ranking_result['data'] print(f"获取到 {len(video_urls)} 个视频 URL") # 第二步：拆分 URL 列表至多个进程 url_chunks = split_urls(video_urls, MAX_PROCESSES) print(f"URL 列表已拆分至 {MAX_PROCESSES} 个进程，每个进程任务数：{[len(chunk) for chunk in url_chunks]}") # 第三步：多进程异步爬取 print("第二步：多进程异步爬取视频数据...") start_time = asyncio.get_event_loop().time() async with Pool(processes=MAX_PROCESSES) as pool: # 每个进程执行 process_task 函数 results = await pool.map(process_task, url_chunks) # 第四步：合并结果 print("第三步：合并爬取结果...") success_data = [] failed_data = [] for process_result in results: for res in process_result: if isinstance(res, Exception): failed_data.append({"url": "unknown", "reason": str(res)}) elif res['status'] == 'success' and res['type'] == 'video': success_data.append(res['data']) elif res['status'] == 'failed': failed_data.append(res) # 第五步：保存结果 async with aiofiles.open("bilibili_distributed_success.json", "w", encoding="utf-8") as f: await f.write(json.dumps(success_data, ensure_ascii=False, indent=2)) if failed_data: async with aiofiles.open("bilibili_distributed_failed.json", "w", encoding="utf-8") as f: await f.write(json.dumps(failed_data, ensure_ascii=False, indent=2)) # 输出统计 total_time = asyncio.get_event_loop().time() - start_time print(f"\n===== 分布式爬取完成 =====") print(f"总耗时：{total_time:.1f}s | 进程数：{MAX_PROCESSES} | 单进程最大并发：{MAX_CONCURRENT}") print(f"成功爬取：{len(success_data)} 条 | 失败：{len(failed_data)} 条") if __name__ == "__main__": # 多进程需在 Windows 下添加保护 if platform.system() == "Windows": multiprocessing.freeze_support() asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy()) asyncio.run(main())

3.4.2 输出结果与原理

控制台输出示例：

plaintext

第一步：爬取榜单页，获取视频 URL 列表... 获取到 20 个视频 URL URL 列表已拆分至 4 个进程，每个进程任务数：[5,5,5,5] 第二步：多进程异步爬取视频数据... 第三步：合并爬取结果... ===== 分布式爬取完成 ===== 总耗时：4.2s | 进程数：4 | 单进程最大并发：20 成功爬取：19 条 | 失败：1 条

核心原理：

1. 多进程调度：使用aiomultiprocess.Pool创建多进程池，每个进程独立运行事件循环，利用多核 CPU 提升爬取效率；
2. 任务拆分：将 URL 列表均分给多个进程，避免单个进程任务过多；
3. 进程内并发：每个进程内通过Semaphore限制并发数，平衡单进程与整体并发；
4. 结果合并：收集所有进程的爬取结果，统一处理成功 / 失败数据并保存。

四、任务调度性能调优

4.1 关键参数调优表

4.2 性能对比测试

调优结论：分布式调度耗时仅为单进程批量调度的 50%，CPU 利用率提升 2 倍，是大规模爬取的最优选择。

五、常见问题与解决方案

六、总结

本文系统讲解了基于 asyncio 的异步爬虫任务调度开发，从基础队列调度、优先级调度，到失败重试、分布式调度，覆盖了异步爬虫在不同场景下的调度需求。通过自定义优先级队列实现核心数据优先爬取，结合 tenacity 实现失败任务重试，利用 aiomultiprocess 实现多进程分布式调度，大幅提升了异步爬虫的灵活性与可扩展性。

asyncio 任务调度的核心价值在于实现 “智能并发”—— 不仅能提升爬取效率，还能通过优先级、限流、监控等机制，适配企业级复杂爬虫场景（如核心数据优先、超大规模 URL 爬取、实时监控）。在实际开发中，可进一步扩展：结合 Redis 实现跨机器分布式队列、集成 Prometheus 实现可视化监控、对接消息队列实现任务异步分发等。掌握 asyncio 任务调度，可实现异步爬虫从 “高并发” 到 “可管控、可监控、可扩展” 的全面升级。