1. 异步批量下载到底解决了什么痛点做过数据采集或者资源归档的人多半遇到过这种场景手里攥着几百上千条文件链接可能是图片、PDF、日志包或者数据集分片用requests写个 for 循环挨个下载跑起来之后就只能干等着。单线程同步下载的问题很直白——每个请求都要经历建立连接、等待响应、接收数据、写入磁盘这一整套流程而其中绝大部分时间 CPU 都在空转全耗在等待网络 IO 上了。假设单个文件平均耗时 0.5 秒1000 个文件就是 500 秒将近 9 分钟这还没算上网络抖动导致的个别慢请求。异步批量文件下载要解决的核心问题就一个把等待时间重叠起来。当一个请求在等服务器响应时事件循环立刻切去处理另一个请求让几十上百个下载任务并发推进。同样是 1000 个文件合理设置并发度之后总耗时往往能压到同步方式的十分之一甚至更低。这不是玄学而是把原本被浪费的 IO 等待窗口充分利用起来了。这篇文章适合三类人看一是写过同步下载脚本、被速度折磨过的开发者二是想系统搞明白asyncio和aiohttp怎么配合干活的中级选手三是需要把下载能力封装成稳定工具、要考虑限流、重试、断点续传这些工程问题的朋友。我会从方案选型讲到完整可跑的代码再到实际踩过的坑尽量把每个参数为什么这么设都说清楚让你看完能直接抄作业也能明白抄的到底是什么。需要先明确一点异步下载不是银弹。它的优势场景是高并发、IO 密集、单文件不算特别大的批量任务。如果你要下载的是几个几十 GB 的大文件或者目标服务器对并发极其敏感那异步带来的收益会打折扣甚至可能因为触发风控而适得其反。搞清楚适用边界比盲目上异步更重要。2. 方案选型为什么是 asyncio 加 aiohttp2.1 同步、多线程、异步三条路对比在动手之前先把可选路径摆清楚免得选错方向白费功夫。批量下载这件事主流有三种实现思路各自的脾气完全不同。同步方式最好理解就是一个接一个下代码简单到闭着眼都能写调试也方便。但它的吞吐量被网络延迟死死卡住并发能力基本为零。多线程方式通过线程池把任务铺开确实能提速但线程的创建和切换有开销几百个线程之后 GIL 和上下文切换的代价就显现出来了而且线程安全问题、异常处理都会变复杂。异步方式则是单线程内用事件循环调度没有线程切换开销能把并发度做得更高资源占用还更低。方案并发能力资源占用代码复杂度适用场景同步 requests极低低最简单少量文件、调试多线程 ThreadPool中等较高中等几十并发、兼容老代码asyncio aiohttp高低较高成百上千并发下载选异步的核心理由是下载任务的瓶颈在网络 IO不在 CPU。异步模型天生就是为 IO 密集场景设计的一个事件循环就能调度成百上千个协程每个协程在等待网络时主动让出控制权这种协作式调度比线程的抢占式调度轻量得多。2.2 aiohttp 相比其他库的优势确定了异步路线接下来是选 HTTP 客户端库。asyncio本身只提供事件循环和协程调度不带 HTTP 能力得配一个异步 HTTP 库。常见的有aiohttp、httpx、requests配线程池这几种。httpx确实支持异步API 也优雅但它在高并发下的性能表现和连接池管理不如aiohttp成熟尤其是在需要精细控制连接数、超时、重定向的场景下。aiohttp是专门为asyncio生态打造的ClientSession内置连接池支持连接复用能显著减少 TCP 握手和 TLS 握手的开销——这一点在批量下载里非常关键因为每个文件都重新建连接的话握手成本会吃掉大量时间。提示aiohttp的ClientSession必须复用不要每个请求都新建一个 session。新建 session 意味着新建连接池连接复用就无从谈起了这是新手最容易犯的性能错误。2.3 并发度的取舍逻辑并发度设多少合适这是异步下载里最需要拿捏的参数。设太小比如 5那和同步差别不大浪费了异步的能力设太大比如 500可能把目标服务器打挂触发限流甚至封禁自己的内存和文件句柄也会吃紧。我的经验值是面向公开稳定的资源站并发度控制在 20 到 50 之间比较稳妥如果是自己可控的内网服务可以拉到 100 以上。这个数字不是拍脑袋来的它受几个因素制约本地文件描述符上限Linux 默认 1024每个连接占一个、目标服务器的承载能力、以及你自己的带宽。带宽往往是隐藏瓶颈——并发 100 个下载如果每个文件 1MB总带宽需求可能瞬间打满反而导致每个连接都变慢。一个实用的做法是用asyncio.Semaphore做信号量限流把并发度做成可配置参数先从小值试起观察目标站响应和本地资源占用再逐步往上调。后面实操部分我会给出具体代码。3. 核心机制拆解事件循环、协程与连接池3.1 事件循环是怎么把等待重叠起来的很多人对异步的理解停留在用了 async 就快了但没搞明白快在哪。这里用一个生活化的类比说清楚。想象你去银行办业务同步方式就像你排一个窗口办完一个才能办下一个中间柜员查资料、盖章的时间你只能干等。异步方式则像你拿了一沓号把材料递给柜员后不等结果立刻去另一个窗口递下一份材料哪个窗口先办完就先处理哪个。事件循环就是那个帮你盯着所有窗口、谁好了就叫你的调度员。技术上当一个协程执行到await语句比如等待网络响应时它会主动把控制权交还给事件循环事件循环转而去运行其他就绪的协程。等原来的网络响应回来了事件循环再把它唤醒继续执行。整个过程是单线程的没有线程切换所以开销极小。这就是为什么异步能在单线程里实现高并发——它靠的是主动让出而不是抢占切换。3.2 协程与任务的区别async def定义出来的是协程函数调用它得到的是协程对象但协程对象本身不会自动运行。要让它跑起来得用await直接等待或者用asyncio.create_task()把它包装成任务Task交给事件循环调度。这个区别在批量下载里很重要。如果你写await download(url)放在循环里那还是一个一个下因为await会阻塞当前协程直到这个下载完成。正确做法是先把所有下载协程用create_task变成任务让它们都进入调度队列然后用asyncio.gather()统一等待。这样所有任务才会真正并发推进。# 错误示范串行执行 for url in urls: await download(session, url) # 正确示范并发执行 tasks [asyncio.create_task(download(session, url)) for url in urls] await asyncio.gather(*tasks)3.3 连接池与 TCP 连接复用aiohttp的ClientSession内部维护一个连接池。当你发起请求时如果池里有到同一主机的空闲连接就直接复用省掉 TCP 三次握手和 TLS 握手的时间。对于批量下载同一站点的文件这个优化效果非常明显。连接池的大小由TCPConnector的limit参数控制默认是 100。这个值要和你的并发度匹配如果并发度是 50连接池 limit 设 100 就够用如果并发度 200 而 limit 只有 100多出来的请求就得排队等连接反而拖慢整体速度。所以这两个参数要一起调。connector aiohttp.TCPConnector(limit100, limit_per_host50) session aiohttp.ClientSession(connectorconnector)limit_per_host限制对单个主机的连接数这个参数在批量下载同一站点时特别有用能防止你把所有连接都怼到一个域名上既保护了对方服务器也避免触发风控。4. 完整实操从零搭一个可用的异步下载器4.1 环境准备与依赖安装先把环境搭起来。Python 版本建议 3.8 以上因为asyncio在 3.8 之后才稳定支持asyncio.run()这种高层 API。依赖就两个核心库外加一个可选的进度显示库。pip install aiohttp aiofiles tqdmaiohttp负责异步 HTTP 请求aiofiles负责异步写文件——这点容易被忽略普通open().write()是阻塞操作在异步环境里会卡住事件循环用aiofiles才能保持全程非阻塞。tqdm用来显示下载进度纯属提升体验不是必需。4.2 基础版下载器代码先上一个能跑的最小可用版本把核心骨架立起来。import asyncio import aiohttp import aiofiles import os async def download_one(session, url, save_dir, semaphore): async with semaphore: filename os.path.join(save_dir, url.split(/)[-1] or index.html) try: async with session.get(url, timeoutaiohttp.ClientTimeout(total60)) as resp: resp.raise_for_status() async with aiofiles.open(filename, wb) as f: async for chunk in resp.content.iter_chunked(8192): await f.write(chunk) return (url, True, None) except Exception as e: return (url, False, str(e)) async def batch_download(urls, save_dir, concurrency30): os.makedirs(save_dir, exist_okTrue) semaphore asyncio.Semaphore(concurrency) connector aiohttp.TCPConnector(limitconcurrency * 2, limit_per_hostconcurrency) async with aiohttp.ClientSession(connectorconnector) as session: tasks [download_one(session, url, save_dir, semaphore) for url in urls] results await asyncio.gather(*tasks) return results if __name__ __main__: urls [fhttps://example.com/file{i}.bin for i in range(100)] results asyncio.run(batch_download(urls, ./downloads)) ok sum(1 for _, success, _ in results if success) print(f成功 {ok} / {len(results)})这段代码里有几个关键点值得展开说。iter_chunked(8192)表示按 8KB 分块读取响应内容而不是一次性把整个文件读进内存。对于大文件来说这个细节决定了你的内存占用是几十 MB 还是几个 GB。Semaphore控制同时进行的下载数量async with semaphore保证进入临界区的协程不超过设定值。gather会等所有任务结束返回结果列表每个结果记录了成功与否和错误信息方便后续统计和重试。4.3 加入重试与断点续传基础版能跑但工程上不够健壮。网络抖动、服务器偶发 5xx、连接超时都是常态没有重试机制的下载器在生产环境里基本没法用。断点续传则能避免大文件下到一半失败后从头再来。async def download_with_retry(session, url, save_dir, semaphore, max_retries3): filename os.path.join(save_dir, url.split(/)[-1] or index.html) for attempt in range(max_retries): try: async with semaphore: headers {} mode wb existing 0 if os.path.exists(filename): existing os.path.getsize(filename) headers[Range] fbytes{existing}- mode ab async with session.get(url, headersheaders, timeoutaiohttp.ClientTimeout(total120)) as resp: if resp.status 416: return (url, True, already complete) resp.raise_for_status() async with aiofiles.open(filename, mode) as f: async for chunk in resp.content.iter_chunked(8192): await f.write(chunk) return (url, True, None) except Exception as e: if attempt max_retries - 1: return (url, False, str(e)) await asyncio.sleep(2 ** attempt)重试策略用的是指数退避第一次失败等 1 秒第二次等 2 秒第三次等 4 秒。这样做的原因是如果失败是因为服务器临时过载立刻重试只会加重它的负担退避等待能给它喘息时间成功率更高。断点续传靠Range请求头实现告诉服务器我已经有前 N 字节了从 N 开始给我。服务器返回 206 状态码表示支持返回 416 表示文件已经完整了。注意断点续传依赖服务器支持Range请求。有些服务器不支持会直接返回 200 和完整内容这时候如果以追加模式写入就会导致文件损坏。稳妥做法是先发一个 HEAD 请求检查Accept-Ranges头或者判断响应状态码是 206 才用追加模式。4.4 进度显示与结果统计批量下载跑起来之后你总得知道进度。tqdm配合asyncio.as_completed可以做到实时更新。from tqdm.asyncio import tqdm_asyncio async def batch_download_with_progress(urls, save_dir, concurrency30): os.makedirs(save_dir, exist_okTrue) semaphore asyncio.Semaphore(concurrency) connector aiohttp.TCPConnector(limitconcurrency * 2, limit_per_hostconcurrency) async with aiohttp.ClientSession(connectorconnector) as session: tasks [download_with_retry(session, url, save_dir, semaphore) for url in urls] results await tqdm_asyncio.gather(*tasks, desc下载进度) return resultstqdm_asyncio.gather是tqdm专门为异步场景提供的封装能在任务完成时自动更新进度条。跑完之后再统计一下成功失败数量把失败的 URL 单独存下来方便二次重试。5. 常见问题与排查技巧实录异步下载的坑大多集中在几个地方我把实际遇到过的典型问题和排查思路整理成表遇到问题可以直接对照。问题现象可能原因排查与解决报错 Unclosed client sessionsession 没正确关闭用async with管理 session 生命周期下载速度反而比同步慢并发度过高导致带宽打满降低 concurrency观察带宽占用大量请求超时目标站限流或连接池太小降低并发、增大 timeout、调大 connector limit文件内容不完整未等待写入完成或断点续传判断错误检查iter_chunked循环是否完整执行内存占用飙升一次性读取整个响应改用iter_chunked分块写入报错 Too many open files文件描述符耗尽降低并发或调高系统 ulimit5.1 事件循环被阻塞的隐蔽陷阱最常见的性能杀手是在异步代码里混入了同步阻塞调用。比如用requests发请求、用普通open写文件、用time.sleep等待这些都会卡住整个事件循环让所有协程一起停摆。表现就是明明用了异步速度却上不去。排查方法很简单在下载函数里打印时间戳如果发现多个任务的日志是严格串行的那基本就是被阻塞了。解决办法是把所有 IO 操作都换成异步版本——requests换aiohttpopen换aiofilestime.sleep换asyncio.sleep。5.2 并发度调优的实测方法并发度不是设完就不管的得实测调优。我的做法是准备一组固定的测试 URL从 concurrency10 开始每次翻倍记录总耗时和失败率。通常会出现一个拐点并发度增加到某个值之后总耗时不再下降甚至反弹失败率开始上升。那个拐点前一点的值就是比较优的配置。实测中还有个反直觉的现象并发度太高时单个请求的平均耗时反而变长。原因是连接数太多每个连接分到的带宽变少加上服务器端排队整体效率下降。所以并发越高越快是个误区找到平衡点才是关键。5.3 目标站风控的应对思路批量下载很容易触发目标站的风控表现为突然大量 403、429或者连接被重置。应对思路有几个层次一是控制并发度别把人家服务器当自家后院二是加请求间隔用asyncio.sleep在请求之间插入随机延迟三是设置合理的 User-Agent别用默认的库标识四是尊重robots.txt和站点的使用条款。提示如果目标站返回 429Too Many Requests通常响应头里会带Retry-After告诉你多久之后可以重试。解析这个头并据此退避比盲目重试要礼貌也有效得多。6. 进阶优化与工程化建议6.1 用队列做生产者消费者解耦当 URL 来源是动态的比如从数据库分页读取或者边解析页面边发现新链接用固定列表就不合适了。这时候可以用asyncio.Queue搭一个生产者消费者模型生产者负责往队列里塞 URL若干个消费者协程从队列里取任务下载。这样能实现边发现边下载内存占用也更可控。async def worker(queue, session, save_dir, semaphore): while True: url await queue.get() if url is None: queue.task_done() break await download_with_retry(session, url, save_dir, semaphore) queue.task_done()消费者数量就是并发度通过启动多个 worker 协程来控制。任务全部处理完后往队列里塞对应数量的None作为结束信号worker 收到就退出。6.2 限速与带宽控制有些场景下你不希望下载占满全部带宽比如同时还在做别的事情。这时候需要限速。aiohttp本身不直接提供限速但可以在写入时手动控制节奏记录已写入的字节数和耗时如果速率超过阈值就await asyncio.sleep等一下。实现起来不复杂核心就是算一下当前速率超了就歇会儿。6.3 日志与可观测性生产环境用的下载器日志不能少。至少记录三类信息每个文件的开始和结束时间、失败原因、整体统计。日志建议用结构化格式比如 JSON方便后续用工具分析。别用print用标准库的logging并且注意日志本身也是 IO高频写日志同样会拖慢速度可以攒一批再写。6.4 打包成命令行工具如果这个下载器要反复用把它包成命令行工具会方便很多。用argparse接收 URL 列表文件、保存目录、并发度等参数再配一个if __name__ __main__入口。这样别人不用看代码直接python downloader.py -f urls.txt -o ./data -c 50就能用。import argparse parser argparse.ArgumentParser(description异步批量文件下载器) parser.add_argument(-f, --file, requiredTrue, helpURL 列表文件每行一个) parser.add_argument(-o, --output, default./downloads, help保存目录) parser.add_argument(-c, --concurrency, typeint, default30, help并发度) args parser.parse_args() with open(args.file) as f: urls [line.strip() for line in f if line.strip()] results asyncio.run(batch_download_with_progress(urls, args.output, args.concurrency))我个人在实际操作中的体会是异步下载器写出来容易写好难。真正拉开差距的不是那几行async/await而是对并发度的拿捏、对异常的处理、对目标站的尊重。我踩过最深的坑就是一开始贪心把并发度设到 200结果目标站直接把我 IP 限了半小时后来老老实实降到 30 加随机延迟反而稳定跑完了全部任务。所以别追求极限并发追求的是稳定把活干完。另外一个小技巧把失败的 URL 单独落盘写个只重试失败项的小脚本比整个任务重跑一遍省事得多。