Python协程进阶：asyncio实现高并发文件处理

Python版本： 3.11+

核心模块： asyncio、aiofiles、asynccontextmanager
适用场景： 大量文件IO操作、日志处理、数据批量导入导出

asyncio 是 Python 3.4 引入的异步 IO 框架，通过单线程事件循环实现高并发。对于文件 IO 密集型任务，asyncio 配合 aiofiles 可以显著提升处理速度。

核心概念

事件循环（Event Loop）

事件循环是 asyncio 的核心，负责调度协程的执行。每个线程有且仅有一个事件循环。

import asyncio

async def main():
    print("开始")
    await asyncio.sleep(1)
    print("结束")

asyncio.run(main())

协程（Coroutine）

使用 async 定义的函数称为协程，不能直接调用，需要 await 或 asyncio.create_task()。

Task 与 Future

• Task：包装协程的对象，可以并发执行多个协程

• Future：表示未来结果的底层对象

文件并发处理实战

基础示例

import asyncio
import aiofiles
import os
from pathlib import Path

async def process_file(filepath: str) -> dict:
    """异步处理单个文件"""
    async with aiofiles.open(filepath, 'r', encoding='utf-8') as f:
        content = await f.read()

    # 模拟耗时处理
    await asyncio.sleep(0.1)

    return {
        'path': filepath,
        'lines': len(content.split('\n')),
        'chars': len(content)
    }

async def batch_process(directory: str, max_concurrency: int = 10):
    """批量处理目录下的所有文件"""
    paths = list(Path(directory).glob('*.txt'))

    # 创建信号量控制并发数
    semaphore = asyncio.Semaphore(max_concurrency)

    async def bounded_process(path):
        async with semaphore:
            return await process_file(str(path))

    # 并发执行
    tasks = [asyncio.create_task(bounded_process(p)) for p in paths]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 过滤异常
    success = [r for r in results if isinstance(r, dict)]
    errors = [r for r in results if isinstance(r, BaseException)]

    return success, errors

# 使用示例
if __name__ == '__main__':
    results, errors = asyncio.run(batch_process('./data'))
    print(f"成功: {len(results)}, 失败: {len(errors)}")

性能对比

异步上下文管理

from contextlib import asynccontextmanager

@asynccontextmanager
async def managed_resource(resource_id: str):
    """异步资源管理器"""
    print(f"获取资源: {resource_id}")
    try:
        yield f"Resource-{resource_id}"
    finally:
        print(f"释放资源: {resource_id}")

async def main():
    async with managed_resource("db-connection") as conn:
        print(f"使用连接: {conn}")
    print("资源已释放")

踩坑记录

1. 不要在 async 函数中使用同步阻塞代码，否则阻塞整个事件循环

2. aiofiles 需要先安装：pip install aiofiles

3. 并发数不宜过高，建议根据 CPU 核数和 IO 类型调整（一般 10-50）

4. 异常处理要完善，使用 asyncio.gather(return_exceptions=True) 避免单个失败导致全部中断