Staskaer

Hi, this is Staskaer

0%

python协程与asyncio

Posted on Edited on In Valine:

Python中的协程使用

协程

yield与线程

Python中的线程是系统级线程，会被操作系统调度；而协程则可以理解成用户级线程，不会被系统调度。

由于协程(用户级线程)需要用户自己去管理线程的状态(就绪、等待、阻塞)，并执行就绪线程，这需要线程能实现主动挂起的机制(不然就变成线程顺序执行了)。python中的协程其实是通过yield关键字来实现的。

当执行到yield处程序会主动交出控制权，然后等待下一次调用时再继续执行。

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

import time

def work1():
    while 1:
        print("hello")
        yield
        time.sleep(0.5)

def work2():
    while 1:
        print("world")
        yield
        time.sleep(0.5)

task1 = work1()
task2 = work2()

while 1:
    next(task1)
    next(task2)

这样就形成了简单的循环，每次next都会执行到yield处，然后交出控制权，等待下一次调用。而循环则依次执行两个任务。

但是上述代码只体现了yield放权，没有体现出异步，异步需要通过send()函数配合使用

一般都是n = yield m这种形式，m是协程传出去的值，n是协程被唤醒时接受到的值。

如果协程处理计算任务，比如m是需要计算的东西，n是计算完成的东西，那么这个过程就类似函数调用

如果协程是事件驱动的，那么m就是请求事件的信息，n是响应事件

因为yield想必已经很熟了，上面就简单提一下不再赘述。

asyncio库

Python对于协程后续引入了asyncio库，这个库提供了对协程的支持，本着学新不学旧的原则，所以后续主要是针对此库来学习下协程。

先来一个简单的例子

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

import asyncio

# 定义一个协程函数
async def work1():
    while 1:
        print("hello")
        await asyncio.sleep(0.5)

# 协程函数返回协程协程对象
task = work1()

# 运行协程，需要在一个循环里运行
asyncio.run(task)

需要说明几点：

  • async def定义的函数是协程函数，可以使用await关键字。所有async函数返回的都是协程对象，就像所有的yield函数返回的都是迭代器对象一样。
  • asyncio.run()其实建立了一个循环，这个循环会不断寻找哪个协程可以运行，知道所有协程运行完毕。(协程在运行时会主动挂起放权)
  • await后只能接async def定义的函数，这个await会做这几件事情：
    • 运行async def定义的函数(不加await的运行时会返回协程对象，不会真正进入协程中)，直到运行不下去为止(比如yield主动放权、新建了一个task然后本函数运行完了)
    • yield出控制权
    • 再次被唤醒的时候，会把async def定义的函数的返回值返回

await之于asyncio类似与yield from之于生成器

Task和futures

  • 循环最小的调度单位是Task，Task是对协程的封装，因此循环每次只能调度一个Task
  • futures其实是更加底层的东西(Task其实是Future的子类)，futures主要用于接受异步执行的结果

两者都是可以直接放在await后面

对于如下两种形式的代码，结果是一样的，但是过程不一样

1
2
3
4
5
6
7
8
9

import asyncio

async def f1():
    await asyncio.sleep(1)

async def f2():
    await asyncio.creat_task(asyncio.sleep(1))

其中，f1()函数会直接进入asyncio.sleep()函数，不会为其创建任务，在这个函数内某个地方放权。

而f2()函数会为asyncio.sleep()创建一个任务，然后f2()函数等待这个任务完成自己阻塞，在下一次task调度的时候，会去调度就绪的延时任务，直到该延时任务完成才回过头来执行f2()

由于await的时候不会新建一个task，所以可能无法很好的利用协程异步，就拿python官方文档这个例子来说：

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

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')}")

asyncio.run(main())

这个例子中，await say_after(1, 'hello')时，需要阻塞当前协程，进入到asyncio.sleep(1)里，此时没有其他协程可以运行，故空等；同样，await say_after(2, 'world')时也是空等。故两个任务顺序执行，总共耗时3s

上述问题的原因就是在执行第一个await的时候，不知道后面还有一个可以同时等待的任务，所以就阻塞了。

解决的方法也很简单，就是提前在循环里注册所有需要等待的任务，这样就可以同时等待，从而实现异步。

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

import asyncio
import time

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

async def main():
    # 先创建task
    task1 = asyncio.create_task(
        say_after(1, 'hello'))

    task2 = asyncio.create_task(
        say_after(2, 'world'))

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

    # Wait until both tasks are completed (should take
    # around 2 seconds.)
    # 然后等待task完成
    await task1
    await task2

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

asyncio.run(main())

上述代码事先创建了任务，从而实现了异步，总共耗时2s

协程与多线程

由于Python中GIL的存在，python的多线程其实比较假，而协程本质上也是单线程异步，所以很容易将两者比较起来。我来浅浅说一下两者区别和应用。

由于线程是系统调度的，不同线程的任务都能运行，因此不同的任务不会产生饥饿。但是协程需要手动放权，所以如果写不好就容易出现只有一个任务一直在运行，其他任务得不到运行的结果。

但是呢，协程相比于线程，更加轻量级，所以性能上有优势，协程在网络上优势大。且协程是单线程，没有竞争冒险。

另外，假设一个这个场景：

  • 有一个计算密集型任务，需要占用大量CPU资源
  • 不时会出现临时借用小量CPU资源的任务

这个时候使用协程就很麻烦，因为对大任务做拆分需要设置很多放权点，而且放权点位置和数目都要根据大任务而改变。且大任务如果调用了第三方函数计算，那么就很难控制放权点。

这个时候使用多线程就比较好。