协程(Coroutine)

也可以被称为微线程,是一种用户态内的上下文切换技术。简而言之,其实就是通过一个线程实现代码块相互切换执行。

说协程之前,先用一张表来说明进程、线程、协程的异同

进程线程协程
定义系统进行资源分配和调度的独立单位是进程的一个实体,是cpu调度和分配的基本单位微线程,比线程小,占用更少,cpu其实不知道协程的存在
功能完成多任务,例如一台电脑同时运行多个QQ,多用于计算密集型完成多任务,例如一个QQ开多个窗口,多用于IO密集型完成多任务,一任务执行过程中会出现等待,此时通过协程去执行其他任务。多用于IO密集型
理解一条流水线,完成资源分配流水线的工人,实现对cpu的带哦度分配一个工人在等待的时候去做别的事
区别资源占用大,能充分利用多核cpu。执行效率高,切换效率低。执行无序,不共享全局变量资源占用一般。执行效率一般,切换效率一般。共享全局变量,执行无序资源占用最少,切换效率高,可以确定谁先执行
联系一个进程必须有一个线程线程必须依赖于一个进程协程必须依赖于一个线程
优点进程最稳定,一个进程死了不会影响其他进程。一个进程可以独立占用一个cpu核心数据共享时很方便资源占用少,切换效率高
缺点切换效率低,资源占用大线程不够稳定,一个线程死了可能会导致进程死了,进而导致进程里所有线程都死了。
python里还存在GIL锁,执行过程会产生资源竞争
协程必须在一个线程里,所有不管怎么切换,同一时间也只能占用一个cpu核心
进程、线程、协程

直接上代码,例如:

同步编程

import time

def func1():
    print(1)
    time.sleep(2)  # 代表耗时操作
    print(2)

def func2():
    print(3)
    time.sleep(2)  # 代表耗时操作
    print(4)

def main():
    st = time.time()
    func1()
    func2()
    print("总耗时:", time.time() - st)

if __name__ == '__main__':
    main()

"""
输出如下:
1
2
3
4
总耗时: 4.008249998092651
"""

上述代码是普通的函数定义和执行,按流程分别执行两个函数中的代码,并先后会输出:1、2、3、4。并且打印1之后,堵塞2s,打印3之后,堵塞2s,总的耗时大概4s

异步编程

但如果介入协程技术,就可以实现函数见代码切换执行,最终输入:1、3、2、4 。总耗时大概2s

import time
import asyncio

async def func1():
    print(1)
    await asyncio.sleep(2)  # 代表耗时操作
    print(2)

async def func2():
    print(3)
    await asyncio.sleep(2)  # 代表耗时操作
    print(4)

def main():
    st = time.time()
    tasks = [
        asyncio.ensure_future(func1()),
        asyncio.ensure_future(func2())
    ]
    loop = asyncio.get_event_loop()
   # wait函数内部用set()做了去重的处理,所以执行就不能保证先后顺序
    loop.run_until_complete(asyncio.wait(tasks))

   # 可以使用gather保证遇到耗时操作前的函数执行先后顺序
   # loop.run_until_complete(asyncio.gather(*tasks))

    print("总耗时:", time.time() - st)

if __name__ == '__main__':
    main()
"""
输出如下:
1
3
2
4
总耗时: 2.005261182785034
"""

通过上面的例子,应该能看出异步编程的好处了吧,能让CPU充分利用IO的耗时等待时间。

协程的意义

  • 计算型的操作,利用协程来回切换执行,没有任何意义,来回切换并保存状态,反倒会降低性能。
  • IO型的操作,利用协程在IO等待时间就去切换执行其他任务,当IO操作结束后再自动回调,那么就会大大节省资源并提供性能,从而实现异步编程(不等待任务结束就可以去执行其他代码)。

耗时的操作一般是一些IO操作,例如网络请求,文件读取等。

一、协程的实现

在Python中有多种方式可以实现协程,例如:

  • greenlet,是一个第三方模块,用于实现协程代码(Gevent协程就是基于greenlet实现)
  • yield,生成器,借助生成器的特点也可以实现协程代码。
  • asyncio,在Python3.4中引入的模块用于编写协程代码。
  • async & awiat,在Python3.5中引入的两个关键字,结合asyncio模块可以更方便的编写协程代码。

