# 錯(cuò)誤示例：阻塞事件循環(huán)
async def write_log_sync():
    with open('log.txt', 'a') as f:  # 同步阻塞！
        f.write('New log entry\n')   # 可能阻塞數(shù)毫秒
    await asyncio.sleep(0.1)

# 安裝：pip install aiofiles
import aiofiles

async def async_file_operations():
    # 異步寫(xiě)入
    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"文件內(nèi)容: {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()
    
    # 寫(xiě)入文件
    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)
    
    # 使用線程池執(zhí)行阻塞操作
    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):
    """異步復(fù)制大文件"""
    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)
                # 定期讓出控制權(quán)
                await asyncio.sleep(0)

async def process_multiple_files(file_paths):
    """并行處理多個(gè)文件"""
    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):
    """處理單個(gè)文件"""
    async with aiofiles.open(path) as f:
        content = await f.read()
        # 模擬處理過(guò)程
        await asyncio.sleep(0.1)
        return len(content)

async def download_and_save(url, file_path):
    """下載網(wǎng)絡(luò)內(nèi)容并保存到文件"""
    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):
    """控制文件操作并發(fā)數(shù)"""
    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):
    """批量寫(xiě)入日志（減少I(mǎi)O次數(shù)）"""
    async with aiofiles.open('app.log', 'a') as f:
        # 合并所有條目一次性寫(xiě)入
        batch_content = '\n'.join(entries) + '\n'
        await f.write(batch_content)

async def buffered_writing(file_path, data_generator, buffer_size=8192):
    """使用緩沖區(qū)寫(xiě)入數(shù)據(jù)流"""
    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)  # 讓出控制權(quán)
        # 寫(xiě)入剩余數(shù)據(jù)
        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錯(cuò)誤: {e}")
        raise

async def reliable_file_write(content, file_path, max_retries=3):
    """帶重試的文件寫(xiě)入"""
    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  # 指數(shù)退避
            await asyncio.sleep(delay)
    return False

import tempfile
import shutil

async def process_with_temp_file():
    """使用臨時(shí)文件處理數(shù)據(jù)"""
    with tempfile.NamedTemporaryFile(delete=False) as tmp:
        temp_path = tmp.name
    
    try:
        # 異步寫(xiě)入臨時(shí)文件
        async with aiofiles.open(temp_path, 'w') as f:
            await f.write("臨時(shí)數(shù)據(jù)")
        
        # 處理數(shù)據(jù)
        await process_data(temp_path)
        
        # 移動(dòng)最終文件
        shutil.move(temp_path, 'final.txt')
    finally:
        if os.path.exists(temp_path):
            os.unlink(temp_path)

import watchfiles

async def monitor_directory(path):
    """監(jiān)控目錄變化（異步迭代器）"""
    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():
    """文件操作性能對(duì)比測(cè)試"""
    test_data = 'test' * 1000000  # 4MB數(shù)據(jù)
    
    # 同步寫(xiě)入
    start = time.time()
    with open('sync.txt', 'w') as f:
        f.write(test_data)
    sync_write_time = time.time() - start
    
    # 異步寫(xiě)入（aiofiles）
    start = time.time()
    async with aiofiles.open('async.txt', 'w') as f:
        await f.write(test_data)
    async_write_time = time.time() - start
    
    # 線程池寫(xiě)入
    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"同步寫(xiě)入耗時(shí): {sync_write_time:.4f}s")
    print(f"異步寫(xiě)入耗時(shí): {async_write_time:.4f}s")
    print(f"線程池寫(xiě)入耗時(shí): {thread_write_time:.4f}s")

# 運(yùn)行測(cè)試
asyncio.run(test_performance())

同步寫(xiě)入耗時(shí): 0.0254s
異步寫(xiě)入耗時(shí): 0.0261s
線程池寫(xiě)入耗時(shí): 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):
        """啟動(dòng)定期刷新任務(wù)"""
        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):
        """添加日志到緩沖區(qū)"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        self.buffer.append(f"[{timestamp}] {message}\n")
        
        # 緩沖區(qū)滿時(shí)立即刷新
        if len(self.buffer) >= self.max_buffer:
            await self.flush_buffer()
    
    async def auto_flush(self):
        """定期刷新緩沖區(qū)"""
        while self.running:
            await asyncio.sleep(self.flush_interval)
            await self.flush_buffer()
    
    async def flush_buffer(self):
        """將緩沖區(qū)內(nèi)容寫(xiě)入文件"""
        if not self.buffer:
            return
        
        # 合并日志條目
        log_lines = ''.join(self.buffer)
        self.buffer.clear()
        
        # 異步寫(xiě)入文件
        try:
            async with aiofiles.open(self.file_path, 'a') as f:
                await f.write(log_lines)
        except IOError as e:
            print(f"日志寫(xiě)入失敗: {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())