高性能编程之协程--------asyncio

Posted pyweb

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了高性能编程之协程--------asyncio相关的知识,希望对你有一定的参考价值。

一、为什么需要协程

在回答这个问题之前,我们先回顾一下同步和异步的概念。

同步:就是发出一个调用时,在没有得到结果之前,该调用就不返回,调用者需要一致等待该调用结束,才能进行下一步工作。
异步:调用在发出去后,不等待结果,直接进行下一步工作,当结果出来后通过状态来通知调用者继续回来处理该调用。

我们先使用普通同步代码实现多个IO任务的案例

# 普通同步代码实现多个IO任务
import time
def taskIO_1():
    print(‘开始运行IO任务1...‘)
    time.sleep(2)  # 假设该任务耗时2s
    print(‘IO任务1已完成,耗时2s‘)
def taskIO_2():
    print(‘开始运行IO任务2...‘)
    time.sleep(3)  # 假设该任务耗时3s
    print(‘IO任务2已完成,耗时3s‘)

start = time.time()
taskIO_1()
taskIO_2()
print(‘所有IO任务总耗时%.5f秒‘ % float(time.time()-start))

执行结果:

开始运行IO任务1...
IO任务1已完成,耗时2s
开始运行IO任务2...
IO任务2已完成,耗时3s
所有IO任务总耗时5.00604秒

如上,实现两个同步IO任务,总耗时是两个IO任务的时间之和5秒。当CPU运算完毕,如果再闲置很长时间去等待IO完成才能进行下一个任务,这样的任务执行效率很低。

所以我们需要有一种异步的方式处理上述任务。步骤就是在上述IO任务执行前终端当前IO任务,进行下一个任务,当该IO任务完成后再唤醒该任务。

而python中生成器的关键词 yiled 就可以实现中断共嗯那个。所以起初,协程是基于生成器的变形进行实现的。

二、使用yield from 和@asyncio.coroutine 实现协程

1、什么是yield from?和yield有什么区别?

我们都知道,yield 在生成器中有中断的功能,可以传出值,也可以从函数外部接收值,而yield from就是简化了yield的操作。
先看个案例:

def generator_1(titles):
    yield titles
def generator_2(titles):
    yield from titles

titles = [‘Python‘,‘Java‘,‘C++‘]
for title in generator_1(titles):
    print(‘生成器1:‘,title)
for title in generator_2(titles):
    print(‘生成器2:‘,title)

执行结果:

生成器1: [‘Python‘, ‘Java‘, ‘C++‘]
生成器2: Python
生成器2: Java
生成器2: C++

这个例子中,yield titles返回了titles完成的列表,而yield from titles等价于:

for title in titles: # 等价于yield from titles
    yield title

同时yield from 内部已经实现了大部分的异常处理。
【举个例子】:下面通过生成器来实现一个整数加和的程序,通过send()函数向生成器中传入要加和的数字,然后最后以返回None结束,total保存最后加和的总数。

def generator_1():
    total = 0
    while True:
        x = yield 
        print(‘加‘,x)
        if not x:
            break
        total += x
    return total
def generator_2(): # 委托生成器
    while True:
        total = yield from generator_1() # 子生成器
        print(‘加和总数是:‘,total)
def main(): # 调用方
    g1 = generator_1()
    g1.send(None)
    g1.send(2)
    g1.send(3)
    g1.send(None)
    # g2 = generator_2()
    # g2.send(None)
    # g2.send(2)
    # g2.send(3)
    # g2.send(None)
    
main()

执行结果,可见对于生成器g1,最后传入None后,程序退出,报StopIteration异常并返回了最后total值是5。

加 2
加 3
加 None
------------------------------------------
StopIteration       
<ipython-input-37-cf298490352b> in main()
---> 19     g1.send(None)
StopIteration: 5

如果把g1.send()那5行注释掉,解注下面的g2.send()代码,则结果如下。可见yield from封装了处理常见异常的代码。对于g2即便传入None也不报异常,其中total = yield from generator_1()返回给total的值是generator_1()最终的return total