1.1 greenlet

greenlet是python里一个半自动的协程切换的库,但是封装难度大,日常编码一般不用

from greenlet import greenlet

def func1():
    print(1)        # 第2步:输出 1
    gr2.switch()    # 第3步:切换到 func2 函数
    print(2)        # 第6步:输出 2
    gr2.switch()    # 第7步:切换到 func2 函数,从上一次执行的位置继续向后执行

def func2():
    print(3)        # 第4步:输出 3
    gr1.switch()    # 第5步:切换到 func1 函数,从上一次执行的位置继续向后执行
    print(4)        # 第8步:输出 4

gr1 = greenlet(func1)
gr2 = greenlet(func2)
gr1.switch() # 第1步:去执行 func1 函数
"""
打印如下:
1 3 2 4
"""

switch中也可以传递参数用于在切换执行时相互传递值。

1.2 yield

基于Python的生成器的yield和yield form关键字实现协程代码。(tips:yield form关键字是在Python3.3中引入的。)

def func1():
    yield 1
    yield from func2()
    yield 2

def func2():
    yield 3
    yield 4

f1 = func1()
for item in f1:
    print(item)
"""
打印如下:
1 3 4 2
"""

1.3 asyncio

在Python3.4之前官方未提供协程的类库,一般大家都是使用greenlet等其他来实现。在Python3.4发布后官方正式支持协程,即:asyncio模块。

import asyncio

@asyncio.coroutine
def func1():
    print(1)
    yield from asyncio.sleep(2)  # 遇到IO耗时操作,自动化切换到tasks中的其他任务
    print(2)

@asyncio.coroutine
def func2():
    print(3)
    yield from asyncio.sleep(2) # 遇到IO耗时操作,自动化切换到tasks中的其他任务
    print(4)

tasks = [
    asyncio.ensure_future( func1() ),
    asyncio.ensure_future( func2() )
]

loop = asyncio.get_event_loop()
loop.run_until_complete(asyncio.wait(tasks))

1.4 async & awit

async & awit 关键字在Python3.5版本中正式引入,基于他编写的协程代码其实就是 上一示例 的加强版,让代码可以更加简便,代码例子可以参考上面的异步编程的代码

1.5 协程总结

  • 协程可以通过一个线程在多个上下文中进行来回切换执行,让CPU能利用IO等待的时间
  • 协程一般应用在有IO操作的程序中
  • Python3.8之后 @asyncio.coroutine 装饰器就会被移除,推荐使用 async & awit 关键字实现协程代码。

为了更好的理解,用生活中的例子来打个比方。如果把CPU看做一个领导,执行IO耗时操作的设备看做是员工的话,那么同步编程就是 领导 向 A员工 发布命令,需要等待 A员工 处理完成,才能再向 B员工 发布命令。

而使用了协程的异步编程,领导(CPU)向 A员工 发布命令,在等待 A员工做完的时间内,领导就可以继续向 B员工,C员工…发布命令,然后等到 A员工 做完,领导再过来验收即可,这样所有的员工的处理过程就可以几乎同时进行,因为领导发布命令(CPU发布指令)是很快的。

二、异步编程

基于async & await关键字的协程可以实现异步编程,这也是目前python异步相关的主流技术。

2.1 事件循环

事件循环,可以把他当做是一个while循环,这个while循环在周期性的运行并执行一些任务,在特定条件下终止循环。

# 伪代码

任务列表 = [ 任务1, 任务2, 任务3,... ]

while True:
    可执行的任务列表,已完成的任务列表 = 去任务列表中检查所有的任务,将'可执行'和'已完成'的任务返回
    
    for 就绪任务 in 已准备就绪的任务列表:
        执行已就绪的任务
        
    for 已完成的任务 in 已完成的任务列表:
        在任务列表中移除 已完成的任务

	如果 任务列表 中的任务都已完成,则终止循环

# 在编写程序时候可以通过如下代码来获取和创建事件循环。
import asyncio
loop = asyncio.get_event_loop()

