Python协程中的文件操作完全指南

异步读写文件的正确姿势与性能优化

在Python协程中执行文件操作是常见的需求，但直接使用同步文件读写会阻塞事件循环，破坏异步并发优势。本文将深入解析协程环境下文件操作的正确方法，涵盖多种场景下的最佳实践和性能优化技巧。

---

一、核心原则：避免阻塞事件循环

1.1 为什么不能直接使用同步IO？

# 错误示例：阻塞事件循环
async def write_log_sync():
    with open('log.txt', 'a') as f:  # 同步阻塞！
        f.write('New log entry\n')   # 可能阻塞数毫秒
    await asyncio.sleep(0.1)

问题分析：

• 文件操作是磁盘IO，属于阻塞操作
• 阻塞期间事件循环无法执行其他任务
• 高并发场景下会导致性能急剧下降

---

二、正确方法：异步文件操作方案

2.1 使用aiofiles库（推荐）

# 安装：pip install aiofiles
import aiofiles

async def async_file_operations():
    # 异步写入
    async with aiofiles.open('data.txt', 'w') as f:
        await f.write('Hello, async world!\n')
        await f.write('Another line\n')
    
    # 异步读取
    async with aiofiles.open('data.txt', 'r') as f:
        content = await f.read()
        print(f"文件内容: {content}")
    
    # 逐行读取
    async with aiofiles.open('large_file.txt') as f:
        async for line in f:
            process_line(line)

优势：

• 原生异步API设计
• 支持上下文管理器
• 行为与内置open函数一致
• 底层使用线程池自动处理阻塞操作

2.2 使用线程池执行同步操作

import asyncio

async def threadpool_file_io():
    loop = asyncio.get_running_loop()
    
    # 写入文件
    def write_file():
        with open('log.txt', 'a') as f:
            f.write('Log entry\n')
    
    # 读取文件
    def read_file():
        with open('config.json') as f:
            return json.load(f)
    
    # 使用线程池执行阻塞操作
    await loop.run_in_executor(None, write_file)
    config = await loop.run_in_executor(None, read_file)
    return config

适用场景：

• 无法安装第三方库的环境
• 需要精细控制线程池资源
• 混合执行多种阻塞操作

---

三、高级文件操作技巧

3.1 大文件分块读写

async def copy_large_file(src, dst, chunk_size=1024 * 1024):
    """异步复制大文件"""
    async with aiofiles.open(src, 'rb') as src_file:
        async with aiofiles.open(dst, 'wb') as dst_file:
            while True:
                chunk = await src_file.read(chunk_size)
                if not chunk:
                    break
                await dst_file.write(chunk)
                # 定期让出控制权
                await asyncio.sleep(0)

3.2 并行处理多个文件

async def process_multiple_files(file_paths):
    """并行处理多个文件"""
    tasks = []
    for path in file_paths:
        task = asyncio.create_task(process_single_file(path))
        tasks.append(task)
    
    results = await asyncio.gather(*tasks)
    return results

async def process_single_file(path):
    """处理单个文件"""
    async with aiofiles.open(path) as f:
        content = await f.read()
        # 模拟处理过程
        await asyncio.sleep(0.1)
        return len(content)

3.3 文件操作与网络请求结合

async def download_and_save(url, file_path):
    """下载网络内容并保存到文件"""
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            content = await response.read()
    
    async with aiofiles.open(file_path, 'wb') as f:
        await f.write(content)
    
    return file_path

---

四、性能优化策略

4.1 控制并发文件操作数量

async def controlled_file_operations(file_paths, max_concurrent=5):
    """控制文件操作并发数"""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_with_limit(path):
        async with semaphore:
            return await process_single_file(path)
    
    tasks = [process_with_limit(path) for path in file_paths]
    return await asyncio.gather(*tasks)

4.2 批量写入优化

async def batch_write_logs(entries):
    """批量写入日志（减少IO次数）"""
    async with aiofiles.open('app.log', 'a') as f:
        # 合并所有条目一次性写入
        batch_content = '\n'.join(entries) + '\n'
        await f.write(batch_content)

4.3 使用内存缓冲区

async def buffered_writing(file_path, data_generator, buffer_size=8192):
    """使用缓冲区写入数据流"""
    buffer = bytearray()
    async with aiofiles.open(file_path, 'wb') as f:
        async for data_chunk in data_generator:
            buffer.extend(data_chunk)
            if len(buffer) >= buffer_size:
                await f.write(buffer)
                buffer.clear()
                await asyncio.sleep(0)  # 让出控制权
        # 写入剩余数据
        if buffer:
            await f.write(buffer)

---

五、错误处理与恢复

5.1 健壮的文件操作

async def safe_file_operation(file_path):
    try:
        async with aiofiles.open(file_path) as f:
            return await f.read()
    except FileNotFoundError:
        print(f"文件不存在: {file_path}")
        return None
    except IOError as e:
        print(f"IO错误: {e}")
        raise

