asyncio.gather本身不解决文件I/O瓶颈，必须配合线程池（run_in_executor）和合理分片才能实现高效并发读写，否则会退化为串行。
asyncio.gather 本身不解决文件 I/O 瓶颈
直接用 asyncio.gather 包裹一堆 open() 或 os.read() 调用，性能不会比同步快——因为 Python 的标准文件对象是阻塞的，协程会在那里卡住，实际退化成串行。
真正可行的路径只有一条：把文件操作交给线程池（loop.run_in_executor），让协程调度不被阻塞。
- 不要写
await asyncio.gather(*[read_file(f) for f in files])，其中read_file内部用open().read() - 必须把每个文件读写封装成可在线程中执行的函数，再丢给
run_in_executor - 分片不是为了“并发数可控”，而是防止线程池过载或内存爆掉（比如一次读 1000 个 100MB 文件）
分片 + run_in_executor 的最小可靠写法
假设你要并发读取一批日志文件并统计行数，以下结构能稳住：
import asyncio
from concurrent.futures import ThreadPoolExecutor
<p>def sync_read_lines(path):
with open(path, "r", encoding="utf-8") as f:
return len(f.readlines())</p><p>async def async_read_lines(path, executor):
loop = asyncio.get_running_loop()
return await loop.run_in_executor(executor, sync_read_lines, path)</p><p>async def main():
files = ["a.log", "b.log", "c.log"]</p><h1>分片大小设为 50 是经验值：兼顾吞吐和内存压力</h1><pre class="brush:php;toolbar:false;">chunk_size = 50
results = []
for i in range(0, len(files), chunk_size):
chunk = files[i:i+chunk_size]
# 每批单独 gather，避免单次传入太多 future 压垮事件循环
chunk_tasks = [async_read_lines(f, executor) for f in chunk]
results.extend(await asyncio.gather(*chunk_tasks))
return results
必须显式创建 executor，且复用（不能在循环里 new）
executor = ThreadPoolExecutor(max_workers=8) asyncio.run(main())
-
max_workers不建议超过 CPU 核心数 × 2，纯 I/O 场景 4–8 更稳妥 - 分片大小取决于单文件平均体积和总内存预算，不是越大越好
- 别在
async_read_lines里做解析逻辑（如正则匹配），那会延长线程占用时间
写文件时要注意原子性与竞争
并发写同一目录下的不同文件一般没问题，但一旦涉及追加（"a"）、覆盖（"w"）或写后立即读，就容易踩坑。
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
- 多个协程同时
open(..., "w")同一个路径？后启动的会清空先写的——没锁，没事务 - 用
"x"模式（独占创建）可防误覆盖，但失败会抛FileExistsError，得捕获 - 如果目标是拼接内容，别用
"a"——POSIX 下它不保证多进程/线程安全，Linux 可能丢数据 - 更稳妥的做法：每个任务写临时文件（
tempfile.NamedTemporaryFile），最后统一 rename
asyncio.gather 返回顺序严格对应输入顺序
这点常被忽略：即使第 3 个文件读得最慢，它的结果仍会排在返回列表的索引 2 上。这方便你和原始文件列表对齐，但也意味着不能靠“谁先完成谁先处理”来优化流式场景。
- 如果你需要“哪个先完成就先处理”，改用
asyncio.as_completed+await迭代 -
gather抛异常时默认全停（return_exceptions=False），想容错得设return_exceptions=True，然后手动检查每个结果是不是Exception实例 - 大量文件下，
gather构造的 future 列表本身有内存开销，分片也缓解这个压力
真正的难点不在怎么调 gather，而在于判断哪些操作真能并发、哪些只是假并发，以及线程池参数和分片策略怎么贴合你的磁盘类型（NVMe vs 机械盘）、文件分布（小文件密集 or 大文件稀疏）和内存上限。
Python免费学习笔记（深入）：立即使用
在学习笔记中，你将探索 Python 的核心概念和高级技巧！
