本文章信息量较大，从 IO 多路复用，到生成器的使用，再到 async、await 背后的实现原理，深入浅出，剖析得非常透彻，非常硬核！

都是单线程，为什么原来低效率的代码用了 async、await 加一些异步库就变得效率高了？

0x00 开始之前

0x01 IO 多路复用

def handler(request):
    # 处理请求
    pass

# 你的 handler 运行在 while 循环中
while True:
    # 获取一个新请求
    request = accept()
    # 根据路由映射获取到用户写的业务逻辑函数
    handler = get_handler(request)
    # 运行用户的handler，处理请求
    handler(request)

# 操作系统的IO复用示例伪代码
# 向操作系统IO注册自己关注的IO操作的id和类型
io_register(io_id, io_type)
io_register(io_id, io_type)

# 获取完成的IO操作
events = io_get_finished()

for (io_id, io_type) in events:
    if io_type == READ:
        data = read_data(io_id) 
    elif io_type == WRITE:
        write_data(io_id，data)

call_backs = {}

def handler(req):
    # do jobs here
    io_register(io_id, io_type)
    def call_back(result):
        # 使用返回的result完成剩余工作...
    call_backs[io_id] = call_back

# 新的循环
while True：
    # 获取已经完成的io事件
    events = io_get_finished()
    for (io_id, io_type) in events:
        if io_type == READ: # 读取
            data = read(io_id) 
            call_back = call_backs[io_id]
            call_back(data)
        else:
            # 其他类型io事件的处理
            pass

    # 获取一个新请求
    request = accept()
    # 根据路由映射获取到用户写的业务逻辑函数
    handler = get_handler(request)
    # 运行用户的handler，处理请求
    handler(request)

0x02 用生成器消除 callback

def handler(request):
    # 业务逻辑代码...
    # 需要执行一次 API 请求
    def call_back(result):
        # 使用 API 返回的result完成剩余工作
        print(result)
    # 没有io_call这个方法，这里只是示意，表示注册一个IO操作
    asyncio.get_event_loop().io_call(api, call_back)

def example():
    value = yield 2
    print("get", value)
    return value

g = example()
# 启动生成器，我们会得到 2
got = g.send(None)
print(got)  # 2

try:
    # 再次启动 会显示 "get 4", 就是我们传入的值
    got = g.send(got*2)
except StopIteration as e:
    # 生成器运行完成，将会print(4)，e.value 是生成器return的值
    print(e.value)

def handler(request):
    # 业务逻辑代码...
    # 需要执行一次 API 请求，直接把 IO 请求信息yield出去
    result = yield io_info
    # 使用 API 返回的result完成剩余工作
    print(result)

# 这个函数注册到ioloop中，用来当有新请求的时候回调
def on_request(request):
    # 根据路由映射获取到用户写的业务逻辑函数
    handler = get_handler(request)
    g = handler(request)
    # 首次启动获得io_info
    io_info = g.send(None)

    # io完成回调函数
    def call_back(result):
        # 重新启动生成器
        g.send(result)

    asyncio.get_event_loop().io_call(io_info, call_back)

1. 业务逻辑中仅发起一次网络 IO，但实际中往往更多
2. 业务逻辑没有调用其他异步函数（协程），但实际中我们往往会调用其他协程

0x03 解决完整调用链

def func1():
    ret = yield request("http://test.com/foo")
    ret = yield func2(ret)
    return ret

def func2(data):
    result = yield request("http://test.com/"+data)
    return result

def request(url):
    # 这里模拟返回一个io操作，包含io操作的所有信息，这里用字符串简化代替
    result = yield "iojob of %s" % url
    return result

1. 调用 func1() 得到生成器
2. 调用 send(None) 启动它得到会得到 request("http://test.com/foo") 的结果，还是生成器对象
3. send(None) 启动由 request() 产生的生成器，会得到 IO 操作，由框架注册到 ioloop 并指定回调
4. IO 完成后的回调函数内唤醒 request 生成器，生成器会走到 return 语句结束
5. 捕获异常得到 request 生成器的返回值，将上一层 func1 唤醒，同时又得到 func2() 生成器
6. 继续执行...

1. 第一个生成器入栈
2. 调用 send，如果得到生成器就入栈并进入下一轮迭代
3. 遇到到 IO 请求 yield 出来，让框架注册到 ioloop
4. IO 操作完成后被唤醒，缓存结果并出栈，进入下一轮迭代，目的让上层函数使用 IO 结果恢复运行
5. 如果一个生成器运行完毕，也需要和4一样让上层函数恢复运行

