AIOFile批量写入并发问题：输出文件数据损坏

问题根源分析 #

你遇到的写入覆盖、行截断问题，核心原因是多次小批量异步写入时，文件指针维护或缓冲区同步出现异常，尽管加了锁，但频繁的write+fsync组合可能触发AIOFile的未预期行为；单次写入一行的模式也放大了并发写入的风险点。

具体解决步骤 #

1. 合并写入操作，减少锁竞争与IO次数 #

将每个批次的所有内容先收集成完整字符串，再一次性写入文件，避免多次小写入导致的指针偏移问题：

async def write_batches(question_writer, answer_writer, results):
    q_content = []
    a_content = []
    for my_result in results:
        for my_record in my_result:
            q_content.append(f"{my_record['question']}\n")
            a_content.append(f"{my_record['answer']}\n")
    # 一次性写入整个批次的内容
    await write_and_flush(question_writer, ''.join(q_content), q_lock)
    await write_and_flush(answer_writer, ''.join(a_content), a_lock)

2. 替换AIOFile为标准库异步文件API #

AIOFile第三方库可能存在异步写入的指针维护bug，改用Python标准库的asyncio.open（3.7+支持），稳定性更有保障：

# 替换原文件打开代码
async with asyncio.open("question_output.txt", "w") as q_writer, asyncio.open(
    "answer_output.txt", "w"
) as a_writer:
    batch_size = 10
    for i in range(0, len(df), batch_size):
        batch_rows = df.iloc[i : i + batch_size].itertuples(index=False)
        batch_results = await process_batch(batch_rows)
        await write_batches(q_writer, a_writer, batch_results)

3. 调整flush策略，避免过度fsync #

fsync是强制同步到磁盘的操作，频繁调用会影响性能且可能干扰文件指针。改用flush刷新内核缓冲区即可，除非你有强持久化需求：

async def write_and_flush(writer, text, lock):
    async with lock:
        await writer.write(text)
        await writer.flush()  # 替换fsync为flush，减少磁盘同步操作

4. 验证文件指针的一致性 #

如果必须使用AIOFile，可在每次写入前显式获取并打印文件指针位置，排查是否存在指针异常重置：

async def write_and_flush(writer, text, lock):
    async with lock:
        pos = await writer.tell()
        print(f"Writing to position: {pos}")
        await writer.write(text)
        await writer.flush()

5. 确保锁的全局唯一性 #

确认你的锁对象是全局唯一的（当前代码中q_lock和a_lock在process_data内定义，是正确的），避免不同批次使用不同锁导致的竞态。

内容的提问来源于stack exchange，提问作者devinbost