Python异步文件操作利器：aiofiles深度解析与实战指南

# 稳定版本安装
pip install aiofiles

# 安装开发版本（可选）
pip install git+https://github.com/Tinche/aiofiles.git

import asyncio
import aiofiles

async def read_file_async(file_path):
    async with aiofiles.open(file_path, mode='r', encoding='utf-8') as f:
        content = await f.read()  # 异步读取文件全部内容
        print(f"文件内容：\n{content}")

# 运行异步函数
asyncio.run(read_file_async("example.txt"))

async def write_file_async(file_path, content):
    async with aiofiles.open(file_path, mode='w', encoding='utf-8') as f:
        await f.write(content)  # 异步写入内容
        await f.flush()  # 手动刷新缓冲区（可选，关闭文件时会自动刷新）
        print("文件写入完成")

asyncio.run(write_file_async("output.txt", "Hello, aiofiles!"))

async def read_lines_async(file_path):
    async with aiofiles.open(file_path, mode='r', encoding='utf-8') as f:
        async for line in f:  # 异步迭代文件对象，逐行读取
            print(f"行内容：{line.strip()}")

asyncio.run(read_lines_async("access.log"))

async def write_lines_async(file_path, lines):
    async with aiofiles.open(file_path, mode='w', encoding='utf-8') as f:
        for line in lines:
            await f.write(line + "\n")  # 逐行写入并添加换行符
    print("多行写入完成")

asyncio.run(write_lines_async("lines.txt", ["Line 1", "Line 2", "Line 3"]))

async def read_multiple_files(file_paths):
    tasks = [read_file_async(path) for path in file_paths]  # 创建任务列表
    await asyncio.gather(*tasks)  # 并发执行所有任务

asyncio.run(read_multiple_files(["file1.txt", "file2.txt", "file3.txt"]))

async def seek_and_read(file_path, offset):
    async with aiofiles.open(file_path, mode='r', encoding='utf-8') as f:
        await f.seek(offset)  # 异步移动文件指针到指定位置
        content = await f.read(100)  # 读取后续100字节内容
        print(f"从位置{offset}开始的内容：{content}")

asyncio.run(seek_and_read("large_file.txt", 500))

import aiohttp
import aiofiles

async def download_and_save(url, save_path):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            content = await response.read()  # 异步获取网络响应内容

    async with aiofiles.open(save_path, mode='wb') as f:
        await f.write(content)  # 异步保存到文件
    print(f"文件已保存至：{save_path}")

# 运行示例
image_url = "https://example.com/image.jpg"
asyncio.run(download_and_save(image_url, "downloaded_image.jpg"))

import asyncio
import aiofiles
from datetime import datetime

class AsyncLogger:
    def __init__(self, log_file="app.log"):
        self.log_file = log_file

    async def log(self, message, level="INFO"):
        """异步写入日志条目"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        log_line = f"[{timestamp}] [{level}] {message}\n"
        async with aiofiles.open(self.log_file, mode='a', encoding='utf-8') as f:
            await f.write(log_line)  # 异步追加日志

    async def log_error(self, message):
        """异步写入错误日志"""
        await self.log(message, level="ERROR")

# 模拟异步业务逻辑
async def handle_request(logger, request_id):
    await logger.log(f"处理请求 {request_id}")
    # 模拟耗时操作
    await asyncio.sleep(0.1)
    await logger.log_error(f"请求 {request_id} 处理失败")

# 主程序：并发处理多个请求并记录日志
async def main():
    logger = AsyncLogger()
    tasks = [handle_request(logger, i) for i in range(10)]  # 模拟10个并发请求
    await asyncio.gather(*tasks)

if __name__ == "__main__":
    asyncio.run(main())
    print("所有请求处理完成")

  [2025-06-04 14:30:00] [INFO] 处理请求 1
  [2025-06-04 14:30:00] [INFO] 处理请求 2
  ...
  [2025-06-04 14:30:01] [ERROR] 请求 1 处理失败
  [2025-06-04 14:30:01] [ERROR] 请求 2 处理失败