asyncio

Python

什么是coroutine

说到 coroutine 一般会指的是两个东西 coroutine function 和 coroutine object 在以下代码中，async def main 就是一个 coroutine function，它定义了一个 coroutine 的过程

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


async def main():
    print('hello')
    await asyncio.sleep(1)
    print('world')


coro = main()

asyncio.run(coro)

当调用一个 coroutine function 时，返回的是一个 coroutine object，并不会运行 coroutine function 里面的代码。

如果要运行一个 coroutine function 里面的代码，需要做以下的事

  1. 进入 async 模式，也就是 event loop控制整个程序的状态
  2. 把 coroutine 变成task

如何进入async模式

我们把正常代码的模式称为 sync 模式，从 sync 模式切换到 async 模式，切换到让 event loop 控制一切，基本上只用一个入口函数 asyncio.run

asyncio.run的参数是一个 coroutine， asyncio.run 的调用会做两件事情

  1. 建立起event loop
  2. 将传给他的 coroutine 变成 建立的 event loop 里面的第一个 task

当 event loop 建立后，他就会开始寻找有哪些 task 可以执行，此时 event loop 里面只有 刚创建出来的 task，event loop 开始执行这个 task

尝试运行上面的**创建coro **代码块，输出如下

1
2
3
4
5
python async_example.py
hello
# 等待了一秒
world

event loop 调度多个 task

event loop 的核心就是有多个 task，由他来决定接下来运行哪个task，接下来尝试往 event loop 里面加入多个 task

如何运行一个coroutine object

使用 await

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


async def say_after(delay, what):
    await asyncio.sleep(delay)
    print(what)


async def main():
    print(f"started at {time.strftime('%X')}")
    await say_after(1, 'hello')
    await say_after(2, 'world')

    print(f"finished at {time.strftime('%X')}")


coro = main()

asyncio.run(coro)

在上面的代码块中，say_after 函数是一个 coroutine function ，asyncio.sleep 也是一个 coroutine function ，调用 say_after 会返回一个 coroutine，但是里面的代码不会被执行，对coroutine 使用await 发生了如下的事情

  1. 调用 coroutine，返回一个 future
  2. await 这个future，提取返回值

await future是一个同步的操作，等待future的返回后才继续往下执行

尝试运行以上代码，输出结果应该如下

1
2
3
4
5
6
python async_example2.py
started at 13:16:06
hello
world
finished at 13:16:09

运行的整个过程如下

  1. 将main作为一个task，放到event loop，event loop 运行这个task
  2. print started at
  3. 运行 say_after(1, ‘hello’) 这个 coroutine function 得到 coroutine object，await这个 coroutine object，得到future ( say_after1)
  4. await future ( say_after1)运行，得到返回值
  5. 等待1秒，print(‘hello’)，future ( say_after1)完成
  6. 运行 say_after(2, ‘world’) 这个 coroutine function 得到 coroutine object，await这个 coroutine object，得到future ( say_after2)
  7. await future ( say_after2)运行，得到返回值
  8. 等待2秒，print(‘world’)，future ( say_after1)完成
  9. print finished at

从上面的过程中可以看出， say_after(1, ‘hello’) say_after(2, ‘world’) 是同步的，一共sleep了3秒，那有没有办法能让他们一共sleep 2秒，也就是说，让他们一起等

在运行过程中，我们可以发现，如果要让他们一起等，那就要让 event loop 一起发现这2个task

使用 create_task

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
import asyncio
import time


async def say_after(delay, what):
    await asyncio.sleep(delay)
    return f"{what} - {delay}"


async def main():
    print(f"started at {time.strftime('%X')}")

    task1 = asyncio.create_task(
        say_after(1, 'hello')
    )
    task2 = asyncio.create_task(
        say_after(2, 'world')
    )
    ret1 = await task1
    ret2 = await task2 # 去掉的话会这么样
    print(ret1)
    print(ret2) # 去掉的话会这么样

    print(f"finished at {time.strftime('%X')}")


