Python的异步编程基于asyncio模块实现，通过异步任务可以让程序在等待网络请求、文件读写等IO操作时，切换到其他任务继续执行，大幅提升资源利用率。异步任务的核心是协程，协程是通过async关键字定义的函数，调用协程函数不会立即执行，而是返回一个协程对象，需要交给事件循环调度执行。
基础异步任务创建方式
定义协程函数
首先需要使用async关键字定义协程函数，协程函数内部可以使用await关键字挂起当前协程，等待其他可等待对象完成。以下是一个简单的协程函数示例：
import asyncio
# 定义协程函数
async def async_task(name, delay):
print(f"任务{name}开始执行，等待{delay}秒")
# 模拟IO等待，await挂起当前协程
await asyncio.sleep(delay)
print(f"任务{name}执行完成")
return f"任务{name}的结果"
通过事件循环运行单个任务
协程对象需要交给事件循环才能执行，Python3.7及以上版本可以直接使用asyncio.run()方法运行协程，该方法会自动创建事件循环并在执行完成后关闭：
import asyncio
async def single_task():
print("单个异步任务开始")
await asyncio.sleep(1)
print("单个异步任务结束")
if __name__ == "__main__":
# 运行单个协程任务
result = asyncio.run(single_task())
print(f"任务返回结果：{result}")
创建多个并发异步任务
实际开发中经常需要同时运行多个异步任务，asyncio提供了多种创建并发任务的方式，常用的有create_task()和gather()。
使用create_task创建任务
asyncio.create_task()可以将协程包装为任务对象，任务会被立即调度执行，返回的任务对象可以用于后续获取结果或添加回调：
import asyncio
async def task_a():
print("任务A开始")
await asyncio.sleep(2)
print("任务A结束")
return "A的结果"
async def task_b():
print("任务B开始")
await asyncio.sleep(1)
print("任务B结束")
return "B的结果"
async def main():
# 创建两个异步任务，立即调度执行
task1 = asyncio.create_task(task_a())
task2 = asyncio.create_task(task_b())
# 等待两个任务完成并获取结果
result1 = await task1
result2 = await task2
print(f"任务1结果：{result1}，任务2结果：{result2}")
if __name__ == "__main__":
asyncio.run(main())
使用gather批量创建任务
asyncio.gather()可以接收多个协程或可等待对象，批量创建任务并等待所有任务完成，返回所有任务的结果列表，结果的顺序和传入的协程顺序一致：
import asyncio
async def task(name, delay):
print(f"{name}开始执行")
await asyncio.sleep(delay)
print(f"{name}执行完成")
return f"{name}耗时{delay}秒"
async def main():
# 批量创建3个异步任务
results = await asyncio.gather(
task("任务1", 1),
task("任务2", 2),
task("任务3", 1.5)
)
print("所有任务完成，结果如下：")
for res in results:
print(res)
if __name__ == "__main__":
asyncio.run(main())
异步任务的异常处理
异步任务执行过程中如果抛出异常，默认不会直接抛出，而是会保存在任务对象中，需要在获取结果时处理。使用gather()时可以通过return_exceptions参数控制异常的处理方式：
import asyncio
async def error_task():
print("异常任务开始执行")
await asyncio.sleep(1)
# 主动抛出异常
raise ValueError("任务执行出错")
async def normal_task():
print("正常任务开始执行")
await asyncio.sleep(0.5)
print("正常任务执行完成")
return "正常结果"
async def main():
# return_exceptions=True时，异常会作为结果返回，不会中断其他任务
results = await asyncio.gather(
normal_task(),
error_task(),
return_exceptions=True
)
for res in results:
if isinstance(res, Exception):
print(f"捕获到异常：{res}")
else:
print(f"任务结果：{res}")
if __name__ == "__main__":
asyncio.run(main())
两种任务创建方式的区别
为了更清晰区分create_task()和gather()的使用场景，以下是两者的核心差异对比：
| 对比项 | create_task() | gather() |
|---|---|---|
| 适用场景 | 需要单独控制单个任务，或者任务执行顺序不固定时 | 需要批量运行多个任务，且需要统一等待所有任务完成时 |
| 返回值 | 返回单个Task对象，需要单独await获取结果 | 返回所有任务结果的列表，顺序和传入顺序一致 |
| 异常传播 | 异常会在await任务对象时抛出 | 默认异常会直接抛出中断执行，设置return_exceptions=True时异常作为结果返回 |
注意事项
- 异步任务中不能使用同步阻塞操作，比如
time.sleep()，否则会阻塞整个事件循环，需要使用asyncio.sleep()替代。 - 只有在事件循环运行的过程中，
create_task()才能正常创建任务，不能在协程外部直接调用该方法。 - 如果不需要等待任务结果，只是需要触发任务执行，可以不保存任务对象，但需要注意任务生命周期，避免任务未执行完事件循环就关闭。
asyncio异步任务event_loopasync_await修改时间：2026-07-21 09:54:26