2.2 协程和异步编程

协程函数,定义形式为 async def 的函数。

协程对象,调用 协程函数 所返回的对象。

2.2.1 基本应用

程序中,如果想要执行协程函数的内部代码,需要 事件循环协程对象 配合才能实现,如:

import asyncio

#加了async 就是协程函数
async def func():
    print("协程内部代码")

# 调用协程函数,返回一个协程对象。
result = func()  # 协程对象不会执行,需要用事件循环来开启它

# 方式一
# loop = asyncio.get_event_loop() # 创建一个事件循环
# loop.run_until_complete(result) # 将协程当做任务提交到事件循环的任务列表中,协程执行完成之后终止。

# 方式二
# 本质上方式一是一样的,内部先 创建事件循环 然后执行 run_until_complete,一个简便的写法。
# asyncio.run 函数在 Python 3.7 中加入 asyncio 模块,
asyncio.run(result)

这个过程可以简单理解为:将 协程 当做任务添加到 事件循环 的任务列表,然后事件循环检测列表中的 协程 是否 已准备就绪(默认可理解为就绪状态),如果准备就绪则执行其内部代码。

2.2.2 await

await是一个只能在协程函数中使用的关键字,用于遇到IO操作时挂起 当前协程(任务),当前协程(任务)挂起过程中 事件循环可以去执行其他的协程(任务),当前协程IO处理完成时,可以再次切换回来执行await之后的代码。代码如下:

import asyncio


async def others():
    print("start")
    await asyncio.sleep(2)
    print('end')
    return '返回值'


async def func():
    print("执行协程函数内部代码")

    # 遇到IO操作挂起当前协程(任务),等IO操作完成之后再继续往下执行。当前协程挂起时,事件循环可以去执行其他协程(任务)。
    response1 = await others()
    print("IO请求结束,结果为:", response1)
    
    response2 = await others()
    print("IO请求结束,结果为:", response2)
    
asyncio.run(func())
""" 打印如下:
执行协程函数内部代码
start
end
IO请求结束,结果为: 返回值
start
end
IO请求结束,结果为: 返回值
"""

上述的所有示例都只是创建了一个任务,即:事件循环的任务列表中只有一个任务,所以在IO等待时无法演示切换到其他任务效果。

在程序想要创建多个任务对象,需要使用Task对象来实现。

2.2.3 Task对象

—-Tasks are used to schedule coroutines concurrently.
When a coroutine is wrapped into a Task with functions like asyncio.create_task() the coroutine is automatically scheduled to run soon。

Tasks用于并发调度协程,通过 asyncio.create_task(协程对象) 的方式创建 Task 对象,这样可以让协程加入事件循环中等待被调度执行。除了使用 asyncio.create_task() 函数以外,还可以用低层级的 loop.create_task()ensure_future() 函数。不建议手动实例化 Task 对象。

注意:asyncio.create_task() 函数在 Python 3.7 中被加入。在 Python 3.7 之前,可以改用低层级的 asyncio.ensure_future() 函数。

import asyncio

async def func():
    print(1)
    await asyncio.sleep(2)
    print(2)
    return "返回值"

async def main():
    print("main开始")

    # 创建协程,将协程封装到Task对象中并添加到事件循环的任务列表中,等待事件循环去执行(默认是就绪状态)。
    # 在调用
    task_list = [
        asyncio.create_task(func(), name="n1"),
        asyncio.create_task(func(), name="n2")
    ]

    print("main结束")

    # 当执行某协程遇到IO操作时,会自动化切换执行其他任务。
    # 此处的await是等待所有协程执行完毕,并将所有协程的返回值保存到done
    # 如果设置了timeout值,则意味着此处最多等待的秒,完成的协程返回值写入到done中,未完成则写到pending中。
    done, pending = await asyncio.wait(task_list, timeout=None)
    print(done, pending)

#run()是py3.7的写法,会先创建loop,再开启协程函数
asyncio.run(main())

2.2.4 asyncio.Future对象

—- A Futureis a special low-level awaitable object that represents an eventual result of an asynchronous operation.

