前段时间整理技术笔记的时候翻到一个老项目用 Python 协程抓取王者荣耀全部英雄皮肤图片。当时正好在啃 asyncio 和 aiohttp顺手拿这个需求当练手跑通之后对“异步并发”的理解直接上了一个台阶。这个项目本身不复杂核心就是“公开接口解析 并发下载”但它把协程爬虫的完整链路都串起来了先请求官方英雄列表 JSON解析每个英雄的皮肤数量和命名规则再用 aiohttp 并发下载所有皮肤原图。不管你是刚开始学 Python 爬虫还是想搞懂协程到底怎么落地都可以跟着这个案例过一遍。跑完之后你会发现协程不是玄学它就是一套“等 IO 的时候顺手干别的活”的调度模型。1. 项目背景与整体设计思路1.1 这个案例实际解决什么问题爬取王者荣耀英雄皮肤图片听起来像是一个“图集采集”的小需求但背后涉及的技术点非常典型。它不是一个孤立的单页面爬虫而是典型的“列表页 详情页 二进制资源”三级模型先拿 JSON 列表再根据列表字段构造出所有图片地址最后批量下载图片文件。这类模型在电商商品图采集、表情包下载、壁纸站抓取、素材库同步等场景里几乎天天都能看到。举个例子你在工作里接到的需求可能是“把公司官网所有产品图同步到本地”或者“把某个素材站的全部分类图标备份一份”本质上和这个案例一模一样。区别只是接口字段不同、图片 URL 规则不同。所以这个项目练的不是“爬王者荣耀”本身而是练一套可以复用到任意“批量图片采集”场景的能力。我当时做这个案例时还特意确认了数据来源王者荣耀官网公开的英雄数据接口不需要登录、不需要加密参数纯粹是静态 JSON。这种数据源特别适合做教学案例因为它把“反爬”难度降到最低让重心能放在“异步请求、任务调度、错误处理”这些真正值得学的部分。1.2 为什么用协程而不是多线程这是整个项目最值得说清楚的问题。很多人一听到“批量下载图片”第一反应就是开多线程。这个思路没有错但如果是 Python就得想清楚 GIL 和线程开销的问题。先说一个基本事实图片下载是典型的 IO 密集型任务大量时间花在网络传输上CPU 基本是空闲的。对于 IO 密集型任务协程是性价比很高的方案。它的核心逻辑是程序在发出一个网络请求后不会傻等响应而是主动让出控制权去执行另一个还没完成的请求等响应回来了事件循环再切回来继续处理。整个过程只用一个线程却可以管理成百上千个“半挂起”的任务。对比一下三种方案同步 requests600 张图片就 600 次串行请求假设单张耗时 200 毫秒总耗时 120 秒纯纯浪费时间。多线程理论上 20 个线程可以把时间压到 6 秒左右但线程本身有创建和切换开销每个线程占用独立栈内存开太多会触发系统调度瓶颈。asyncio aiohttp同样 20 个并发一个线程就能搞定任务切换开销远小于线程切换可以把并发数开得更高实测 30~50 并发也非常稳定。这里不是要把多线程一棒子打死。多线程在“阻塞式 IO 复杂业务逻辑”的场景里依然有它的位置。但在纯粹的 HTTP 批量请求场景里协程的代码写起来更简洁资源占用更低可控性也更强。尤其是asyncio.Semaphore可以直接控制并发上限想限速就限速调整一个数字的事。1.3 整体流程拆解这个项目的完整流程可以分为四个阶段。第一个阶段是抓取英雄列表。通过官方接口拿到一个 JSON 数组里面每个元素代表一个英雄包含英雄 ID、名字、皮肤名等字段。第二个阶段是解析字段构造图片 URL。这一步是关键因为皮肤图片并不是现成给出来的而是按照固定规则生成的。只要知道英雄 ID 和皮肤序号就能拼出对应的图片地址。第三个阶段是构建协程任务。每一个皮肤图片对应一个下载任务把所有任务丢给事件循环去调度。因为任务数量很大需要控制并发避免一次发起太多请求把服务器惹毛。第四个阶段是保存到本地。按英雄分目录用皮肤名字作为文件名同时做非法字符过滤防止 Windows 路径报错。这四个阶段看起来简单但每一步都有细节坑。尤其是第二阶段如果字段理解错了拼出来的 URL 就是 404第四阶段如果文件名不处理在 Windows 上会直接抛异常。后面我会把每一步的代码都拆开讲清楚。2. 环境准备与接口分析2.1 环境依赖安装项目基于 Python 3.8 以上的版本即可不需要太新也不需要太老。核心依赖只有两个aiohttp 负责异步 HTTP 请求当然你也可以顺手装一个 aiofiles 用来异步写文件但图片体量不大普通open()写也能接受。pip install aiohttp aiofiles验证安装是否成功直接进 Python 执行import aiohttp不报错就说明环境没问题。如果你用的是 Anaconda建议先建一个独立的虚拟环境再装避免和 base 环境里的包冲突。conda create -n spider python3.10 conda activate spider pip install aiohttp aiofiles在 Windows 上有一个额外注意点Python 3.8 之后的版本默认事件循环策略已经改成了 ProactorEventLoopaiohttp 可以正常工作。但如果你在旧代码里看到asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())这类写法并且跑的 Python 版本比较老需要留意 SelectorEventLoop 对某些功能的限制。建议直接用 Python 3.10 或者更新的版本省掉这些兼容性问题。2.2 英雄列表接口字段解读王者荣耀官网有一个公开的英雄列表接口https://pvp.qq.com/web201605/js/herolist.json这个接口返回的是一个 JSON 数组数组中每个对象代表一个英雄。截取一段实际数据来看[ { ename: 105, cname: 廉颇, title: 正义爆轰, skin_name: 正义爆轰|地狱岩魂 }, { ename: 106, cname: 小乔, title: 恋之微风, skin_name: 恋之微风|万圣前夜|天鹅之梦|纯白花嫁|缤纷独角兽 } ]这里最重要的两个字段是ename和skin_name。ename是英雄的数字 ID构造图片 URL 时必须用到skin_name是英雄所有皮肤的名称多个皮肤名之间用竖线|分隔皮肤的数量就是分割后的列表长度。有一个比较容易踩的坑skin_name在某些英雄里可能不存在或者为空字符串。比如刚上线的新英雄如果接口还没更新完整skin_name可能是一个空值直接调用.split(|)会报错。所以解析时要用get(skin_name)配合兜底处理例如hero.get(skin_name, ) or 。另外ename是数字类型不是字符串。在拼 URL 时如果你用字符串格式化要注意类型转换。虽然format会自动转成字符串但如果你是手动拼字符串比如hero-info/ hero_id /就要提前str()一下否则会直接报 TypeError。2.3 皮肤图片 URL 构造规律皮肤图片的地址遵循一套非常固定的模板。我抓包验证过原图的链接格式如下https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/{ename}/{ename}-bigskin-{seq}.jpg其中{ename}是英雄 ID{seq}是皮肤序号从 1 开始递增。也就是说小乔有 5 个皮肤那么对应的图片就是https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/106/106-bigskin-1.jpg https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/106/106-bigskin-2.jpg https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/106/106-bigskin-3.jpg ...这个规律的意义在于你不用去网页源码里找每一个皮肤的 CDN 链接只需要英雄 ID 和皮肤数量就可以推导出所有图片地址。这也是很多素材类爬虫的通用思路接口返回的只是元数据真实资源地址往往藏在命名规则里。除了bigskin原图还有smallskin小图和中间尺寸的图片。日常练习爬取bigskin就够了图片质量最高下载下来还能当壁纸。如果你想要更清晰的图片也可以观察一下同一套皮肤是否存在-bigskin-和-m-之类的变体不同游戏的 CDN 规则不太一样。需要注意不是所有序号都能拼出有效图片。极少数皮肤可能因为下架或数据更新延迟导致对应序号的图片返回 404。所以在下载函数里必须做状态码判断和重试机制这一点后面代码部分会详细说。3. 协程爬虫核心代码实现3.1 异步获取英雄列表既然要用协程那第一步请求接口就直接上 aiohttp。这里有几个关键点一个是复用ClientSession不要每个请求都新建一个 session另一个是设置合理的请求头至少带上 User-Agent最好也带上 Referer模拟真实浏览器行为。import asyncio import aiohttp import json import os import re BASE_URL https://pvp.qq.com/web201605/js/herolist.json IMG_URL_TPL https://game.gtimg.cn/images/yxzj/img201606/skin/hero-info/{hero_id}/{hero_id}-bigskin-{seq}.jpg HEADERS { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36, Referer: https://pvp.qq.com/ } SAVE_DIR hero_skins async def fetch_hero_list(session): async with session.get(BASE_URL, headersHEADERS, timeout15) as resp: resp.raise_for_status() return await resp.json()这段代码中timeout15是整体超时aiohttp 也支持aiohttp.ClientTimeout(total15)单独设置。网络不稳定的时候建议把超时设得宽松一点但也不要太长否则某个卡死的请求会拖住整个任务。如果接口请求失败raise_for_status()会抛出异常后续可以在main()里统一处理重试或者记录日志。await resp.json()是 aiohttp 内置的 JSON 反序列化方法它会读取响应体并用标准库解析。如果你接口返回的是 GBK 编码的文本就需要用await resp.text(encodinggbk)拿到文本后再json.loads()。这个案例的接口是 UTF-8所以直接.json()没问题。3.2 皮肤图片任务队列构建拿到英雄列表后要做两件事一是按英雄创建本地目录二是把所有皮肤图片的下载任务收集到一个列表里。这里的核心循环逻辑很直白def build_tasks(heroes): tasks [] for hero in heroes: ename hero.get(ename) cname hero.get(cname, unknown) skin_names hero.get(skin_name, ) if not skin_names: skin_names 默认 skin_list skin_names.split(|) hero_dir os.path.join(SAVE_DIR, f{ename}_{cname}) os.makedirs(hero_dir, exist_okTrue) for seq, skin_name in enumerate(skin_list, 1): img_url IMG_URL_TPL.format(hero_idename, seqseq) safe_name re.sub(r[\\/:*?|], _, skin_name.strip()) file_path os.path.join(hero_dir, f{seq}_{safe_name}.jpg) tasks.append((img_url, file_path)) return tasksskin_list的长度就是皮肤数量enumerate从 1 开始正好对应图片 URL 中的序号。文件名用序号_皮肤名.jpg好处是排序自然而且即使皮肤名重复序号也能保证文件不覆盖。这里说一下为啥要对skin_name做非法字符替换。Windows 文件系统不允许文件名包含\ / : * ? |这几个字符而皮肤名里确实可能带特殊符号比如“铠·绛天战甲”里的点还有个别皮肤里的“/”。如果不处理open()写文件的时候就会直接抛OSError整个协程任务中断。正则替换成下划线是最简单的做法保留可读性。3.3 协程下载函数与信号量控制下载函数是整个项目的心脏。它要做三件事限制并发量、发起 GET 请求、保存字节内容。先说并发限制使用asyncio.Semaphore可以把同时进行的请求数限制在指定值以内。async def download_one(session, sem, url, path): async with sem: for attempt in range(3): try: async with session.get(url, headersHEADERS, timeout10) as resp: if resp.status 200: data await resp.read() with open(path, wb) as f: f.write(data) return True else: print(f状态码异常: {url} - {resp.status}) if resp.status in (404, 403): break except (asyncio.TimeoutError, aiohttp.ClientError) as e: print(f第 {attempt 1} 次失败: {url} | {e}) await asyncio.sleep(2) return False这里有几个细节要展开。一个是sem必须放在 for 循环外面。async with sem保证同一时间最多只有 N 个协程进入下载逻辑。如果放在 for 循环里面重试时也会占用一个信号量实际并发会比预期高不太合适。另一个是状态码为 404 时直接break不再重试。404 表示资源不存在重试 100 次还是 404没必要浪费请求。403 大概率是风控拦截了重试也没意义。只有超时和连接类异常才值得重试因为网络抖动是临时性的。还一个是await resp.read()会把整个图片内容读到内存中。单张皮肤原图通常只有几十 KB 到几百 KB这个消耗可以忽略。如果你要下载的是视频或者大文件那就必须改用resp.content.iter_chunked()流式写入不能一次性读进内存否则内存会炸。3.4 主入口与任务聚合asyncio.gather()可以并发执行所有任务并拿到每个任务的返回值。配合asyncio.Semaphore(20)就是把 500 多个下载任务丢给事件循环同时最多只有 20 个在真正跑。async def main(): sem asyncio.Semaphore(20) async with aiohttp.ClientSession() as session: heroes await fetch_hero_list(session) print(f获取英雄数量: {len(heroes)}) task_args build_tasks(heroes) print(f共需下载图片: {len(task_args)}) results await asyncio.gather( *[download_one(session, sem, url, path) for url, path in task_args] ) success sum(1 for r in results if r) failed len(results) - success print(f下载完成: 成功 {success} 张, 失败 {failed} 张) if __name__ __main__: asyncio.run(main())这里有一个容易被忽略的优化点创建 500 多个协程任务本身是开销很小的但如果你的图片数量到了几万张一次性把所有任务都创建出来再gather会导致事件循环里的任务对象很多。更稳妥的做法是使用asyncio.Queue 固定数量 worker 的生产者消费者模型。图片数量在几千以内的场景直接用gather更简洁代码也更容易读。另外注意ClientSession必须在async with块里使用。之前见过有人把ClientSession()放到协程外面创建导致Unclosed session警告严重时还会出现连接池耗尽。正确做法就是上面代码里的写法会话结束后自动关闭。3.5 完整代码整合把上面三段代码拼起来就是一个可以直接运行的协程爬虫。再把关键参数做一个汇总参数建议值说明并发数20太低跑得慢太高容易触发限流超时时间10~15 秒网络差时可调大重试次数33 次以内足够覆盖大部分临时错误重试间隔2 秒避免连续失败时请求太密集图片格式jpg官方接口返回的就是 JPEG完整代码的流程是获取英雄列表 - 构造任务 - 并发下载 - 收集结果。每个环节都有打印输出方便排查。下载完成后本地会生成一个hero_skins目录里面每个英雄一个子目录。补充一个提升健壮性的小技巧在main()里加一个try/except捕获顶层异常保证即使某个环节崩了控制台也能看到完整堆栈而不是直接无响应。try: asyncio.run(main()) except KeyboardInterrupt: print(用户中断) except Exception as e: print(f主流程异常: {e})4. 常见问题与排查技巧实录4.1 SSL 证书报错运行 aiohttp 时最常见的报错是ssl.SSLCertVerificationError或者aiohttp.client_exceptions.ClientConnectorCertificateError。这个问题的根源是 Python 的 SSL 校验没通过。官方 CDN 的证书理论上是正常的但某些网络环境下公司代理、路由器劫持等证书链会出问题。最简单的处理方式是创建一个不校验证书的TCPConnectorimport ssl ssl_ctx ssl.create_default_context() ssl_ctx.check_hostname False ssl_ctx.verify_mode ssl.CERT_NONE connector aiohttp.TCPConnector(sslssl_ctx) session aiohttp.ClientSession(connectorconnector)我自己实际测试下来这个方案在实验室网络和企业代理环境下都能跑通。但要注意关闭 SSL 校验只是绕过问题不是解决问题。如果追求严谨可以把目标证书链下载下来手动指定cafile参数。不过对于学习项目来说关闭校验完全够用。还有一类连接报错是Can not connect to host或者Connection refused这通常不是代码问题而是网络本身无法访问目标主机。先试试浏览器能不能打开图片地址如果浏览器能打开而代码打不开再检查 headers 和代理设置。4.2 图片下载不完整或打不开下载下来的图片放到本地后发现文件损坏或者打不开。原因通常是两种一种是请求被中断read()只读到了一部分字节另一种是返回的 200 响应体根本不是图片而是 HTML 错误页或者验证页面。排查方法是检查文件后缀和文件头。一个标准 JPEG 文件的前三个字节是FF D8 FF可以用 Python 快速验证with open(path, rb) as f: head f.read(3) if head ! b\xff\xd8\xff: print(不是有效 JPEG 文件:, path)如果出现大量无效图片最好把resp.content_type打出来看看。如果 CDN 返回了text/html说明请求被路由到了错误页。另外图片大小也可能是一个指标。很多正常皮肤图至少 1 KB 以上如果下载到的是 100 字节不到的文件基本可以判定下载异常。在download_one里判断len(data) 1000就是校验手段之一。if len(data) 1000: print(f文件过小疑似异常: {url}) return False4.3 并发效率上不去有些同学跑这个脚本发现同时下载的数量始终上不去或者整体耗时和串行差不多。这种情况十有八九是Semaphore没生效或者代码把await写掉了。协程的特性是一个任务遇到await才会让出控制权整个事件循环才会去执行其他任务。如果你在下载函数里用了requests.get()这种同步库await不会起作用并发自然就没了反而会卡死事件循环。所以下载请求必须用 aiohttp不能用 requests哪怕是requests的单次请求非常快也不行。还有一个容易踩的坑是ClientSession的maxsize连接池限制。aiohttp 默认的 TCPConnector 连接数限制是 100一般情况下够用。但如果你的并发开到 50 以上又确认代码没问题可以显式设置connector aiohttp.TCPConnector(limit100, limit_per_host30)limit_per_host可以限制同一主机的并发连接数对单个 CDN 域名的友好访问也是必要的。4.4 反爬限流应对策略王者荣耀官方接口本身没有做很严格的反爬但我们写爬虫时要主动做好限速。不要用 100 并发去怼同一个接口很容易触发 CDN 的风控返回 403 或者验证码页面。一个成熟的策略是“动态并发 可变延迟”。核心思路是给每个协程加一个随机延迟async def download_one(session, sem, url, path): async with sem: await asyncio.sleep(random.uniform(0.1, 0.5)) # 然后发起请求这样 20 并发下每个请求之间会有 100 到 500 毫秒的随机间隔整体速率波动不会太规律。还可以在状态码 429 或 503 时根据Retry-After响应头动态等待retry_after resp.headers.get(Retry-After) if retry_after: await asyncio.sleep(float(retry_after) 1)此外不要忘记设置Accept和Accept-Language请求头。有些 CDN 会检查这些头字段缺失时可能返回压缩版或者错误内容。参考浏览器的正常请求头是最稳妥的。5. 优化扩展与爬虫素养5.1 断点续传与增量更新第一次跑全量下载很顺利但过了俩月游戏更新了新皮肤这时候再跑一次全量会把已经下载好的几百张图又下一遍。合理的做法是增量更新。简单方案是维护一个已下载清单。每次下载前检查本地文件是否存在存在就跳过。但这会有一个问题如果之前下载的是损坏文件跳过就永远错下去。所以建议用“文件存在且大小大于阈值”作为判断条件if os.path.exists(path) and os.path.getsize(path) 1024: return True更严谨的方案是做一个manifest.json记录每个皮肤的下载时间和文件大小。增量更新时只请求那些不在 manifest 中或者大小变化的图片。这个方案还能顺便排查哪个皮肤在历史上曾经下载失败。5.2 多进程融合的扩展思路如果图片规模上升到几十万张单机单进程的协程可能还是不够。这时候可以考虑multiprocessing和asyncio的结合每个进程负责一部分英雄进程内部再用协程并发下载。实现上不需要手动管进程池写一个worker_process(hero_ids)函数然后用multiprocessing.Pool分发即可。这种架构的好处是能利用多核 CPU 来做解析和文件写入坏处是代码复杂度上升而且每个进程都会创建独立的 ClientSession内存占用也翻倍。对于学习项目我的建议是先跑通单进程协程版再考虑多进程。不要一上来就套一个复杂框架否则出了问题你连排查的基础都没有。5.3 爬虫的合规与自我约束最后必须聊一下边界问题。写爬虫可以做技术研究但不代表可以随便爬。具体到这个项目需要注意几个点第一数据来源是官方公开接口本身没有登录墙也没有付费内容。爬这类公开数据相对安全但仍要控制频率不要给对方服务器造成压力。第二下载的图片资源版权归属原公司。自己收藏学习、做个人壁纸没问题但不能拿去二次分发、商用或者包装成所谓的“图包”售卖。第三如果目标是其他网站务必先看robots.txt再看看网站的用户协议。很多平台明确禁止爬虫抓取这种情况下即使技术上可行也不应该去碰。第四爬虫代码不要写“死循环 高并发”的野蛮模式。给自己设一个下载上限跑完就停这对维护自己的 IP 和对方的服务器都友好。说实话我在写爬虫这条路上踩过不少坑最大的一个体会就是技术能力决定你能走多快边界感决定你能走多远。协程、异步这些工具是放大器用在学习上收获巨大用在投机取巧上风险也巨大。希望大家拿这个案例跑通流程之后能把它当成理解异步编程的垫脚石而不是批量采集素材的免费工具。