Python协程 Coroutine-JXin的博客

学习笔记说明

这是一篇Python的学习笔记，关于Python协程 Coroutine，仅做记录

回调之痛
协程函数
协程 VS 线程
驱动协程对象
Task
asyncio.run
深度总结
作业
作业答案
- 一、实现 gather 函数
- 二、实现 Event 类

回调之痛

在之前的课程中，我们用 Future 把延时函数和网络请求封装成了异步模式，但使用者依然需要注册回调：

async_delay(2).add_done_callback(lambda _: print("2秒后执行"))

如果请求之后还有请求，就会出现回调地狱：

async_request("host1").add_done_callback(lambda r1:
    async_request("host2").add_done_callback(lambda r2:
        async_request("host3").add_done_callback(lambda r3:
            ...
        )
    )
)

1
2
3
4
5
6
7

代码横向增长，可读性急剧下降。

协程函数

Python 中，使用 async def 定义一个协程函数。调用协程函数会得到一个协程对象。

from async_delay import async_delay
from async_request import async_request
import asyncio

async def test():
    print("main begins")
    resp1 = await async_request("localhost", 5500, "/index.html")
    print("resp1", resp1[:9])
    await async_delay(1)
    print("delayed")
    resp2 = await async_request("localhost", 5500, "/index.html")
    print("resp2", resp2[:9])
    return "ok"

1
2
3
4
5
6
7
8
9
10
11
12
13

协程 VS 线程

对比维度	🧵 线程 (Thread)	🍃 协程 (Coroutine)
运行载体	运行在操作系统内核态，受系统管理。	完全运行在用户态，由事件循环（如 `asyncio`）管理。
资源开销	较高每个线程需要独立的栈空间（通常数MB）和内核资源，创建和切换开销大。	极低协程栈空间很小（通常几KB），切换只需保存少量CPU寄存器上下文。
切换成本	昂贵涉及用户态与内核态切换，需要数百个CPU时钟周期。	非常廉价纯用户态操作，仅需几十个时钟周期。
数据同步	复杂且易出错多线程共享内存，需要使用 `锁(Lock)`、`信号量(Semaphore)` 等机制，容易产生死锁、竞态条件。	相对安全单线程内运行，同一时刻只有一个协程在执行，天然避免了数据竞争。但使用多线程事件循环时仍需注意。
利用多核	✅ 原生支持可将不同线程分配到不同CPU核心，实现并行计算（CPU密集型任务）。	❌ 单线程内不支持一个事件循环默认只运行在一个线程、一个核心上。需要配合 `asyncio` 的 `run_in_executor` 或多进程才能利用多核。
适用场景	CPU密集型任务或对实时性要求高的I/O任务（通过多线程掩盖阻塞）	海量I/O密集型任务网络爬虫、Web服务器（如FastAPI）、聊天服务、高并发数据库访问等。可以轻松创建成千上万个协程。
代表库/语法	`threading`, `concurrent.futures.ThreadPoolExecutor`	`asyncio`, `async`/`await`, `trio`, `gevent` (基于协程的库)
典型数量级	百级别（数百个线程开销已相当可观）	万/十万级别（轻松创建数万个协程）

驱动协程对象


def main():
    coro = test()
    loop = asyncio.get_event_loop()
    f = loop.create_future()
    r = coro.send(None)

    def done_callback(fut):
        try:
            r = coro.send(fut.result())
            r.add_done_callback(done_callback)
        except StopIteration as e:
            f.set_result(e.value)

    r.add_done_callback(done_callback)
    return f

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

def run():
    fut = main()

    def done_callback(fut):
        loop.stop()
        print(fut.result())

    fut.add_done_callback(done_callback)

loop.call_soon(run)
loop.run_forever()
print("over")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

Task

Task是Future的子类，它专门用于驱动协程对象的执行

def main():
    coro = test()
    loop = asyncio.get_event_loop()
    f = loop.create_future()
    r = coro.send(None)

    def done_callback(fut):
        try:
            r = coro.send(fut.result())
            r.add_done_callback(done_callback)
        except StopIteration as e:
            f.set_result(e.value)

    r.add_done_callback(done_callback)
    return f

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

def run():
    fut = main()

    def done_callback(fut):
        loop.stop()
        print(fut.result())

    fut.add_done_callback(done_callback)

loop.call_soon(run)
loop.run_forever()
print("over")