5.2 带重试机制的操作

async def reliable_file_write(content, file_path, max_retries=3):
    """带重试的文件写入"""
    for attempt in range(max_retries):
        try:
            async with aiofiles.open(file_path, 'w') as f:
                await f.write(content)
            return True
        except IOError as e:
            if attempt == max_retries - 1:
                raise
            delay = 2 ** attempt  # 指数退避
            await asyncio.sleep(delay)
    return False

---

六、特殊场景处理

6.1 临时文件处理

import tempfile
import shutil

async def process_with_temp_file():
    """使用临时文件处理数据"""
    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        temp_path = tmp.name
    
    try:
        # 异步写入临时文件
        async with aiofiles.open(temp_path, 'w') as f:
            await f.write("临时数据")
        
        # 处理数据
        await process_data(temp_path)
        
        # 移动最终文件
        shutil.move(temp_path, 'final.txt')
    finally:
        if os.path.exists(temp_path):
            os.unlink(temp_path)

6.2 文件系统监控

import watchfiles

async def monitor_directory(path):
    """监控目录变化（异步迭代器）"""
    async for changes in watchfiles.awatch(path):
        for change_type, file_path in changes:
            if change_type == watchfiles.Change.added:
                print(f"新文件: {file_path}")
                await process_new_file(file_path)

---

七、性能对比测试

import time
import asyncio
import aiofiles

async def test_performance():
    """文件操作性能对比测试"""
    test_data = 'test' * 1000000  # 4MB数据
    
    # 同步写入
    start = time.time()
    with open('sync.txt', 'w') as f:
        f.write(test_data)
    sync_write_time = time.time() - start
    
    # 异步写入（aiofiles）
    start = time.time()
    async with aiofiles.open('async.txt', 'w') as f:
        await f.write(test_data)
    async_write_time = time.time() - start
    
    # 线程池写入
    start = time.time()
    loop = asyncio.get_running_loop()
    def write_sync():
        with open('thread.txt', 'w') as f:
            f.write(test_data)
    await loop.run_in_executor(None, write_sync)
    thread_write_time = time.time() - start
    
    print(f"同步写入耗时: {sync_write_time:.4f}s")
    print(f"异步写入耗时: {async_write_time:.4f}s")
    print(f"线程池写入耗时: {thread_write_time:.4f}s")

# 运行测试
asyncio.run(test_performance())

典型结果：

同步写入耗时: 0.0254s
异步写入耗时: 0.0261s
线程池写入耗时: 0.0287s

结论：

• 单次文件操作：同步最快（无额外开销）
• 高并发场景：异步/线程池避免阻塞，整体吞吐量更高

---

八、最佳实践总结

1. 首选aiofiles：简单直接的异步文件API
2. 大文件分块处理：避免内存溢出，定期让出控制权
3. 控制并发数：使用信号量限制同时打开的文件数
4. 批量操作优化：减少IO次数提升性能
5. 错误处理：添加重试机制和异常捕获
6. 混合操作：结合线程池处理特殊场景
7. 资源清理：确保文件正确关闭，使用上下文管理器

---

完整示例：异步日志系统

import aiofiles
import asyncio
import time
from collections import deque

class AsyncLogger:
    def __init__(self, file_path, max_buffer=100, flush_interval=5):
        self.file_path = file_path
        self.buffer = deque()
        self.max_buffer = max_buffer
        self.flush_interval = flush_interval
        self.flush_task = None
        self.running = True
    
    async def start(self):
        """启动定期刷新任务"""
        self.flush_task = asyncio.create_task(self.auto_flush())
    
    async def stop(self):
        """停止日志记录器"""
        self.running = False
        if self.flush_task:
            self.flush_task.cancel()
        await self.flush_buffer()
    
    async def log(self, message):
        """添加日志到缓冲区"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        self.buffer.append(f"[{timestamp}] {message}\n")
        
        # 缓冲区满时立即刷新
        if len(self.buffer) >= self.max_buffer:
            await self.flush_buffer()
    
    async def auto_flush(self):
        """定期刷新缓冲区"""
        while self.running:
            await asyncio.sleep(self.flush_interval)
            await self.flush_buffer()
    
    async def flush_buffer(self):
        """将缓冲区内容写入文件"""
        if not self.buffer:
            return
        
        # 合并日志条目
        log_lines = ''.join(self.buffer)
        self.buffer.clear()
        
        # 异步写入文件
        try:
            async with aiofiles.open(self.file_path, 'a') as f:
                await f.write(log_lines)
        except IOError as e:
            print(f"日志写入失败: {e}")

# 使用示例
async def main():
    logger = AsyncLogger('app.log')
    await logger.start()
    
    # 模拟日志记录
    for i in range(1, 101):
        await logger.log(f"Processing item {i}")
        await asyncio.sleep(0.1)
    
    await logger.stop()

asyncio.run(main())

---