1
import asyncio
2

3
async def fetch_name() -> str:
4
    await asyncio.sleep(0.1)  # 模拟网络等待，不阻塞线程
5
    return "Enkidu"
6

7
async def main() -> None:
8
    name = await fetch_name()
9
    print(name)
10

11
asyncio.run(main())

1
任务 A 开始 → 等待异步网络响应，暂停 A
2
→ 事件循环运行已就绪的任务 B
3
→ A 的响应就绪 → 等到调度机会后恢复 A

1
# ❶ Coroutine — 协程对象（最常用）
2
coro = fetch_data(url)    # async def 不加 await 返回的就是这个
3
result = await coro       # await 它才开始执行
4

5
# ❷ Task — 任务（立即排入事件循环）
6
task = asyncio.create_task(fetch_data(url))  # 创建即排入！不等 await
7
# ... 这期间 task 已经在后台跑了 ...
8
result = await task       # 拿结果（可能已经好了，当场返回）
9

10
# ❸ Future — 底层占位符（通常不需要手动创建）
11
# Task 是 Future 的子类，日常只用 Coroutine 和 Task 就够了

1
import asyncio
2

3
# 限制：最多同时 3 个请求
4
semaphore = asyncio.Semaphore(3)
5

6
async def fetch_with_limit(url: str):
7
    async with semaphore:           # 拿号（满了就等）
8
        return await fetch(url)     # 执行请求
9
    # 出 with 块自动还号
10

11
async def fetch_many(urls: list):
12
    tasks = [fetch_with_limit(u) for u in urls]
13
    return await asyncio.gather(*tasks)
14

15
# 如果有 100 个 URL，同时只有 3 个在请求
16
# 第 4 个必须等前面有人完成才进去

1
import time
2
import asyncio
3

4
# 这是一个同步阻塞函数（比如别人写的库）
5
def cpu_heavy_task(n: int) -> int:
6
    time.sleep(2)           # 阻塞！整个线程卡住 2 秒
7
    return sum(range(n))
8

9
async def main():
10
    loop = asyncio.get_running_loop()
11

12
    # ❌ 直接调：整个事件循环卡死 2 秒
13
    # result = cpu_heavy_task(10000000)
14

15
    # ✅ 扔进线程池：其他协程不受影响
16
    result = await loop.run_in_executor(None, cpu_heavy_task, 10000000)
17
    return result

1
async def safe_fetch(url: str):
2
    try:
3
        return await fetch(url)
4
    except asyncio.TimeoutError:
5
        return f"{url} 超时了"
6
    except Exception as e:
7
        return f"{url} 出错：{e}"
8

9
# gather 的错误处理：return_exceptions=True
10
async def fetch_all(urls: list):
11
    results = await asyncio.gather(
12
        *[fetch(u) for u in urls],
13
        return_exceptions=True   # 🔑 单个失败不影响其他
14
    )
15
    for i, r in enumerate(results):
16
        if isinstance(r, Exception):
17
            print(f"{urls[i]} 失败了: {r}")
18
        else:
19
            print(f"{urls[i]} 成功: {r}")

1
# ─── 定义 ───
2
async def f():           # 协程函数
3
await f()                # 等待协程完成
4
asyncio.create_task(f()) # 创建 Task（立即排入事件循环）
5

6
# ─── 并发 ───
7
await asyncio.gather(a(), b(), c())          # 同时跑，全完成返回
8
for t in asyncio.as_completed([a(), b()]):   # 谁先好先处理谁
9

10
# ─── 控制 ───
11
async with asyncio.Semaphore(n):  # 限制并发数
12
await asyncio.sleep(n)            # 异步等 n 秒（不阻塞）
13
await asyncio.wait_for(f(), 5)    # 超时抛 TimeoutError
14

15
# ─── 混合 ───
16
await loop.run_in_executor(None, sync_func, arg)  # 同步函数丢线程池
17

18
# ─── 顶层 ───
19
asyncio.run(main())       # 启动事件循环（脚本入口，FastAPI 不用）

1
# 另存为 async_playground.py 跑一下感受区别
2
import asyncio, time
3

4
# ─── 实验 1：同步 vs 异步等待 ───
5
async def async_wait(name, n):
6
    await asyncio.sleep(n)
7
    return f"{name} 完成"
8

9
def sync_wait(name, n):
10
    time.sleep(n)
11
    return f"{name} 完成"
12

13
# 异步并发：3 秒
14
async def test_async():
15
    t0 = time.time()
16
    results = await asyncio.gather(
17
        async_wait("A", 1), async_wait("B", 1), async_wait("C", 1)
18
    )
19
    print(f"异步: {time.time()-t0:.1f}s → {results}")
20

21
# 同步顺序：3 秒
22
def test_sync():
23
    t0 = time.time()
24
    results = [sync_wait("A",1), sync_wait("B",1), sync_wait("C",1)]
25
    print(f"同步: {time.time()-t0:.1f}s → {results}")
26

27
asyncio.run(test_async())  # 异步: 1.0s
28
test_sync()                # 同步: 3.0s ← 三倍！