def wrapper(gen):
    # 第一层调用 入栈
    stack = Stack()
    stack.push(gen)

    # 开始逐层调用
    while True:
        # 获取栈顶元素
        item = stack.peak()

        result = None
        # 生成器
        if isgenerator(item):
            try:
                # 尝试获取下层调用并入栈
                child = item.send(result)
                stack.push(child)
                # result 使用过后就还原为None
                result = None
                # 入栈后直接进入下次循环，继续向下探索
                continue
            except StopIteration as e:
                # 如果自己运行结束了，就暂存result，下一步让自己出栈
                result = e.value
        else:  # IO 操作
            # 遇到了 IO 操作，yield 出去，IO 完成后会被用 IO 结果唤醒并暂存到 result
            result = yield item

        # 走到这里则本层已经执行完毕，出栈，下次迭代将是调用链上一层
        stack.pop()
        # 没有上一层的话，那整个调用链都执行完成了，return        
        if stack.empty():
            print("finished")
            return result

w = wrapper(func1())
# 将会得到 "iojob of http://test.com/foo"
w.send(None)
# 上个iojob foo 完成后的结果"bar"传入，继续运行，得到 "iojob of http://test.com/bar"
w.send("bar")
# 上个iojob bar 完成后的结构"barz"传入，继续运行，结束。
w.send("barz")

# 维护一个就绪列表，存放所有完成的IO事件，格式为（wrapper，result） 
ready = []

def on_request(request):
    handler = get_handler(request)
    # 使用 wrapper 包装后，可以只通过 send 处理 IO 了
    g = wrapper(func1())
    # 把开始状态直接视为结果为None的就绪状态
    ready.append((g, None))

# 让ioloop每轮循环都执行此函数，用来处理的就绪的IO
def process_ready(self):
    def call_back(g, result):
        ready.append((g, result)) 

    # 遍历所有已经就绪生成器，将其向下推进
    for g, result in self.ready:  
        # 用result唤醒生成器，并得到下一个io操作
        io_job = g.send(result)
        # 注册io操作 完成后把生成器加入就绪列表，等待下一轮处理
        asyncio.get_event_loop().io_call(
            io_job, lambda result: ready.append((g, result)

0x04 提高扩展性

class Future：
    # 设置结果
    def set_result(result): pass
    # 获取结果
    def result():  pass
    #  表示这个future对象是不是已经被设置过结果了
    def done(): pass
    # 设置在他被设置结果时应该执行的回调函数，可以设置多个
    def add_done_callback(callback):  pass

# 现在 request 函数，不是生成器，它返回future
def request(url):
    # future 理解为占位符
    fut = Future()

    def callback(result):
        # 当网络IO完成回调的时候给占位符赋值
        fut.set_result(result)
    asyncio.get_event_loop().io_call(url, callback)

    # 返回占位符
    return future

def process_ready(self):
    def callback(fut):
        # future被设置结果会被放入就绪列表
        ready.append((g, fut.result()))

    # 遍历所有已经就绪生成器，将其向下推进
    for g, result in self.ready:  
        # 用result唤醒生成器，得到的不再是io操作，而是future
        fut = g.send(result)
        # future被设置结果的时候会调用callback
        fut.add_done_callback(callback)

0x05 发展和变革

def func1():
    # 注意 yield from
    ret = yield from request("http://test.com/foo")
    # 注意 yield from
    ret = yield from func2(ret) 
    return ret

def func2(data):
    # 注意 yield from
    result = yield from request("http://test.com/"+data)
    return result

# 现在 request 函数，不是生成器，它返回future
def request(url):
    # 同上基于future实现的request

g = func1()
# 返回第一个请求的 future 
g.send(None)
# 继续运行，自动进入func2 并得到第它里面的那个future
g.send("bar")
# 继续运行，完成调用链剩余逻辑，抛出StopIteration异常
g.send("barz")

0x06 总结和比较

1. 基于 IO 多路复用技术，让整个应用在 IO 上非阻塞，实现高效率
2. 通过生成器让分散的 callback 代码变成同步代码，减少业务编写困难

来源：https://zhuanlan.zhihu.com/p/330549526

作者：毛豆花生