asyncio中的Future对象是一个相对更偏向底层的可对象,通常我们不会直接用到这个对象,而是直接使用Task对象来完成任务的并和状态的追踪。( Task 是 Futrue的子类 )

Future为我们提供了异步编程中的 最终结果 的处理(Task类也具备状态处理的功能)。

示例1

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # # 创建一个任务(Future对象),这个任务什么都不干。
    fut = loop.create_future()

    # 等待任务最终结果(Future对象),没有结果则会一直等下去。
    await fut

asyncio.run(main())

示例2

import asyncio

async def set_after(fut):
    await asyncio.sleep(2)
    fut.set_result("666")

async def main():
    # 获取当前事件循环
    loop = asyncio.get_running_loop()

    # 创建一个任务(Future对象),没绑定任何行为,则这个任务永远不知道什么时候结束。
    fut = loop.create_future()

    # 创建一个任务(Task对象),绑定了set_after函数,函数内部在2s之后,会给fut赋值。
    # 即手动设置future任务的最终结果,那么fut就可以结束了。
    await loop.create_task(set_after(fut))

    # 等待 Future对象获取 最终结果,否则一直等下去
    data = await fut
    print(data)

asyncio.run(main())

Future对象本身函数进行绑定,所以想要让事件循环获取Future的结果,则需要手动设置。而Task对象继承了Future对象,其实就对Future进行扩展,他可以实现在对应绑定的函数执行完成之后,自动执行set_result,从而实现自动结束

注:支持 await 对象语 法的对象课成为可等待对象,所以 协程对象Task对象Future对象 都可以被成为可等待对象。

2.2.5 concurrent.futures.Future

在Python的concurrent.futures模块中也有一个Future对象,这个对象是基于线程池和进程池实现异步操作时使用的对象。

import time
from concurrent.futures import Future
from concurrent.futures.thread import ThreadPoolExecutor
from concurrent.futures.process import ProcessPoolExecutor

def func(value):
    time.sleep(1)
    print(value)

pool = ThreadPoolExecutor(max_workers=5)
# 或 pool = ProcessPoolExecutor(max_workers=5)

for i in range(10):
    fut = pool.submit(func, i)  # 这里返回的是一个Future对象
    print(fut)

这个 Future对象 和 asyncio.Future 是不同的,他们是为不同的应用场景而设计,例如:concurrent.futures.Future不支持await语法,不接收 timeout参数等。

其实,一般在程序开发中我们要么统一使用 asycio 的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果 协程的异步进程池/线程池的异步 混搭时,那么需要使用 asynic.wrap_future 了。

2.2.5 异步上下文管理器

此种对象通过定义 __aenter__()__aexit__() 方法来对 async with 语句中的环境进行控制。

import asyncio

class AsyncContextManager:
	def __init__(self):
        self.conn = conn
        
    async def do_something(self):
        # 异步操作数据库
        return 666

    async def __aenter__(self):
        # 异步链接数据库
        self.conn = await asyncio.sleep(1)
        return self

    async def __aexit__(self, exc_type, exc, tb):
        # 异步关闭数据库链接
		await asyncio.sleep(1)

async def func():
    async with AsyncContextManager() as f:
        result = await f.do_something()
        print(result)

asyncio.run(func())

这个异步的上下文管理器还是比较有用的,平时在开发过程中 打开、处理、关闭 操作时,就可以用这种方式来处理。

uvloop

Python标准库中提供了asyncio模块,用于支持基于协程的异步编程。

uvloop是 asyncio 中的事件循环的替代方案,替换后可以使得asyncio性能提高。事实上,uvloop要比nodejs、gevent等其他python异步框架至少要快2倍,性能可以比肩Go语言。

# 安装
pip3 install uvloop

在项目中想要使用uvloop替换asyncio的ioloop也非常简单,只要在代码中这么做就行。

import asyncio
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# 编写asyncio的代码,与之前写的代码一致。

# 内部的事件循环自动化会变为uvloop
asyncio.run(...)

tonado用的事件循环是asyncio的ioloop,而像sanic、FastAPI更高性能的异步框架都采用了uvloop