核心内容摘要
(200分)- 九宫格按键输入（Java JS Python）
在网络爬虫场景中图片批量下载是高频需求传统同步下载模式受限于 I/O 阻塞面对成百上千张图片时效率极低。
asyncio作为 Python 内置的异步 I/O 框架通过非阻塞协程机制可大幅提升图片下载并发能力但仅基础使用难以发挥最优性能需从并发控制、请求优化、异常处理、资源管理等维度进行系统性优化。
本文将详细讲解基于asyncio的大量图片异步下载优化方案结合aiohttp实现高效、稳定的图片爬取。
asyncio aiohttp 基础异步图片下载原理asyncio的核心是事件循环和协程协程在 I/O 操作如网络请求、文件写入时主动让出 CPU事件循环调度其他协程执行避免同步模式下的线程阻塞。
图片下载属于典型的 I/O 密集型任务aiohttp作为异步 HTTP 客户端与asyncio深度适配可发起非阻塞的图片请求。
基础实现代码框架python运行import asyncio import aiohttp import os from pathlib import Path # 图片保存目录 SAVE_DIR Path(async_images) SAVE_DIR.mkdir(exist_okTrue) async def download_single_image(session: aiohttp.ClientSession, img_url: str, img_name: str): 单张图片异步下载 try: async with session.get(img_url, timeout
as response: if response.status 200: # 读取图片二进制数据 img_data await response.read() # 异步写入文件 with open(SAVE_DIR / img_name, wb) as f: f.write(img_data) print(f成功下载{img_name}) else: print(f下载失败状态码{response.status}URL{img_url}) except Exception as e: print(f下载异常{str(e)}URL{img_url}) async def batch_download_images(img_urls: list): 批量图片异步下载入口 # 创建异步HTTP会话 async with aiohttp.ClientSession() as session: # 创建协程任务列表 tasks [ download_single_image(session, url, fimage_{idx}.jpg) for idx, url in enumerate(img_urls) ] # 并发执行所有任务 await asyncio.gather(*tasks) if __name__ __main__: # 测试图片URL列表 test_img_urls [ https://example.com/image
jpg, https://example.com/image
jpg, # 更多图片URL... ] # 启动事件循环 asyncio.run(batch_download_images(test_img_urls))上述代码是基础异步下载实现但存在明显缺陷无并发限制大量协程同时发起请求会导致目标服务器拒绝连接、本地网络阻塞甚至触发反爬机制无异常重试机制单次网络波动就会导致图片下载失败文件写入同步阻塞openwrite为同步操作会拖慢整体效率无资源复用与请求优化重复请求头、无效连接会浪费资源。
这些问题是大量图片下载时的核心瓶颈需针对性优化。
核心优化点一并发控制避免无限制协程创建asyncio.gather(*tasks)会一次性创建所有协程任务当图片数量达数千、数万时瞬间发起大量 HTTP 请求会引发三大问题目标服务器触发流量控制返回 429请求过多、503服务不可用状态码本地 TCP 连接数超限导致端口耗尽、请求超时触发网站反爬策略封禁 IP。
基于信号量Semaphore的并发限制asyncio.Semaphore是异步锁机制可限制同时执行的协程数量将并发数控制在合理范围建议
根据服务器承受力调整。
优化代码实现python运行import asyncio import aiohttp from pathlib import Path SAVE_DIR Path(async_images) SAVE_DIR.mkdir(exist_okTrue) # 定义最大并发数根据实际情况调整 MAX_CONCURRENT_TASKS 20 # 创建信号量限制并发 semaphore asyncio.Semaphore(MAX_CONCURRENT_TASKS) async def download_single_image(session: aiohttp.ClientSession, img_url: str, img_name: str): # 申请信号量超过并发数则阻塞等待 async with semaphore: try: async with session.get(img_url, timeout
as response: if response.status 200: img_data await response.read() with open(SAVE_DIR / img_name, wb) as f: f.write(img_data) print(f成功下载{img_name}) else: print(f下载失败状态码{response.status}URL{img_url}) except Exception as e: print(f下载异常{str(e)}URL{img_url}) async def batch_download_images(img_urls: list): async with aiohttp.ClientSession() as session: tasks [ download_single_image(session, url, fimage_{idx}.jpg) for idx, url in enumerate(img_urls) ] await asyncio.gather(*tasks)
分批次并发执行除信号量外可将图片 URL 列表切分为多个批次每批次执行固定数量的协程避免一次性创建大量任务降低内存占用和请求压力。
python运行async def batch_download_images(img_urls: list, batch_size: int
: async with aiohttp.ClientSession() as session: # 切分批次 for i in range(0, len(img_urls), batch_size): batch_urls img_urls[i:ibatch_size] tasks [ download_single_image(session, url, fimage_{idxi}.jpg) for idx, url in enumerate(batch_urls) ] # 每批次执行完成后再处理下一批 await asyncio.gather(*tasks) print(f完成第{i//batch_size 1}批次下载共{len(batch_urls)}张)
核心优化点二异常处理与重试机制提升下载成功率网络环境不稳定、服务器临时故障、DNS 解析失败等问题会导致部分图片下载失败。
基础代码无重试逻辑大量图片下载时失败率会显著上升需通过异常捕获分类、自动重试、失败记录优化。
基于 tenacity 的异步重试tenacity是 Python 重试库支持异步函数可设置重试次数、重试间隔、重试条件针对网络超时、连接错误等可恢复异常自动重试。
安装依赖bash运行pip install tenacity优化代码实现python运行import asyncio import aiohttp from pathlib import Path from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type SAVE_DIR Path(async_images) SAVE_DIR.mkdir(exist_okTrue) MAX_CONCURRENT_TASKS 20 semaphore asyncio.Semaphore(MAX_CONCURRENT_TASKS) # 记录失败的图片URL failed_urls [] # 重试配置最多重试3次指数退避等待1s、2s、4s仅重试网络相关异常 retry( stopstop_after_attempt(
, waitwait_exponential(multiplier1, min1, max
, retryretry_if_exception_type(( aiohttp.ClientError, # 客户端异常连接失败、超时等 asyncio.TimeoutError, # 异步超时 OSError # 系统I/O异常 )), reraiseTrue # 重试失败后抛出异常 ) async def download_single_image(session: aiohttp.ClientSession, img_url: str, img_name: str): async with semaphore: try: async with session.get(img_url, timeout
as response: if response.status 200: img_data await response.read() with open(SAVE_DIR / img_name, wb) as f: f.write(img_data) print(f成功下载{img_name}) else: # 非200状态码不重试直接记录失败 raise Exception(fHTTP状态码异常{response.status}) except Exception as e: print(f下载失败重试后{str(e)}URL{img_url}) failed_urls.append(img_url) raise # 抛出异常触发重试或记录
失败 URL 持久化与二次下载将最终失败的图片 URL 保存至文件任务结束后可二次扫描下载避免数据丢失。
python运行def save_failed_urls(): 保存失败URL至文件 with open(failed_images.txt, w, encodingutf-
as f: f.write(\n.join(failed_urls)) print(f失败URL已保存共{len(failed_urls)}条) async def batch_download_images(img_urls: list): async with aiohttp.ClientSession() as session: tasks [ download_single_image(session, url, fimage_{idx}.jpg) for idx, url in enumerate(img_urls) ] await asyncio.gather(*tasks) # 任务结束后保存失败URL save_failed_urls()
核心优化点三异步文件写入消除同步阻塞基础代码中with open(...) as f: f.write(img_data)是同步文件操作当大量协程同时写入文件时会因磁盘 I/O 阻塞导致协程等待降低整体并发效率。
需使用异步文件库实现非阻塞写入。
基于 aiofiles 的异步文件操作aiofiles是 Python 异步文件操作库与asyncio兼容实现文件读写的非阻塞执行。
安装依赖bash运行pip install aiofiles优化代码实现python运行import asyncio import aiohttp import aiofiles from pathlib import Path from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type SAVE_DIR Path(async_images) SAVE_DIR.mkdir(exist_okTrue) MAX_CONCURRENT_TASKS 20 semaphore asyncio.Semaphore(MAX_CONCURRENT_TASKS) failed_urls [] retry( stopstop_after_attempt(
, waitwait_exponential(multiplier1, min1, max
, retryretry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError, OSError)), reraiseTrue ) async def download_single_image(session: aiohttp.ClientSession, img_url: str, img_name: str): async with semaphore: try: async with session.get(img_url, timeout
as response: if response.status 200: img_data await response.read() # 异步写入文件 async with aiofiles.open(SAVE_DIR / img_name, wb) as f: await f.write(img_data) print(f成功下载{img_name}) else: raise Exception(fHTTP状态码异常{response.status}) except Exception as e: print(f下载失败重试后{str(e)}URL{img_url}) failed_urls.append(img_url) raise
核心优化点四HTTP 请求优化减少无效开销大量图片下载时HTTP 请求的细节优化可显著提升效率包括请求头复用、连接池配置、超时优化、压缩支持、SSL 验证控制等。
aiohttp.ClientSession 精细化配置aiohttp.ClientSession是异步 HTTP 会话通过配置连接池、请求头、超时参数可减少重复连接建立、优化请求效率。
python运行async def batch_download_images(img_urls: list): # 配置ClientSession timeout aiohttp.ClientTimeout(total60, connect10, sock_read
# 总超时60s连接超时10s读取超时20s connector aiohttp.TCPConnector( limitMAX_CONCURRENT_TASKS, # 连接池最大连接数与信号量一致 ttl_dns_cache300, # DNS缓存5分钟减少重复解析 sslFalse # 关闭SSL验证仅用于测试生产环境建议开启避免安全风险 ) # 通用请求头模拟浏览器避免反爬 headers { User-Agent: Mozilla/
0 (Windows NT
1
0; Win64; x
AppleWebKit/
5
36 (KHTML, like Gecko) Chrome/
120.
0.
0 Safari/
5
36, Accept: image/webp,image/apng,image/*,*/*;q
8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive } # 创建优化后的会话 async with aiohttp.ClientSession( connectorconnector, timeouttimeout, headersheaders ) as session: tasks [ download_single_image(session, url, fimage_{idx}.jpg) for idx, url in enumerate(img_urls) ] await asyncio.gather(*tasks) save_failed_urls()
流式下载大图片降低内存占用对于高清大图如 10MB 以上await response.read()会一次性将图片数据加载至内存大量并发时易导致内存溢出。
可通过流式读取分块写入文件减少内存占用。
python运行async def download_single_image(session: aiohttp.ClientSession, img_url: str, img_name: str): async with semaphore: try: async with session.get(img_url, timeout
as response: if response.status 200: async with aiofiles.open(SAVE_DIR / img_name, wb) as f: # 流式分块读取每块4KB async for chunk in response.content.iter_chunked(
: await f.write(chunk) print(f成功下载{img_name}) else: raise Exception(fHTTP状态码异常{response.status}) except Exception as e: print(f下载失败重试后{str(e)}URL{img_url}) failed_urls.append(img_url) raise
核心优化点五资源管理与性能监控大量图片下载任务耗时较长需完善资源释放逻辑同时监控下载进度、耗时、成功率便于问题排查和性能调优。
事件循环与资源安全释放Python
7 推荐使用asyncio.run()启动事件循环其会自动管理事件循环生命周期对于复杂场景需手动关闭会话、释放连接池避免资源泄漏。
进度监控与性能统计通过计数器、时间戳记录下载进度统计总耗时、成功数、失败数直观展示任务状态。
python运行import time import asyncio import aiohttp import aiofiles from pathlib import Path from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type SAVE_DIR Path(async_images) SAVE_DIR.mkdir(exist_okTrue) MAX_CONCURRENT_TASKS 20 semaphore asyncio.Semaphore(MAX_CONCURRENT_TASKS) # 全局计数器 success_count 0 failed_count 0 start_time time.time() retry( stopstop_after_attempt(
, waitwait_exponential(multiplier1, min1, max
, retryretry_if_exception_type((aiohttp.ClientError, asyncio.TimeoutError, OSError)), reraiseTrue ) async def download_single_image(session: aiohttp.ClientSession, img_url: str, img_name: str): global success_count, failed_count async with semaphore: try: async with session.get(img_url, timeout
as response: if response.status 200: async with aiofiles.open(SAVE_DIR / img_name, wb) as f: async for chunk in response.content.iter_chunked(
: await f.write(chunk) success_count 1 print(f[{success_count}/{success_countfailed_count}] 成功下载{img_name}) else: raise Exception(fHTTP状态码异常{response.status}) except Exception as e: failed_count 1 print(f[{success_count}/{success_countfailed_count}] 下载失败{str(e)}URL{img_url}) raise async def batch_download_images(img_urls: list): # 配置ClientSession同前文优化代码 timeout aiohttp.ClientTimeout(total60, connect10, sock_read
connector aiohttp.TCPConnector(limitMAX_CONCURRENT_TASKS, ttl_dns_cache300, sslFalse) headers { User-Agent: Mozilla/
0 (Windows NT
1
0; Win64; x
AppleWebKit/
5
36 (KHTML, like Gecko) Chrome/
120.
0.
0 Safari/
5
36, Accept: image/webp,image/apng,image/*,*/*;q
8, Accept-Encoding: gzip, deflate, br, Connection: keep-alive } async with aiohttp.ClientSession(connectorconnector, timeouttimeout, headersheaders) as session: tasks [ download_single_image(session, url, fimage_{idx}.jpg) for idx, url in enumerate(img_urls) ] await asyncio.gather(*tasks) # 统计最终结果 total_time time.time() - start_time print(f\n 下载任务完成 ) print(f总图片数{len(img_urls)}) print(f成功数{success_count}失败数{failed_count}) print(f成功率{success_count/len(img_urls)*100:.2f}%) print(f总耗时{total_time:.2f}秒) print(f平均下载速度{len(img_urls)/total_time:.2f}张/秒)
反爬规避与合规性优化大量图片异步下载时并发请求易触发网站反爬机制需在合规前提下优化请求行为避免 IP 封禁随机 User-Agent使用fake_useragent库生成随机浏览器请求头避免固定 UA 被识别随机延时在协程中加入await asyncio.sleep(random.uniform(
5,
)模拟人类操作间隔代理 IP 池对于高反爬网站结合aiohttp-socks使用代理 IP分散请求 IP遵守 robots.txt爬取前检查目标网站robots.txt禁止爬取的图片资源需规避请求频率限制除信号量外可通过asyncio.sleep控制单位时间内的请求数。
优化效果对比与
总结
优化前后性能对比以 1000 张图片下载为例优化维度基础异步实现全量优化后实现总耗时120 秒
秒下载成功率
%95%内存占用高一次性加载数据低流式分块读取反爬风险高无并发限制低并发 延时控制异常处理无重试失败率高自动重试失败记录
核心优化要点
总结并发控制通过Semaphore或分批次执行将并发数控制在合理范围平衡效率与服务器压力异常重试使用tenacity实现可恢复异常的自动重试结合失败 URL 持久化提升成功率异步 I/Oaiohttp实现异步 HTTP 请求aiofiles实现异步文件写入消除同步阻塞请求优化精细化配置ClientSession复用连接池、优化请求头、流式下载大文件监控与合规实时监控下载进度统计性能指标同时遵守反爬规则与合规要求。
基于asyncio的大量图片异步下载核心是平衡并发效率与资源占用通过多层级优化解决 I/O 阻塞、异常丢失、反爬封禁等问题可实现数千、数万张图片的高效稳定下载。
实际应用中需根据目标网站的反爬策略、网络环境、图片大小动态调整并发数、重试次数、请求间隔等参数达到最优下载效果。