加 2
加 3
加 None
加和总数是: 5

借上面的例子,几个重要的概念

  • 子生成器: yield from后的generator_1()生成器函数是子生成器
  • 委托生成器:generator_2()是程序中的委托生成器,它负责委托子生成器完成具体任务。
  • 调用方:main()是程序中的调用方,负责调用委托生成器。
  • 在上述代码中main()每一次在调用send(value)时,value不是传递给了委托生成器generator_2(),而是借助yield from传递给了子生成器generator_1()中的yield
  • 同理,子生成器中的数据也是通过yield直接发送到调用方main()中。

之后我们的代码都是依据 调用方 - 委托生成器 - 子生成器的规范形式书写,

2、结合@asyncio.coroutine实现协程

上面那个同步IO任务的代码中,修改成协程的用法如下:

# 使用同步方式编写异步功能
import time
import asyncio
@asyncio.coroutine # 标志协程的装饰器
def taskIO_1():
    print(‘开始运行IO任务1...‘)
    yield from asyncio.sleep(2)  # 假设该任务耗时2s
    print(‘IO任务1已完成,耗时2s‘)
    
@asyncio.coroutine # 标志协程的装饰器
def taskIO_2():
    print(‘开始运行IO任务2...‘)
    yield from asyncio.sleep(3)  # 假设该任务耗时3s
    print(‘IO任务2已完成,耗时3s‘)
@asyncio.coroutine # 标志协程的装饰器
def main(): # 调用方
    tasks = [taskIO_1(), taskIO_2()]  # 把所有任务添加到task中
    done,pending = yield from asyncio.wait(tasks) # 子生成器
    for r in done: # done和pending都是一个任务,所以返回结果需要逐个调用result()
        print(‘协程无序返回值:‘+r.result())

if __name__ == ‘__main__‘:
    start = time.time()
    loop = asyncio.get_event_loop() # 创建一个事件循环对象loop
    try:
        loop.run_until_complete(main()) # 完成事件循环,直到最后一个任务结束
    finally:
        loop.close() # 结束事件循环
    print(‘所有IO任务总耗时%.5f秒‘ % float(time.time()-start))

执行结果:

开始运行IO任务1...
开始运行IO任务2...
IO任务1已完成,耗时2s
IO任务2已完成,耗时3s
协程无序返回值:taskIO_2
协程无序返回值:taskIO_1
所有IO任务总耗时3.00209秒

使用方法: @asyncio.coroutine装饰器是协程函数的标志,我们需要在每一个任务函数前加这个装饰器,并在函数中使用yield from。在同步IO任务的代码中使用的time.sleep(2)来假设任务执行了2秒。但在协程中yield from后面必须是子生成器函数,而time.sleep()并不是生成器,所以这里需要使用内置模块提供的生成器函数asyncio.sleep()。
功能:通过使用协程,极大增加了多任务执行效率,最后消耗的时间是任务队列中耗时最多的时间。上述例子中的总耗时3秒就是taskIO_2()的耗时时间。
执行过程
1. 上面代码先通过get_event_loop()获取了一个标准事件循环loop(因为是一个,所以协程是单线程)
2. 然后,我们通过run_until_complete(main())来运行协程(此处把调用方协程main()作为参数,调用方负责调用其他委托生成器),run_until_complete的特点就像该函数的名字,直到循环事件的所有事件都处理完才能完整结束。
3. 进入调用方协程,我们把多个任务[taskIO_1()和taskIO_2()]放到一个task列表中,可理解为打包任务。
4. 现在,我们使用asyncio.wait(tasks)来获取一个awaitable objects即可等待对象的集合(此处的aws是协程的列表),并发运行传入的aws,同时通过yield from返回一个包含(done, pending)的元组,done表示已完成的任务列表,pending表示未完成的任务列表;如果使用asyncio.as_completed(tasks)则会按完成顺序生成协程的迭代器(常用于for循环中),因此当你用它迭代时,会尽快得到每个可用的结果。【此外,当轮询到某个事件时(如taskIO_1()),直到遇到该任务中的yield from中断,开始处理下一个事件(如taskIO_2())),当yield from后面的子生成器完成任务时,该事件才再次被唤醒】
5. 因为done里面有我们需要的返回结果,但它目前还是个任务列表,所以要取出返回的结果值,我们遍历它并逐个调用result()取出结果即可。(注:对于asyncio.wait()和asyncio.as_completed()返回的结果均是先完成的任务结果排在前面,所以此时打印出的结果不一定和原始顺序相同,但使用gather()的话可以得到原始顺序的结果集)
6. 最后我们通过loop.close()关闭事件循环。

