python async/coroutine 异步/协程 研学笔记

• python async/coroutine 异步/协程 研学笔记
• • 前言
  • • 免责声明
    • 目的
    • 前置知识
    • 备注
  • 引入包
  • 测试-1
  • 测试-2
  • 测试-3
  • • 工具函数
    • 解释
  • 结论

本笔记完全基于 Allen Lv 的粗浅认知，不对正确性做保证，请读者自己实践来确认。

研究异步的基本使用方式。

1. 异步 (并发) 的本质是 单线程多协程。

   • 举例: 有 2 个任务: (1) 洗衣机洗衣服，然后晾衣服; (2) 吃饭，然后洗碗机洗碗。

   • 解决方案: 先启动洗衣机洗衣服，然后在等洗衣机的这段时间内吃饭，吃完饭把碗放到洗碗机，在等洗碗机的时间内晾衣服，等晾完衣服碗也洗好了。

2. 特征

   1. 一个事件循环 (只有 1 个人)。
   2. 多个包含非阻塞操作 (洗衣服/洗碗虽然要等待，但你没必要干等) 的任务。

3. 概念

   • 协程

     • 一个允许中途退出，之后再从退出点进入的函数，类似于一个生成器。
     • 打个比方：允许干活干到一半人跑了，过会再回来接着干 (而不用从头干)。

   • 事件循环

     1. 事件循环 (任务队列) 中存在多个协程 (任务)。
     2. 首先开始执行第一个协程。
     3. 如果遇到需要等待的操作，那就在这留下记号，等会再来看看等待结束了没有。
     4. 现在离开这个需要等待的协程，回到事件循环中，去执行下一个协程。
     5. 如果在接下来的协程执行中遇到了需要等待的操作，就依照此办法，先留个记号然后离开。
     6. 直到再次轮到之前没干完活的协程，从记号处进入，看看有没有等完。
     • 如果等完了，那就接着执行后面的代码，如果再遇到等待操作，如法炮制。
     • 如果还没等完，那就回到事件循环，进入并执行下一个协程。
     7. 重复上述操作直到所有协程完成。

请勿在 IPython 或 Jupyter Notebook 环境使用这些代码。如果你需要，请参照:

https://stackoverflow.com/questions/55409641/asyncio-run-cannot-be-called-from-a-running-event-loop

# coding=utf-8
import asyncio
import time
from asyncio import sleep, all_tasks

引入 asyncio 库用于

1. 执行协程
2. 模拟非阻塞IO等待

引入 time 库用于统计不同策略的耗时

async def factorial(n):
    product = 1
    for i in range(1, n+1):
        product *= i
        await asleep(1)
    return product


def countTasks(name, i):
    print(f"{name}-{i}: len(tasks) == {len(all_tasks())}")

    
async def test1():
    for i in range(1, 10):
        await factorial(i)
        countTasks("test1", i)


def start(test):
    begin = time.time()
    asyncio.run(test())
    end = time.time()
    cost = end - begin
    
    print(f"cost: {cost:.4f}s")


start(test1)

预期输出

test1-1: len(tasks) == 1
test1-2: len(tasks) == 1
test1-3: len(tasks) == 1
test1-4: len(tasks) == 1
test1-5: len(tasks) == 1
test1-6: len(tasks) == 1
test1-7: len(tasks) == 1
test1-8: len(tasks) == 1
test1-9: len(tasks) == 1
cost: 45.4508s

1. factorial 计算阶乘，在计算时会进行 非阻塞等待，用于模拟网络请求的等待。
2. countTasks 打印当前事件循环中的任务 (协程) 数量。
3. start 执行目标异步函数，并统计耗时。该异步函数应当是入口函数，它包含 (将会调用) 所有需要的协程，或者说事件循环上的全部任务。

理解：耗时约 45 秒，低于同步的 (本文未给出)，但有优化空间。

async def getProduct(i):
    return await factorial(i)


async def test2():
    tasks = []
    for i in range(1, 10):
        tasks.append(getProduct(i))
        countTasks("test2", i)
    await asyncio.gather(*tasks)
    

start(test2)

预计输出

