在 Python 中创建异步程序主要依赖 asyncio 库（Python 3.4+ 内置），以下是详细步骤和示例：
1. 核心概念
- 协程（Coroutine）：通过
async def定义的函数，是异步任务的基本单元。 - 事件循环（Event Loop）：管理协程的执行和调度。
await：挂起当前协程，等待另一个协程或异步操作完成。
2. 基本步骤
(1) 定义异步函数
import asyncio
async def task1():
print("开始 task1")
await asyncio.sleep(2) # 模拟耗时操作（非阻塞）
print("结束 task1")
async def task2():
print("开始 task2")
await asyncio.sleep(1)
print("结束 task2")
(2) 创建事件循环并运行任务
async def main():
# 并行运行多个任务
await asyncio.gather(task1(), task2())
# 启动事件循环
asyncio.run(main())
输出：
开始 task1
开始 task2
结束 task2 # task2 先完成（因为 sleep 更短）
结束 task1
3. 关键方法
asyncio.run(coro)：启动事件循环（Python 3.7+ 推荐）。await coro：等待单个协程完成。asyncio.gather(*coros)：并行运行多个协程。asyncio.create_task(coro)：将协程包装为任务并行执行（需在事件循环中）。
4. 实际应用示例
异步HTTP请求（需安装 aiohttp）
import aiohttp
import asyncio
async def fetch(url):
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
return await response.text()
async def main():
html = await fetch("https://example.com")
print(html[:200]) # 打印前200字符
asyncio.run(main())
5. 注意事项
- 阻塞代码会破坏异步：避免在协程中使用
time.sleep()或同步 IO 操作，改用await asyncio.sleep()或异步库（如aiofiles）。 - 错误处理：用
try/except捕获协程中的异常。 - Python 版本：确保使用 Python 3.7+（推荐 3.10+）。
6. 进阶技巧
- 任务控制：使用
asyncio.wait()或asyncio.as_completed()管理任务。 - 超时设置：
await asyncio.wait_for(coro, timeout=5)。 - 底层事件循环：通过
loop = asyncio.get_event_loop()自定义配置（高级场景）。
通过以上方法，你可以高效地编写异步程序，提升 IO 密集型应用的性能。
