Python await案例如何等待异步执行

wen python案例 2026-06-30 206

Python await 等待异步执行的案例详解

基础等待示例

import asyncio
# 定义一个异步函数
async def delayed_task(delay, name):
    print(f"任务 {name} 开始，等待 {delay} 秒...")
    await asyncio.sleep(delay)  # 模拟异步操作
    print(f"任务 {name} 完成")
    return f"{name} 的结果"
# 主函数
async def main():
    print("开始执行...")
    # 等待单个异步任务
    result = await delayed_task(2, "A")
    print(f"获取到结果: {result}")
# 运行
asyncio.run(main())

多个任务并行等待

import asyncio
import time
async def fetch_data(url, delay):
    print(f"开始获取 {url}...")
    await asyncio.sleep(delay)  # 模拟网络请求
    print(f"完成获取 {url}")
    return f"{url} 的数据"
async def main():
    start = time.time()
    # 创建多个任务
    task1 = asyncio.create_task(fetch_data("api1.com", 2))
    task2 = asyncio.create_task(fetch_data("api2.com", 3))
    task3 = asyncio.create_task(fetch_data("api3.com", 1))
    # 等待所有任务完成
    results = await asyncio.gather(task1, task2, task3)
    print(f"所有结果: {results}")
    print(f"总耗时: {time.time() - start:.2f} 秒")
asyncio.run(main())

超时控制

import asyncio
async def long_running_task():
    await asyncio.sleep(10)
    return "任务完成"
async def main():
    try:
        # 设置超时时间
        result = await asyncio.wait_for(long_running_task(), timeout=3)
        print(f"结果: {result}")
    except asyncio.TimeoutError:
        print("任务超时！")
asyncio.run(main())

等待第一个完成的任务

import asyncio
async def task_a():
    await asyncio.sleep(3)
    return "任务A完成"
async def task_b():
    await asyncio.sleep(1)
    return "任务B完成"
async def main():
    # 创建任务列表
    tasks = [task_a(), task_b()]
    # 等待第一个任务完成
    done, pending = await asyncio.wait(
        tasks,
        return_when=asyncio.FIRST_COMPLETED
    )
    for task in done:
        print(f"完成的任务结果: {task.result()}")
    print(f"未完成的任务数: {len(pending)}")
asyncio.run(main())

实际应用：HTTP请求示例

import asyncio
import aiohttp
async def fetch_url(session, url):
    print(f"开始请求: {url}")
    async with session.get(url) as response:
        data = await response.text()
        return len(data)  # 返回响应长度
async def main():
    urls = [
        "http://example.com",
        "http://httpbin.org/get",
        "http://google.com"
    ]
    async with aiohttp.ClientSession() as session:
        # 创建所有任务
        tasks = [fetch_url(session, url) for url in urls]
        # 等待所有请求完成
        results = await asyncio.gather(*tasks, return_exceptions=True)
        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"{url} 请求失败: {result}")
            else:
                print(f"{url} 响应大小: {result} 字节")
asyncio.run(main())

带进度通知的等待

import asyncio
async def worker(name, duration, progress_callback):
    for i in range(duration):
        await asyncio.sleep(1)
        progress = (i + 1) / duration * 100
        await progress_callback(name, progress)
    return f"{name} 完成"
async def progress_reporter(name, progress):
    print(f"任务 {name}: {progress:.1f}%")
async def main():
    async def custom_callback(name, progress):
        await progress_reporter(name, progress)
    # 启动多个任务
    task1 = asyncio.create_task(worker("A", 3, custom_callback))
    task2 = asyncio.create_task(worker("B", 5, custom_callback))
    # 等待所有任务完成
    results = await asyncio.gather(task1, task2)
    print(f"最终结果: {results}")
asyncio.run(main())

错误处理与部分结果等待

import asyncio
async def unstable_task(name, delay, fail=False):
    await asyncio.sleep(delay)
    if fail:
        raise ValueError(f"任务 {name} 出错")
    return f"任务 {name} 成功"
async def main():
    tasks = [
        unstable_task("A", 2, fail=False),
        unstable_task("B", 1, fail=True),
        unstable_task("C", 3, fail=False)
    ]
    # 方式1: 获取所有结果（包含异常）
    results = await asyncio.gather(*tasks, return_exceptions=True)
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"任务 {['A','B','C'][i]} 异常: {result}")
        else:
            print(f"任务 {['A','B','C'][i]} 结果: {result}")
    # 方式2: 只处理成功结果
    success_results = [r for r in results if not isinstance(r, Exception)]
    print(f"成功的结果数: {len(success_results)}")
asyncio.run(main())

使用回调的方式

import asyncio
async def async_operation():
    await asyncio.sleep(2)
    return "操作结果"
def callback(future):
    try:
        result = future.result()
        print(f"回调函数收到结果: {result}")
    except Exception as e:
        print(f"回调函数收到异常: {e}")
async def main():
    # 创建future对象
    loop = asyncio.get_event_loop()
    future = loop.create_future()
    # 添加回调
    future.add_done_callback(callback)
    # 执行异步操作并设置future结果
    async def set_result():
        result = await async_operation()
        future.set_result(result)
    await asyncio.gather(set_result())
    await asyncio.sleep(0.1)  # 等待回调执行
asyncio.run(main())