Python异步编程通过协程和事件循环机制，能够在单线程内实现多个任务的并发执行，尤其适合处理网络请求、文件读写等IO密集型场景，避免程序因等待IO操作而阻塞，大幅提升资源利用率。
异步编程核心概念
协程
协程是异步编程的基础执行单元，它是比线程更轻量级的存在，由程序自身控制调度，不需要操作系统参与上下文切换。在Python中，使用async def定义的函数就是协程函数，调用协程函数不会立即执行函数体，而是返回一个协程对象。
事件循环
事件循环是异步编程的核心调度器，负责监听和分发事件，不断检查是否有可执行的协程任务，按照注册顺序和执行状态调度协程运行。所有的异步任务都需要在事件循环中注册才能被执行。
基础协程定义与执行
首先来看最简单的协程定义和执行方式，通过asyncio.run()可以快速启动一个事件循环并运行指定的协程。
import asyncio
# 定义协程函数
async def hello_async():
print("开始执行协程")
# 模拟IO等待，不会阻塞事件循环
await asyncio.sleep(1)
print("协程执行结束")
# 启动事件循环运行协程
if __name__ == "__main__":
asyncio.run(hello_async())
上述代码中，await关键字用于挂起当前协程，等待后面的可等待对象（这里是asyncio.sleep返回的协程）执行完成，挂起期间事件循环可以去执行其他任务。
多协程并发执行
单个协程无法体现异步的优势，实际开发中通常需要同时运行多个异步任务，这时候可以使用asyncio.gather()来并发执行多个协程。
import asyncio
import time
# 定义两个协程任务
async def task1():
print("任务1开始")
await asyncio.sleep(2)
print("任务1结束")
return "task1_result"
async def task2():
print("任务2开始")
await asyncio.sleep(1)
print("任务2结束")
return "task2_result"
async def main():
start_time = time.time()
# 并发执行两个任务
results = await asyncio.gather(task1(), task2())
end_time = time.time()
print(f"两个任务返回结果：{results}")
print(f"总耗时：{end_time - start_time:.2f}秒")
if __name__ == "__main__":
asyncio.run(main())
运行上述代码可以看到，两个任务的总耗时接近2秒，而不是两个任务耗时的总和3秒，说明两个任务是并发执行的，task2在task1等待的期间也得到了执行。
同步与异步执行对比
为了更直观地看到异步编程的优势，我们对比同步执行和异步执行多个IO任务的情况。
同步执行示例
import time
def sync_task(task_id, sleep_time):
print(f"同步任务{task_id}开始")
time.sleep(sleep_time)
print(f"同步任务{task_id}结束")
return f"sync_task{task_id}_result"
def sync_main():
start_time = time.time()
results = []
for i in range(3):
# 依次执行3个任务，每个任务等待1秒
res = sync_task(i, 1)
results.append(res)
end_time = time.time()
print(f"同步执行结果：{results}")
print(f"同步总耗时：{end_time - start_time:.2f}秒")
if __name__ == "__main__":
sync_main()
异步执行示例
import asyncio
import time
async def async_task(task_id, sleep_time):
print(f"异步任务{task_id}开始")
await asyncio.sleep(sleep_time)
print(f"异步任务{task_id}结束")
return f"async_task{task_id}_result"
async def async_main():
start_time = time.time()
# 创建3个协程任务
tasks = [async_task(i, 1) for i in range(3)]
results = await asyncio.gather(*tasks)
end_time = time.time()
print(f"异步执行结果：{results}")
print(f"异步总耗时：{end_time - start_time:.2f}秒")
if __name__ == "__main__":
asyncio.run(async_main())
同步执行3个各耗时1秒的任务总耗时约3秒，而异步执行总耗时约1秒，效率提升非常明显。
事件循环的高级用法
除了使用asyncio.run()这种简化的启动方式，也可以手动获取事件循环对象，更灵活地控制事件循环的运行。
import asyncio
async def demo_task():
print("手动控制事件循环的任务执行")
await asyncio.sleep(0.5)
print("任务执行完成")
# 获取当前事件循环
loop = asyncio.get_event_loop()
try:
# 将协程加入事件循环并执行
loop.run_until_complete(demo_task())
finally:
# 关闭事件循环
loop.close()
这种方式适合需要在现有程序中嵌入异步逻辑的场景，不过在Python 3.7及以上版本，更推荐使用asyncio.run()，它会自动处理事件循环的创建和关闭，避免资源泄漏问题。
注意事项
- 不要在协程中使用
time.sleep()这类阻塞线程的方法，会导致整个事件循环阻塞，应该使用await asyncio.sleep()。 - 只有
await后面的对象是可等待对象（协程、Task、Future）时才能正常挂起协程，普通函数调用不能加await。 - 异步编程适合IO密集型场景，对于CPU密集型任务，异步无法提升性能，反而可能因为调度开销降低效率，这类场景更适合使用多进程。
asyncio协程事件循环Python异步编程修改时间：2026-06-10 17:57:34