综上所述:协程的完整实现是靠①事件循环+②协程。

三、使用async 和await 实现协程

在python3.4中使用yield from和@asyncio.coroutine实现协程。
在python3.5开始引入了新的语法 async + await,以简化更好的标识异步IO。
要使用新的语法,只需要两步简单的替换。

  • @asyncio.coroutine替换为async
  • yield from替换为await
    更改上面的代码,可以得到同样的结果:
import time
import asyncio
async def taskIO_1():
    print(‘开始运行IO任务1...‘)
    await asyncio.sleep(2)  # 假设该任务耗时2s
    print(‘IO任务1已完成,耗时2s‘)
    return taskIO_1.__name__
async def taskIO_2():
    print(‘开始运行IO任务2...‘)
    await asyncio.sleep(3)  # 假设该任务耗时3s
    print(‘IO任务2已完成,耗时3s‘)
    return taskIO_2.__name__
async def main(): # 调用方
    tasks = [taskIO_1(), taskIO_2()]  # 把所有任务添加到task中
    done,pending = await asyncio.wait(tasks) # 子生成器
    for r in done: # done和pending都是一个任务,所以返回结果需要逐个调用result()
        print(‘协程无序返回值:‘+r.result())

if __name__ == ‘__main__‘:
    start = time.time()
    loop = asyncio.get_event_loop() # 创建一个事件循环对象loop
    try:
        loop.run_until_complete(main()) # 完成事件循环,直到最后一个任务结束
    finally:
        loop.close() # 结束事件循环
    print(‘所有IO任务总耗时%.5f秒‘ % float(time.time()-start))

四、总结

最后将整个过程串一遍:

【引出问题】:

  1. 同步编程的并发性不高
  2. 多进程编程效率受CPU核数限制,当任务数量远大于CPU核数时,执行效率会降低。
  3. 多线程编程需要线程之间的通信,而且需要锁机制来防止共享变量被不同线程乱改,而且由于Python中的GIL(全局解释器锁),所以实际上也无法做到真正的并行。

【产生需求】:

  1. 可不可以采用同步的方式来编写异步功能代码?
  2. 能不能只用一个单线程就能做到不同任务间的切换?这样就没有了线程切换的时间消耗,也不用使用锁机制来削弱多任务并发效率!
  3. 对于IO密集型任务,可否有更高的处理方式来节省CPU等待时间?

【结果】:
所以协程应运而生。当然,实现协程还有其他方式和函数,以上仅展示了一种较为常见的实现方式。此外,多进程和多线程是内核级别的程序,而协程是函数级别的程序,是可以通过程序员进行调用的。以下是协程特性的总结:

协程 属性
所需线程 单线程(因为仅定义一个loop,所有event均在一个loop中)
编程方式 同步
实现效果 异步
是否需要锁机制
程序级别 函数级
实现机制 事件循环+协程
总耗时 最耗时事件的时间
应用场景 IO密集型任务等

以上是关于高性能编程之协程--------asyncio的主要内容,如果未能解决你的问题,请参考以下文章

Python异步IO之协程:使用asyncio的不同方法实现协程

Python asyncio之协程学习总结

并发编程之协程

你必须要会uvloop!让Python asyncio异步编程性能直逼Go协程性能

“按史索骥”:Python异步编程之协程进化史

重修课程day35(网络编程九之协程)