async def read_log_lines() 必须用 async for 配合 aiofiles.open 等异步文件对象，不可用同步 for line in open(...)；逐行读需 rstrip('\n')，大文件或无换行日志应改用分块读+手动缓冲；所有处理逻辑必须 await，CPU 密集操作需 asyncio.to_thread。
async def read_log_lines() 必须用 async for 而不是 for
同步的 for line in open(...) 在异步上下文中会阻塞整个事件循环，直接导致并发退化为串行。Python 3.12 的 asyncio 不允许在协程中调用阻塞 I/O，所以不能把普通文件对象丢进 async for。
正确做法是用支持异步读取的类或手动分块 + await loop.run_in_executor。但更轻量、更符合流式意图的是用 aiofiles（需 pip install aiofiles）：
import aiofiles
<p>async def read_log_lines(filepath):
async with aiofiles.open(filepath, mode='r', encoding='utf-8') as f:
async for line in f:
yield line.rstrip('\n')
-
aiofiles.open返回的是真正支持async for的异步迭代器，底层基于线程池非阻塞调度 - 别漏掉
rstrip('\n')——aiofiles返回的line包含换行符，和内置open行为一致 - 如果日志含 GB 级别且编码不统一（比如混合 GBK/UTF-8），
aiofiles默认不处理编码错误；建议加errors='ignore'或errors='replace'
yield 之后必须 await 消费者处理，否则会压爆内存
异步生成器本身不控制消费速度。如果你写成这样：
async for line in read_log_lines('access.log'):
process_line(line) # ❌ 没 await，纯同步函数，不暂停生成器
那么 read_log_lines() 会一路狂吐，直到缓冲区撑满或 OOM。关键点在于：异步生成器的 yield 只让出控制权给事件循环，不保证消费者已处理完上一条。
立即学习“Python免费学习笔记（深入）”；
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
- 所有实际处理逻辑（解析、过滤、入库）必须是
awaitable，比如await insert_to_db(line)或await asyncio.to_thread(parse_json, line) - 若处理逻辑 CPU 密集（如正则匹配大量字段），务必用
await asyncio.to_thread()包裹，避免阻塞事件循环 - 可加简单背压：在生成器内部用
await asyncio.sleep(0)让出一次调度权，但这只是“礼貌性暂停”，不能替代真实 await 消费
大文件分块读取比逐行更稳，尤其当日志无换行或超长行时
有些日志（如 minified JSON 日志、二进制嵌入日志）没有可靠换行符，或单行长达数 MB。此时 async for line in f 可能卡住或 OOM。稳妥做法是放弃“按行”假设，改用固定大小块读取 + 手动缓冲拆分。
示例：每次读 64KB，维护未完成行缓冲区：
async def read_log_chunks(filepath, chunk_size=65536):
buffer = b''
async with aiofiles.open(filepath, mode='rb') as f:
while True:
chunk = await f.read(chunk_size)
if not chunk:
if buffer:
yield buffer.decode('utf-8', errors='ignore')
break
buffer += chunk
lines = buffer.split(b'\n')
buffer = lines[-1] # 保留不完整行
for line in lines[:-1]:
yield line.decode('utf-8', errors='ignore')
- 用
mode='rb'避免文本模式下编码异常中断读取 -
buffer是字节对象，拆分必须用b'\n'，不能用'\n' - 解码放在最后一步，且显式指定
errors，防止某一行损坏导致整个流中断
asyncio.StreamReader 不适合直接读文件，别被名字误导
看到 StreamReader 容易想当然拿来读磁盘文件，但它设计目标是网络 socket 流（TCP/UDP），依赖底层 transport 的 feed_data() 推送机制。文件 I/O 没有这种推送能力，强行套用会导致 RuntimeError: Stream is not readable 或永远挂起。
-
StreamReader适用于asyncio.open_connection或subprocessstdout 场景，不适用本地文件 - 需要类似行为（如带缓冲、支持
.readuntil()）？可用aiofiles+io.BytesIO封装，但通常没必要 —— 分块读 + 手动缓冲更可控 - Python 3.12 新增的
asyncio.to_thread()是更通用的兜底方案：对老代码或复杂解析逻辑，直接扔进线程池，比折腾流式抽象更省心
真正难的不是“怎么启动异步读”，而是确保每条日志从磁盘到处理完成的全链路都保持异步呼吸感——中间任何一个 await 缺失或阻塞调用，都会让流式变成假流式。缓冲区大小、编码容错、错误行跳过策略，这些细节比语法更决定成败。
