Python脚本如何获取异步任务返回值

wen python案例 2026-06-27 25

本文目录导读：

Python脚本如何获取异步任务返回值

使用 asyncio.run() 直接运行
使用 await 获取单个任务返回值
使用 asyncio.gather() 获取多个任务返回值
使用 asyncio.create_task() 和 await 获取返回值
使用 asyncio.wait() 等待任务完成
使用 asyncio.as_completed() 按完成顺序获取
异常处理
完整示例：模拟下载多个文件
关键要点

在Python中获取异步任务返回值主要有以下几种方式：

使用 `asyncio.run()` 直接运行

最简单的场景,直接获取单个协程的返回值：

import asyncio
async def fetch_data():
    await asyncio.sleep(1)
    return "数据获取成功"
# 直接获取返回值
result = asyncio.run(fetch_data())
print(result)  # 输出: 数据获取成功

使用 `await` 获取单个任务返回值

在异步函数内部等待协程完成：

async def main():
    result = await fetch_data()
    print(f"任务结果: {result}")
asyncio.run(main())

使用 `asyncio.gather()` 获取多个任务返回值

并发执行多个任务并获取所有返回值：

async def task1():
    await asyncio.sleep(2)
    return "任务1完成"
async def task2():
    await asyncio.sleep(1)
    return "任务2完成"
async def main():
    # 并行执行两个任务，返回结果列表
    results = await asyncio.gather(
        task1(),
        task2(),
        return_exceptions=True  # 如果有异常也返回，不抛出
    )
    print(results)  # 输出: ['任务1完成', '任务2完成']
asyncio.run(main())

使用 `asyncio.create_task()` 和 `await` 获取返回值

创建任务后,稍后等待获取结果：

async def main():
    # 创建任务（马上开始执行）
    task1 = asyncio.create_task(fetch_data())
    # 执行其他操作...
    await asyncio.sleep(0.5)
    # 等待任务完成并获取返回值
    result = await task1
    print(f"任务返回值: {result}")
asyncio.run(main())

使用 `asyncio.wait()` 等待任务完成

async def main():
    tasks = [
        asyncio.create_task(task1()),
        asyncio.create_task(task2())
    ]
    # 等待所有任务完成
    done, pending = await asyncio.wait(
        tasks,
        timeout=None,  # 无超时
        return_when=asyncio.ALL_COMPLETED
    )
    # 获取任务返回值
    for task in done:
        result = task.result()
        print(f"任务结果: {result}")
asyncio.run(main())

使用 `asyncio.as_completed()` 按完成顺序获取

async def main():
    tasks = [task1(), task2()]
    # 按完成顺序逐个获取结果
    for coro in asyncio.as_completed(tasks):
        result = await coro
        print(f"完成的先: {result}")
asyncio.run(main())

异常处理

获取返回值时处理可能的异常：

async def risky_task():
    await asyncio.sleep(1)
    raise ValueError("发生错误")
async def main():
    try:
        # 单个任务异常
        result = await risky_task()
    except ValueError as e:
        print(f"捕获异常: {e}")
    # 多个任务异常处理
    tasks = [task1(), risky_task()]
    results = await asyncio.gather(
        *tasks,
        return_exceptions=True  # 将异常作为返回值返回
    )
    for r in results:
        if isinstance(r, Exception):
            print(f"任务异常: {r}")
        else:
            print(f"任务结果: {r}")
asyncio.run(main())

完整示例：模拟下载多个文件

import asyncio
import time
async def download_file(url, delay):
    """模拟下载文件"""
    print(f"开始下载: {url}")
    await asyncio.sleep(delay)
    print(f"完成下载: {url}")
    return f"{url} 的内容"
async def main():
    urls = [
        ("http://example.com/file1.txt", 3),
        ("http://example.com/file2.txt", 1),
        ("http://example.com/file3.txt", 2),
    ]
    # 创建所有下载任务
    tasks = [download_file(url, delay) for url, delay in urls]
    # 方法1: 等待所有完成并获取结果
    print("=== 等待所有任务完成 ===")
    start = time.time()
    results = await asyncio.gather(*tasks)
    print(f"所有结果: {results}")
    print(f"总用时: {time.time() - start:.2f}秒")
    # 方法2: 按完成顺序获取结果
    print("\n=== 按完成顺序获取结果 ===")
    tasks2 = [download_file(url, delay) for url, delay in urls]
    for coro in asyncio.as_completed(tasks2):
        result = await coro
        print(f"优先完成: {result}")
if __name__ == "__main__":
    asyncio.run(main())