# 等效于
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

def run():
    task = asyncio.create_task(test())  # 将协程包装成一个Task

    def done_callback(task):
        loop.stop()
        print(task.result())

    task.add_done_callback(done_callback)

loop.call_soon(run)
loop.run_forever()
print("over")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

asyncio.run

asyncio.run可以直接驱动一个协程对象，在内部会将其转换为Task

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

def run():
    task = asyncio.create_task(test())  # 将协程包装成一个Task

    def done_callback(task):
        loop.stop()
        print(task.result())

    task.add_done_callback(done_callback)

loop.call_soon(run)
loop.run_forever()
print("over")

# 等效于
result = asyncio.run(test())
print(result)
print("over")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

深度总结

async修饰的函数称之为协程函数/异步函数，调用后返回协程对象
协程对象可以通过asyncio.create_task包装成一个Task，用于驱动协程对象
1. Task创建后，会立即启动协程对象的执行，执行放到ready队列中
2. 协程对象执行结束，Task完成，完成的数据即是协程对象的返回值
await关键字可以等待一个awaitable对象
1. await必须在协程函数中
2. 常见的awaitable对象：协程对象、Task、Future
asyncio.run可接收一个协程对象，其在内部转换为Task

作业

一、实现 gather 函数

import asyncio
from async_delay import async_delay
from typing import Coroutine

def gather(*aws: Coroutine) -> asyncio.Future:
    # 你的代码
    pass

async def coro(name: str, duration: int):
    await async_delay(duration)
    return f"{name} 完成"

async def main():
    results = await gather(
        coro("A", 2),
        coro("B", 1),
        coro("C", 3),
    )
    print(results)  # 预期: ['A 完成', 'B 完成', 'C 完成']

asyncio.run(main())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

二、实现 Event 类

import asyncio
from async_delay import async_delay
from gather import gather

class Event:
    # 你的代码
    pass

async def test():
    event = Event()

    async def waiter():
        print("waiter: 开始等待")
        await event.wait()
        print("waiter: 被唤醒")

    async def setter():
        print("setter: 1秒后设置事件")
        await async_delay(1)
        event.set()
        print("setter: 事件已设置")

    await gather(waiter(), setter())

asyncio.run(test())

# 预期结果：
""" 
waiter: 开始等待
setter: 1秒后设置事件
setter: 事件已设置
waiter: 被唤醒 
"""

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

作业答案

一、实现 gather 函数

import asyncio
from async_delay import async_delay
from typing import Coroutine

def gather(*aws: Coroutine) -> asyncio.Future:
    future = asyncio.Future()
    finish_count = 0  # 已完成的任务数量
    count = len(aws)  # 任务总数
    results = [None] * count  # 存储结果的列表

    def on_done(f: asyncio.Future, idx: int) -> None:
        nonlocal finish_count
        results[idx] = f.result()  # 将结果存储在对应位置
        finish_count += 1
        if finish_count == count:  # 所有任务完成
            future.set_result(results)  # 设置最终结果

    for i, aw in enumerate(aws):
        asyncio.create_task(aw).add_done_callback(lambda f, idx=i: on_done(f, idx))

    return future

async def coro(name: str, duration: int):
    await async_delay(duration)
    return f"{name} 完成"

async def main():
    results = await gather(
        coro("A", 2),
        coro("B", 1),
        coro("C", 3),
    )
    print(results)  # 预期: ['A 完成', 'B 完成', 'C 完成']

asyncio.run(main())

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

二、实现 Event 类

import asyncio
from async_delay import async_delay
from gather import gather

# 以下是简易版的实现
class Event:
    def __init__(self):
        self._fut = None

    def set(self):
        if self._fut:
            self._fut.set_result(None)

    async def wait(self):
        loop = asyncio.get_running_loop()
        self._fut = loop.create_future()
        await self._fut

async def test():
    event = Event()

    async def waiter():
        print("waiter: 开始等待")
        await event.wait()
        print("waiter: 被唤醒")

    async def setter():
        print("setter: 1秒后设置事件")
        await async_delay(1)
        event.set()
        print("setter: 事件已设置")

    await gather(waiter(), setter())

if __name__ == "__main__":
    asyncio.run(test())

    """ 
    waiter: 开始等待
    setter: 1秒后设置事件
    setter: 事件已设置
    waiter: 被唤醒 
    """

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43