【一起学 Python】第 69 天:Asyncio 基础:async 与 await

async def hello():    return "Hello, Asyncio!"

async def example():    passcoro = example()print(coro)  # <coroutine object example at 0x...># ⚠️ 这只是创建了对象，还没有执行！

import asyncioasync def my_task():    print("Task is running")    await asyncio.sleep(1)    print("Task completed")# 创建 Tasktask = asyncio.create_task(my_task())# 或者使用 ensure_future（旧写法）# task = asyncio.ensure_future(my_task())

# Future 通常由底层库自动创建# 例如 asyncio.sleep() 返回的就是一个 Future

# ✅ 正确用法async def fetch_data():    return "data"# ❌ 错误用法def fetch_data():  # 这是普通函数，不是协程    return "data"

async def main():    result = await some_async_function()  # 等待返回结果    print(result)

# ✅ 正确async def good():    await asyncio.sleep(1)# ❌ 错误def bad():    await asyncio.sleep(1)  # SyntaxError: 'await' outside async function

import asyncioasync def say_hello():    print("你好")    await asyncio.sleep(1)  # 模拟耗时操作    print("世界")# 方式1：使用 asyncio.run()（推荐，Python 3.7+）asyncio.run(say_hello())# 方式2：使用事件循环（旧写法）# loop = asyncio.get_event_loop()# loop.run_until_complete(say_hello())

你好（等待 1 秒）世界

import asyncioasync def fetch_data(url, delay):    print(f"开始获取: {url}")    await asyncio.sleep(delay)  # 模拟网络请求    print(f"完成获取: {url}")    return f"Data from {url}"async def main():    # 并发执行三个任务    task1 = asyncio.create_task(fetch_data("http://api1.com", 1 ))    task2 = asyncio.create_task(fetch_data("http://api2.com", 2 ))    task3 = asyncio.create_task(fetch_data("http://api3.com", 1.5 ))    # 等待所有任务完成    results = await asyncio.gather(task1, task2, task3)    return results# 执行results = asyncio.run(main())print("所有结果:", results)

开始获取: http://api1.com开始获取: http://api2.com开始获取: http://api3.com完成获取: http://api1.com完成获取: http://api3.com完成获取: http://api2.com所有结果: ['Data from http://api1.com', 'Data from http://api2.com', 'Data from http://api3.com']

import asyncioasync def task_a():    print("[A] 开始执行")    await asyncio.sleep(1)    print("[A] 完成执行")    return "A的结果"async def task_b():    print("[B] 开始执行")    await asyncio.sleep(0.5)    print("[B] 完成执行")    return "B的结果"async def main():    # 顺序等待（同步风格）    result_a = await task_a()  # 等待 1 秒    result_b = await task_b()  # 再等待 0.5 秒，总共 1.5 秒    print(f"结果A: {result_a}")    print(f"结果B: {result_b}")asyncio.run(main())

async def main_concurrent():    # 并发执行（推荐）    results = await asyncio.gather(task_a(), task_b())  # 总共只需 1 秒！    print(f"结果: {results}")asyncio.run(main_concurrent())

import asyncioasync def risky_task():    await asyncio.sleep(0.5)    raise ValueError("发生了一个错误")async def safe_main():    try:        await risky_task()    except ValueError as e:        print(f"捕获异常: {e}")asyncio.run(safe_main())

# ❌ 错误：没有 await，协程没有执行async def main():    fetch_data()  # 这只是创建了协程对象，并未执行# ✅ 正确async def main():    await fetch_data()  # 执行协程

# Coroutine：还没有被调度执行coro = fetch_data()# Task：已被调度，等待执行task = asyncio.create_task(fetch_data())# 在 asyncio.gather 中，两者都可以用await asyncio.gather(fetch_data(), fetch_data())

# ❌ 错误def synchronous_function():    result = await async_function()  # SyntaxError# ✅ 正确：需要在异步上下文中async def wrapper():    result = await async_function()    return result

┌─────────────────────────────────────────────────────────┐│              进阶阶段学习成果                             │├─────────────────────────────────────────────────────────┤│                                                           ││  📊 类型提示模块（第 52-56 天）                          ││     ├─ 联合类型、类型别名、泛型                          ││     └─ 让代码更安全、更易维护                            ││                                                           ││  🛡️ 异常与文件模块（第 57-64 天）                        ││     ├─ 异常处理、文件 I/O、上下文管理器                  ││     └─ 编写健壮的错误处理机制                            ││                                                           ││  ⚡ 并发与迭代模块（第 65-69 天）                        ││     ├─ 迭代器、生成器、GIL、Asyncio                      ││     └─ 高效的数据处理与并发编程                          ││                                                           │└─────────────────────────────────────────────────────────┘