test2-1: len(tasks) == 1
test2-2: len(tasks) == 1
test2-3: len(tasks) == 1
test2-4: len(tasks) == 1
test2-5: len(tasks) == 1
test2-6: len(tasks) == 1
test2-7: len(tasks) == 1
test2-8: len(tasks) == 1
test2-9: len(tasks) == 1
cost: 9.0746s

getProdcut 确保阶乘计算完毕，然后返回其结果。

理解：耗时约 9 秒，低于 test1。

推测的原因：

1. 先都把这 10 次任务加到事件循环中 但 先不执行它们中的任何一个。
2. 等 10 个任务都加到循环里了，再去执行它们。
3. 这时候 (开始执行这 10 个阶乘任务的时候) 无论谁需要等待了，总有别人有活干，可以不闲着。

而对比 test1，它一边把阶乘任务加到事件循环，一边又立刻开始做，然后又因为等待而做不下去，开始找别的活干，可这时候 (循环没完成时) 加入循环的任务总是少于全部 10 个的，也就是说，活少，空等的时间等多。

具体实现：

1. 生成任务列表。
2. 生成协程。
3. 添加该协程到列表。
4. 直到 10 个任务均添加完毕。
5. 执行并等待列表中的每个任务。

但这个写法有个麻烦的地方，就是需要手动维护一个 任务列表，即 tasks，最后一起 (并发地) 执行它们。

所以有了 test3，就是为了简化写法而设计的。

def loop():
    try:
        return asyncio.get_running_loop()
    except RuntimeError:
        return asyncio.get_event_loop()


def addTask(coro):
    return loop().create_task(coro)


async def loopDone():
    tasks = asyncio.all_tasks()
    for task in tasks:
        if not (task is asyncio.current_task()):
            await task

            
async def test3():
    for i in range(1, 10):
        addTask(getProduct(i))
        countTasks("test3", i)
    await loopDone()
    

start(test3)

预期输出

test3-1: len(tasks) == 2
test3-2: len(tasks) == 3
test3-3: len(tasks) == 4
test3-4: len(tasks) == 5
test3-5: len(tasks) == 6
test3-6: len(tasks) == 7
test3-7: len(tasks) == 8
test3-8: len(tasks) == 9
test3-9: len(tasks) == 10
cost: 9.0821s

1. loop 用于取得当前运行着的 事件循环。

2. 如果在 协程 或 (协程的)回调函数 中使用，说明当前肯定运行着一个事件循环，返回该事件循环。

3. 如果当前线程中不存在事件循环，则取得 get_event_loop 立刻设置的事件循环。

4. addTask 用于添加一个协程到事件循环中。这 类似 于 启动 (或者说调用) 该协程。

   1. 说 类似，是因为该协程被添加到事件循环中了。这意味着当事件循环轮到该协程时 (该协程拿到 控制权 了)，就会进入协程函数里执行其内部代码。【轮到该协程运行其代码】是 预期会自动发生的。
   2. 不说 等于，是因为这种启动不是立刻发生的。【不立刻】是因为【协程被添加到事件循环】的操作本身就处于一个协程中。非标准的，我将其称之为 父协程。考虑到该操作是一个 同步的 操作，所以不会移交该父协程正享有的控制权给事件循环 (让它去找下家)。

5. loopDone 用于等价于 async main 函数的尾部使用，用来确保事件循环中的所有协程都被执行完毕才退出 main。那行 if not 是必要的，因为不可以自己等待自己。这会造成无限的递归：自己等待 (自己等待 (自己等待 (...)))。

从效果上看 test3 与 test2 一致，就是先把任务加到事件循环中而不立即处理，之后再统一处理。

具体实现：

1. 生成协程。
2. 取得当前事件循环，添加该协程到事件循环。
3. 直到 10 个任务均添加完毕。
4. 执行并等待事件循环中的每个任务。

相比于 test2 不需要自己手动维护一个任务列表，不需要考虑缺漏任务。

使用异步可以有效提高多个非阻塞等待任务的执行效率，而在此基础上再优化的方式是尽早把所有任务加入事件循环，让线程尽可能少的闲着干等。