前段时间我一直在琢磨协程爬虫的实际落地场景正好朋友让我帮他把王者荣耀的全英雄皮肤图片整理成一份干净的数据集。这活儿表面是图片爬取其实真正考的是异步IO的调度和容错设计。我先用最朴素的 requests 循环写了一遍结果一张图一眨眼、一千张图就得喝一壶茶。换成 asyncio aiohttp 之后同样的任务量压缩到原来的零头过程里还暴露出不少容易忽略的细节。这篇就把我的方案、代码和踩坑记录一次说完想用 Python 协程做批量采集的读者可以直接照抄。1. 项目概述与整体思路拆解1.1 目标拆解英雄皮肤图片到底怎么拿王者荣耀里的英雄皮肤图片很多人第一反应是去官网页面手动另存为或者用 selenium 模拟浏览器滚动截图。但我实际看了一眼页面结构发现根本不用费那个劲。官网前端的英雄资料页会加载一份公开的 JSON 文件里面把每个英雄的编号、名字、皮肤名称都列得清清楚楚皮肤图片的 URL 又有非常固定的拼接规律。只要把这份 JSON 拉下来解析出英雄 ID 和皮肤名称再按规则拼出图片地址就能批量下载。这个思路的优势很直接不依赖页面 DOM不用等浏览器渲染数据噪声也小。JSON 字段不稳定、页面改版HTML 解析经常跟着改而接口字段只要官网不改数据结构爬虫代码基本能稳定跑很久。项目总量也很好估算英雄数量目前在一百二十个左右每个英雄少则一两款皮肤、多则十几款整体图片任务量通常在六七百张这个量级。对协程来说这个规模刚好能体现出并发优势又不至于把目标服务器打得太狠。1.2 为什么这里非用协程不可爬虫任务八成时间都耗在网络 IO 上发请求、等响应、接收数据CPU 其实一直在旁边划水。我之前试过 ThreadPoolExecutor任务一多线程切换的开销就开始拖后腿换多进程更离谱为了下载几百张图片先给每个进程铺一整套解释器环境内存白白烧掉。协程是另一种玩法单线程内跑事件循环遇到网络等待就主动挂起当前任务去执行另一个还没拿到响应的任务等到数据回来了再接着往下走。同样是并发三者差别非常明显方案上下文切换内存占用IO密集任务表现代码复杂度多线程内核线程切换每线程约1MB栈空间尚可但数量多时开销大中等多进程进程切换最重每个进程独立解释器浪费CPU任务才划算较高协程用户态事件循环调度极低协程轻量最优单机也能顶大量任务中低这个项目里下载的全是静态图片每个文件尺寸不大但请求数量多、响应等待时间长天然就是协程的甜蜜区。再加上 aiohttp 自带连接池把连接复用做好之后几百个请求在同一个事件循环里跑得又稳又省资源。当然协程不是银弹如果是 CPU 密集型的图片压缩、特征计算那该上多进程还是得老老实实上。1.3 整体流程设计整个采集链路拆成四步拉英雄列表、解析皮肤信息、生成图片 URL、异步下载落盘。第一步用一个请求拿到英雄 JSON第二步遍历每个英雄的skin_name字段按竖线分割出所有皮肤名称第三步把英雄 ID 和皮肤索引拼成图片链接第四步是核心用协程并发下载所有图片同时做超时重试、断点续传和文件校验。程序结构上我分成了三个模块数据加载、任务构建、异步下载器。数据加载只负责拉 JSON 和基本解析任务构建负责把英雄信息转成一个个待下载任务异步下载器负责单个图片的请求、重试和文件写入。模块拆开之后后面想换成快手评论或者商品详情页爬虫只需要替换任务构建部分下载器完全不用动。这也是很多工程化爬虫的基本套路。2. 核心细节解析与实操要点2.1 公开接口解析思路从 JSON 到图片链接我用的公开接口返回的数组结构大致长这样[ { ename: 105, cname: 廉颇, skin_name: 正义重拳|地狱岩魂 }, { ename: 106, cname: 小乔, skin_name: 万圣前夜|天鹅之梦|纯白花嫁|缤纷独角兽 } ]ename是英雄数字 IDcname是英雄名skin_name用竖线把每个皮肤的名字串在一起。皮肤图片 URL 的规则是https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/{ename}/{ename}-bigskin-{idx}.jpg其中idx从 1 开始递增对应skin_name里每个皮肤的顺序。比如廉颇第一款皮肤图片就是105-bigskin-1.jpg第二款就是105-bigskin-2.jpg。这个规律很关键它意味着我不需要去解析任何 HTML只要把 JSON 里的皮肤数量和图片索引对齐即可。这里有个容易踩的小坑skin_name字段可能为空或者某些特殊英雄有异常数据。解析时一定要先判断skin_name是否存在再用split(|)分割同时用enumerate(..., start1)生成索引避免从 0 开始导致 URL 拼错。另外一定不要只拿英雄 cname 拼 URL部分英雄 ID 和历史数据不一致老老实实以ename为准。2.2 并发控制信号量、连接池、超时与重试协程最大的优点是可以同时挂起几十上百个请求但如果不加节制地全开目标服务器很容易出现响应变慢甚至拒绝连接。我用了两个机制来兜底asyncio.Semaphore控制协程任务并发数aiohttp.TCPConnector控制连接池上限。Semaphore(10)意思是同时最多只有 10 个下载任务在真正执行。这里和线程池类似超过 10 个的协程会排队等待等里面的任务完成一个才放一个进来。TCPConnector(limit20)则限制底层同时建立的 TCP 连接总数避免文件描述符被耗尽。两者配合既能平滑控制请求频率又不会让本机资源被拖垮。我自己实测下来并发 10 已经能跑满普通家庭宽带的下载速度再调大只是让服务器压力变大收益却不明显。超时和重试也是必选项。对于每张图片的请求我设置了aiohttp.ClientTimeout(total10)超过 10 秒就放弃。遇到连接错误、超时或者 5xx 状态码连续重试 3 次每次重试等待时间按0.5 * 2**attempt指数退避。这样应对服务器临时抖动很有效也能避免在高并发下因为一次网络抖动导致大片任务失败。2.3 文件命名与完整性校验图片下载下来之后直接用英雄名加皮肤名作为文件名最方便人看但 Windows 文件名不允许出现\ / : * ? |这些字符而英雄皮肤名字里偶尔会带特殊符号。我写了个清洗函数统一替换成下划线import re def clean_filename(name: str) - str: return re.sub(r[\\/:*?|\s], _, name).strip(_)保存目录按英雄 ID 分文件夹比如data/skins/105/廉颇_地狱岩魂.jpg这样按照英雄维度管理最省心。文件重名问题靠图片索引兜底我在文件名里加上了idx像小乔_4_缤纷独角兽.jpg避免名字相同的皮肤互相覆盖。完整性校验这块很容易被忽略。网络传输中断可能写出一半的损坏文件如果只检查文件是否存在下一次运行就会跳过这个坏文件。所以我用了一个小技巧先写入一个.tmp临时文件写完后用Path.replace()原子替换成正式文件。下载函数开头会检查正式文件是否存在且大小大于 0存在就直接跳过这样天然支持断点续传也不会写入半个文件。3. 实操过程与核心环节实现3.1 环境准备与依赖安装我用的 Python 版本是 3.10协程相关的asyncio是标准库不需要额外安装。第三方依赖只有两个aiohttp用于异步 HTTP 请求aiofiles用于异步文件写入。创建虚拟环境后安装即可python -m venv venv source venv/bin/activate pip install aiohttp aiofiles为什么特意用aiofiles因为在协程里直接使用open()写文件会阻塞事件循环一旦文件写入比较慢整个爬虫的节奏都会被拖慢。aiofiles把文件 IO 放到线程池里执行虽然是异步写法但底层不会卡住事件循环对高并发下载场景很重要。如果你的环境网络有特殊限制或者需要科学配置证书链可以在TCPConnector里传入自定义的ssl.SSLContext。不过我这个项目跑的是公开 CDN 图片默认证书校验就能通过不建议为了省事直接关闭校验。3.2 核心代码实现从英雄列表到批量下载下面这个是完整可运行版本我把关键逻辑都写在里面了import asyncio import logging import re from pathlib import Path import aiofiles import aiohttp HERO_LIST_URL https://pvp.qq.com/web201605/js/herolist.json SKIN_URL_TPL https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/{hero_id}/{hero_id}-bigskin-{idx}.jpg SAVE_ROOT Path(data/skins) CONCURRENCY 10 CONN_LIMIT 20 TIMEOUT 10 logging.basicConfig(levellogging.INFO, format%(asctime)s %(message)s) logger logging.getLogger(__name__) def clean_filename(name: str) - str: return re.sub(r[\\/:*?|\s], _, name).strip(_) async def fetch_hero_list(session: aiohttp.ClientSession) - list: async with session.get(HERO_LIST_URL, timeoutaiohttp.ClientTimeout(totalTIMEOUT)) as resp: resp.raise_for_status() return await resp.json() def build_tasks(heroes: list) - list: tasks [] for hero in heroes: hero_id hero.get(ename) hero_name hero.get(cname, str(hero_id)) skin_names hero.get(skin_name, ).split(|) for idx, skin_name in enumerate(skin_names, start1): if not skin_name: continue url SKIN_URL_TPL.format(hero_idhero_id, idxidx) tasks.append( { hero_id: hero_id, hero_name: hero_name, idx: idx, skin_name: skin_name, url: url, } ) return tasks async def download_one(session: aiohttp.ClientSession, sem: asyncio.Semaphore, task: dict) - str: hero_id task[hero_id] hero_name task[hero_name] idx task[idx] skin_name task[skin_name] url task[url] save_dir SAVE_ROOT / str(hero_id) filename clean_filename(f{hero_name}_{idx}_{skin_name}.jpg) save_path save_dir / filename if save_path.exists() and save_path.stat().st_size 0: return skipped async with sem: data None for attempt in range(3): try: async with session.get(url, timeoutaiohttp.ClientTimeout(totalTIMEOUT)) as resp: if resp.status 404: return not_found resp.raise_for_status() data await resp.read() break except (aiohttp.ClientError, asyncio.TimeoutError) as exc: logger.warning(下载失败 %s 第 %d 次: %s, url, attempt 1, exc) if attempt 2: return ffailed: {exc} await asyncio.sleep(0.5 * (2 ** attempt)) if data is None: return failed_no_data save_dir.mkdir(parentsTrue, exist_okTrue) tmp_path save_path.with_suffix(.tmp) async with aiofiles.open(tmp_path, wb) as fp: await fp.write(data) tmp_path.replace(save_path) return ok async def main() - None: connector aiohttp.TCPConnector(limitCONN_LIMIT) timeout aiohttp.ClientTimeout(totalTIMEOUT) async with aiohttp.ClientSession(connectorconnector, timeouttimeout) as session: heroes await fetch_hero_list(session) logger.info(英雄总数: %d, len(heroes)) tasks build_tasks(heroes) logger.info(待下载图片任务数: %d, len(tasks)) sem asyncio.Semaphore(CONCURRENCY) results await asyncio.gather( *(download_one(session, sem, task) for task in tasks), return_exceptionsTrue, ) stats: dict[str, int] {} for result in results: key result if isinstance(result, str) else exception stats[key] stats.get(key, 0) 1 logger.info(下载统计: %s, stats) if __name__ __main__: asyncio.run(main())代码里几个设计点我单独说一下。fetch_hero_list里用resp.json()直接解析 JSONaiohttp 内部会处理响应体编码比手动resp.text()再json.loads()更省事。build_tasks把每个皮肤作为一个独立任务对象字段齐全方便下载函数直接读取不会出现一边解析一边下载的混乱依赖。download_one里先检查文件是否存在再进入async with sem这个顺序很重要否则断点续传也会占用并发名额白白排队。3.3 运行结果与参数调整我在本地跑了一轮最终下载统计大致是这样总任务数在 700 左右成功 600 多张404 跳过 30 来张重试后失败的只有个位数整体耗时大约两分半。这个成绩和最开始 requests 单线程循环的十五分钟相比提升非常可观。实际调整参数时也要冷静一点。我把CONCURRENCY从 10 调到 30耗时确实又缩短了大半分钟但目标服务器明显出现了一些响应变慢的迹象个别请求开始连续重试。为了不给公共接口添麻烦我又调回了 10。对个人学习项目来说稳定和克制比单纯的速度更重要。另外如果本地磁盘比较慢可以试着把CONN_LIMIT调小一点避免大量临时文件同时写入导致 IO 抖动。4. 常见问题与排查技巧实录4.1 请求被拒、超时或 SSL 错误怎么办遇到 403、418 这类状态码大概率是并发太高触发了服务端的访问控制或者没带合理的请求头。我的方案是先调低CONCURRENCY再给ClientSession传一个常规的User-Agent比如headers {User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36} async with aiohttp.ClientSession(headersheaders) as session: ...如果出现超时优先看是不是单张图片体积太大或者目标 CDN 波动可以适当把TIMEOUT调大一点但不要超过 30 秒否则一条坏链路会卡住整个任务队列。SSL 证书报错通常是因为本机缺少根证书安装certifi后手动加载证书链即可不建议随便设sslFalse那会让流量明文传输在公共网络上尤其不合适。4.2 皮肤图片数量对不上或链接 404我刚开始跑的时候发现一个英雄的皮肤总数和下载成功数对不上查了半天才明白原因某些英雄的skin_name字段里皮肤名本身可能带竖线但接口返回时没有做转义导致我用split(|)多拆出几个不存在的皮肤索引。这类任务拼出来的 URL 自然就是 404。解决办法也很朴素遇到 404 时不要立刻当成程序错误先记录到日志里同时把该英雄的皮肤名数量和实际成功数对比。如果某个英雄的 404 数量明显异常就单独拉出原始 JSON 手工核对。更稳妥的做法是下载结束后按英雄维度做一次汇总看到哪个英雄任务多、成功少直接锁定问题范围比盯着几千行日志强多了。4.3 协程任务崩溃与断点续传协程代码如果不加保护一个下载任务抛出未捕获异常asyncio.gather会直接把整个主流程搅乱。所以我用了return_exceptionsTrue让每个任务的异常作为返回值返回再统一统计。这样个别图片失败只会产生一条记录程序不会中途死掉。断点续传是我这个案例里最实用的小设计。下载函数开头那句save_path.exists() and save_path.stat().st_size 0看起来不起眼实际上让整个程序具备了非常强的容错能力跑到一半 CtrlC 退出或者电脑断网下次重新执行会跳过所有已完成的文件只补剩下的。这也是我敢把并发调高的底气之一反正哪怕中途挂了重跑一遍成本也很低。5. 项目复盘与几点实在建议5.1 数据合规与君子协定这种基于公开接口的爬虫技术难度不高但它给数据源服务器带来的压力是真实存在的。我跑这个项目的时候特意把请求频率控制在很低的范围也没有做任何绕过访问控制的操作。图片数据仅用于个人学习研究不会打包传播或商用。想拿爬虫做产品化数据服务的朋友一定要先去读目标网站的用户协议和 robots.txt必要时联系官方获取授权。这不仅是职业操守问题也是保护自己不踩法律坑的基本常识。5.2 协程爬虫还能扩展什么这个项目的代码结构改成生产消费者模型也很容易。把asyncio.gather换成一个执行队列主线程负责往队列里塞任务消费者协程从队列里取任务执行可以应对任务动态追加的场景比如需要边爬边发现新链接。另外把图片下载换成 JSON 接口请求这套思路可以直接搬到商品价格监控、新闻聚合、开源社区数据采集这些方向上。对初学者来说先把这个图片爬虫吃透再去看asyncio.Queue、aiohttp的 WebSocket 支持会顺畅很多。最后再分享一个我实际摸出来的小经验跑协程爬虫时别光盯着速度一定要看错误日志。这个项目第一次全速跑完我发现有十几张图是坏文件但日志里没有明显报错排查后才发现是写入时没有用临时文件原子替换。后来把原子写和文件校验加上整个流程才真正算稳定。工具永远只是手段真正值钱的是你怎么设计容错和校验这也是爬虫从“会跑”到“跑得稳”之间那层窗户纸。