Python 的 async 异步编程主要基于协程与 `asyncio` 库，能有效提升程序在 I/O 密集型任务中的性能。 ### 异步迭代器与异步可迭代对象 可在 `async for` 语句中使用的对象，需通过其 `__aiter__()` 方法返回一个异步迭代器，这是由 PEP 492 引入的特性。以下是自定义异步迭代器和异步可迭代对象的示例代码： ```python import asyncio class Reader(object): """ 自定义异步迭代器（同时也是异步可迭代对象） """ def __init__(self): self.count = 0 async def readline(self): # await asyncio.sleep(1) self.count += 1 if self.count == 100: return None return self.count def __aiter__(self): return self async def __anext__(self): val = await self.readline() if val == None: raise StopAsyncIteration return val async def func(): # 创建异步可迭代对象 async_iter = Reader() # async for 必须要放在async def函数内，否则语法错误。 async for item in async_iter: print(item) asyncio.run(func()) ``` 上述代码中，`Reader` 类实现了异步迭代器和异步可迭代对象的接口，`func` 函数中使用 `async for` 遍历异步可迭代对象 [^1][^2]。 ### 异步函数与 `async/await` `async/await` 是 Python 异步编程的核心语法，用于定义异步函数和等待异步操作完成。以下是一个使用 `async/await` 重构的示例代码，用于文件的读取、处理和保存： ```python import asyncio async def read_file(file_path): print("Reading file...") await asyncio.sleep(1) # 模拟 I/O with open(file_path, 'r') as f: return f.read() async def process_data(data): print("Processing data...") await asyncio.sleep(1) # 模拟处理 return data.upper() async def save_file(data, file_path): print("Saving file...") await asyncio.sleep(1) # 模拟写入 with open(file_path, 'w') as f: f.write(data) return "Save successful" async def main(): data = await read_file("input.txt") processed = await process_data(data) msg = await save_file(processed, "output.txt") print(msg) asyncio.run(main()) ``` 使用 `async/await` 有诸多优点，如语法清晰，代码按顺序执行，逐步等待任务完成；代码扁平化，没有多层嵌套，逻辑更加直观；异常处理简单，可使用标准的 `try...except` 结构处理错误 [^3]。