Python协程环境下文件操作的正确方法

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

# 安装：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)

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

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)

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)

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

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)

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)

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)

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

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

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)

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

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())