coro = main()

asyncio.run(coro)

使用 create_task，发生了如下的事情

  1. 将一个 coroutine object 变成 task
  2. 将task注册给event loop

await 一个task，与await 一个 coroutine object 不同，他执行如下步骤

  1. 告诉event loop，当前的task 需要 依赖这个task完成，完成后提取返回值
  2. 交还控制权给event loop

运行过程如下

  1. 将main作为一个task，放到event loop，event loop 运行这个task
  2. print start at
  3. 运行 say_after(1, ‘hello’) 这个 coroutine function 得到 coroutine object，使用 create_task 将 coroutine object 变成 task，并且将task注册到event loop
  4. 运行 say_after(2, ‘world’) 这个 coroutine function 得到 coroutine object，使用 create_task 将 coroutine object 变成 task，并且将task注册到event loop
  5. await task1，告诉event loop，当前的的task(main) 需要等待 task1 完成
  6. await task2，告诉event loop，当前的的task(main) 需要等待 task2 完成
  7. event loop 检查task，有3个task，[main, task1, task2]，main需要等待 task1，task2 完成，尝试运行task1
  8. task2 运行完成，将控制权交给event loop
  9. event loop 检查task，有2个task，[main, task2]，main需要等待 ，task2 完成，尝试运行task2
  10. task2 运行完成，将控制权交给event loop
  11. event loop 检查task，有1个task，[main]，尝试运行main
  12. main 运行完成
  13. 退出

在整个过程中控制权的交回是显式的，也就是说，event loop不能强行从task拿回控制权，所以如果task中有死循环，整个event loop 就卡死了，交回控制权的方式有两种

  1. await 一个 coroutine object
  2. 函数完成 return

运行结果如下

1
2
3
4
5
6
python async_example3.py
started at 16:15:25
hello - 1
world - 2
finished at 16:15:27

如果不 await task2 ，虽然 task2 已经加入了event loop，但是在 task1 结束后 main task 也结束了，所以没有机会执行到 task2 的输出

gather

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import asyncio
import time


async def say_after(delay, what):
    await asyncio.sleep(delay)
    return f"{what} - {delay}"


async def main():
    print(f"started at {time.strftime('%X')}")

    ret = await asyncio.gather(
        say_after(1, 'hello'),
        say_after(2, 'world')
    )
    print(ret)
    print(f"finished at {time.strftime('%X')}")


coro = main()

asyncio.run(coro)

为了方便，当需要一起创建多个task时，可以使用gather，gather的返回时是future

gather 参数可以是task，future也可以是coroutine object，当参数是coroutine object时，他会把coroutine object变成task

对gather await，等待会所有的task都完成，把他们的返回值放入一个list，list中值的顺序就是task的顺序

所以以上代码块的输出如下

1
2
3
4
5
python async_example4.py
started at 14:00:49
['hello - 1', 'world - 2']
finished at 14:00:51

await coro与await task的区别

使用 await 一个 coroutine object 会直接执行这个coroutine ，而不是把控制权交给event loop 

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import asyncio


async def coro():
    asyncio.sleep(1)
    return 1
    
    
async def get_1(key):
   value = await coro()
   return value

async def get_2(key):
   value = await asyncio.ensure_future(coro())
   # value = await asyncio.create_task(coro())
   return value

get_1中的形式，不会把控制权交还给event loop

总结

  1. evnet loop控制task执行，task没办法控制event loop接下来执行哪个task，只能告诉event loop我需要等待哪个 task 完成
  2. coroutine object 是变成 task 才能运行的
  3. task 一旦开始运行，就需要显式的把任务交还给 event loop

所以同时刻实际上只有一段代码在运行，他只是想办法利用代码中间的等待时间，所以如果代码中没有等待，协程对代码是没有帮助的