Python基础入门 Day123 异步文件操作：使用 aiofiles 实现高性能文件读写

6次阅读

没有评论

共计 1842 个字符，预计需要花费 5 分钟才能阅读完成。

在前面的章节中，我们学习了网络异步请求、任务系统以及异步协作模型。本篇继续拓展 异步 I/O 的应用场景 —— 文件处理。

在传统同步方式中，文件读写属于 I/O 密集操作，会阻塞主线程。
而使用 aiofiles 可以让我们以异步方式进行文件读写，提升并发性能。

一、为什么需要异步文件操作？

常见场景：

需要同时读取或写入大量文件
服务器端批量处理日志
异步任务过程中持续保存数据
大型爬虫系统同时下载与写入文件

同步写文件会阻塞当前任务，而异步文件写入能够让程序 不中断地继续处理其他任务。

二、安装 aiofiles

pip install aiofiles

三、异步写入文件示例

import asyncio
import aiofiles

async def write_file():
    async with aiofiles.open("demo.txt", "w") as f:
        await f.write(" 这是异步写入的内容。\n")

asyncio.run(write_file())

说明：

async with 负责管理文件上下文
f.write() 变成异步调用
不会阻塞事件循环

四、异步读取文件示例

import asyncio
import aiofiles

async def read_file():
    async with aiofiles.open("demo.txt", "r") as f:
        content = await f.read()
        print(content)

asyncio.run(read_file())

适用于：

异步加载配置
批量读取文档
流式输出日志

五、逐行读取（适合大文件）

import asyncio
import aiofiles

async def read_lines():
    async with aiofiles.open("bigfile.txt", "r") as f:
        async for line in f:
            print(" 读取：", line.strip())

asyncio.run(read_lines())

优势：

不需要一次加载全部内容
内存占用极低
非常适合大文件（日志、数据集等）

六、并发写入多个文件（结合 create_task）

import asyncio
import aiofiles

async def write_to_file(i):
    filename = f"log_{i}.txt"
    async with aiofiles.open(filename, "w") as f:
        await f.write(f" 这是文件 {i} 的内容 ")

async def main():
    tasks = [asyncio.create_task(write_to_file(i)) for i in range(5)]
    await asyncio.gather(*tasks)

asyncio.run(main())

结果：

五个文件同时并发写入
不会阻塞主线程
非常适合批量生成数据文件

七、结合队列实现异步写日志系统（实战）

import asyncio
import aiofiles

async def log_writer(queue):
    async with aiofiles.open("system.log", "a") as f:
        while True:
            msg = await queue.get()
            if msg is None:
                break
            await f.write(msg + "\n")
            queue.task_done()

创建生产日志的任务：

async def produce_logs(queue):
    for i in range(10):
        await queue.put(f"[INFO] 日志消息 {i}")
        await asyncio.sleep(0.2)
    await queue.put(None)

整合：

async def main():
    queue = asyncio.Queue()
    writer = asyncio.create_task(log_writer(queue))
    producer = asyncio.create_task(produce_logs(queue))

    await producer
    await queue.join()
    await writer

asyncio.run(main())

效果：