Python的asyncio库为开发者提供了完善的异步编程支持，协程任务管理是其中非常实用的部分，能够帮助我们灵活控制多个并发协程的执行流程，适配不同的业务场景需求。
asyncio协程任务基础
在asyncio中，协程本身只是定义异步逻辑的函数，需要通过任务（Task）包装后才能被事件循环调度执行。我们可以通过asyncio.create_task()方法将协程转换为任务，任务创建后会自动加入事件循环等待执行。
下面是一个最简单的任务创建示例：
import asyncio
# 定义异步协程
async def demo_coroutine(name, delay):
print(f"任务{name}开始执行")
await asyncio.sleep(delay)
print(f"任务{name}执行完成，耗时{delay}秒")
return f"任务{name}结果"
async def main():
# 创建任务
task1 = asyncio.create_task(demo_coroutine("A", 2))
task2 = asyncio.create_task(demo_coroutine("B", 1))
# 等待所有任务完成
results = await asyncio.gather(task1, task2)
print(f"所有任务结果：{results}")
if __name__ == "__main__":
asyncio.run(main())
常用任务管理方法
批量任务管理
当需要处理多个同类任务时，可以使用asyncio.gather()批量管理，它支持传入多个任务或者协程，等待所有传入对象执行完成后返回结果列表，结果的顺序和传入参数的顺序一致。
如果需要等待第一个完成的任务就返回结果，可以使用asyncio.wait()配合FIRST_COMPLETED参数：
import asyncio
async def task_func(task_id, delay):
await asyncio.sleep(delay)
return f"任务{task_id}完成"
async def main():
tasks = [asyncio.create_task(task_func(i, i+1)) for i in range(3)]
# 等待第一个任务完成
done, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
for t in done:
print(f"首个完成任务结果：{t.result()}")
# 取消剩余未完成任务
for p in pending:
p.cancel()
if __name__ == "__main__":
asyncio.run(main())
任务超时控制
实际开发中经常需要给任务设置超时时间，避免任务长时间阻塞。可以使用asyncio.wait_for()方法，传入任务对象和超时时间，如果任务在超时时间内未完成会抛出asyncio.TimeoutError异常。
import asyncio
async def long_task():
await asyncio.sleep(5)
return "任务完成"
async def main():
try:
# 设置3秒超时
result = await asyncio.wait_for(long_task(), timeout=3)
print(result)
except asyncio.TimeoutError:
print("任务执行超时")
if __name__ == "__main__":
asyncio.run(main())
任务异常处理
任务执行过程中如果出现异常，不会被直接抛出，而是会保存在任务对象中。我们可以通过任务的result()方法获取结果，如果任务执行时抛出了异常，调用result()时会重新抛出该异常，因此需要在获取结果时做异常捕获。
import asyncio
async def error_task():
raise ValueError("任务执行出错")
async def main():
task = asyncio.create_task(error_task())
try:
await task
result = task.result()
print(result)
except ValueError as e:
print(f"捕获到任务异常：{e}")
if __name__ == "__main__":
asyncio.run(main())
实战场景示例
假设我们需要并发请求多个接口，同时控制最大并发数量，避免对目标服务造成过大压力，可以结合任务管理和信号量实现：
import asyncio
import random
# 模拟接口请求
async def request_api(api_id, semaphore):
async with semaphore:
delay = random.uniform(0.5, 2)
await asyncio.sleep(delay)
print(f"接口{api_id}请求完成，耗时{delay:.2f}秒")
return f"接口{api_id}返回数据"
async def main():
# 控制最大并发数为3
semaphore = asyncio.Semaphore(3)
# 创建10个请求任务
tasks = [asyncio.create_task(request_api(i, semaphore)) for i in range(10)]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果，区分正常返回和异常
for i, res in enumerate(results):
if isinstance(res, Exception):
print(f"接口{i}请求失败：{res}")
else:
print(f"接口{i}请求成功：{res}")
if __name__ == "__main__":
asyncio.run(main())
注意事项
- 不要在同步函数中直接调用异步协程，必须通过事件循环驱动执行
- 任务取消后，被取消的任务会抛出
asyncio.CancelledError异常，需要在协程内部做对应处理 - 使用
asyncio.gather()时如果传入return_exceptions=True，异常会被作为结果返回而不是抛出，方便统一处理 - 避免在异步代码中执行耗时的同步阻塞操作，会阻塞整个事件循环，必要时可以使用
asyncio.to_thread()将同步操作放到线程中执行
asyncio协程任务管理Python异步编程修改时间：2026-07-20 04:54:33