御气篇-Day84 综合 Kata：异步批量模拟数据查询实战

📖 知识点回顾

1. async def 定义协程函数，await 让出执行权

2. asyncio.create_task 创建异步任务，实现并发

3. asyncio.run() 作为异步程序入口

4. 异步适合 IO 等待场景；等待期间事件循环切换执行其他任务

任务需求

1. 导入 asyncio 与 time 模块

2. 编写协程函数 db_query(query_name, delay)

   • 打印：【查询】{query_name} 发起请求，预计等待 {delay}s

   • await asyncio.sleep (delay) 模拟 IO 等待

   • 返回结果字符串：{query_name} 查询成功

3. 编写 main 主协程

   • 准备查询任务列表：4 个查询，每个延迟 2 秒

   • 使用 create_task 批量创建任务，并发执行

   • await 等待全部任务完成，收集结果并打印

   • 统计总耗时

4. 对比思考：如果同步串行执行，理论耗时是 8 秒；异步并发预期约 2 秒

参考代码

import asyncio
import time

# 模拟数据库查询协程
async def db_query(query_name, delay):
    print(f"【查询】{query_name} 发起请求，预计等待 {delay}s")
    await asyncio.sleep(delay)
    result = f"{query_name} 查询成功"
    print(f"【查询】{query_name} 请求结束")
    return result


async def main():
    start_time = time.time()
    # 准备4个查询任务
    task_list = []
    for i in range(1, 5):
        task = asyncio.create_task(db_query(f"数据库查询{i}", 2))
        task_list.append(task)

    # 等待所有任务完成，收集返回结果
    results = []
    for task in task_list:
        res = await task
        results.append(res)

    end_time = time.time()
    print("\n==== 全部查询结果 ====")
    for item in results:
        print(item)
    print(f"\n✅ 异步执行总耗时：{end_time - start_time:.2f} 秒")


asyncio.run(main())

运行输出

【查询】数据库查询1 发起请求，预计等待 2s
【查询】数据库查询2 发起请求，预计等待 2s
【查询】数据库查询3 发起请求，预计等待 2s
【查询】数据库查询4 发起请求，预计等待 2s
【查询】数据库查询1 请求结束
【查询】数据库查询2 请求结束
【查询】数据库查询3 请求结束
【查询】数据库查询4 请求结束

==== 全部查询结果 ====
数据库查询1 查询成功
数据库查询2 查询成功
数据库查询3 查询成功
数据库查询4 查询成功

✅ 异步执行总耗时：2.01 秒

🧠 思考题

1. 如果这 4 个任务不是 sleep 等待，而是大量 CPU 计算，异步还能提速吗？为什么？

2. 把代码改成 asyncio.gather 写法，简化批量任务创建，尝试自己改写。