Python协程

zy010101

发布于 2022-05-05 15:58:42

1.1K0

文章被收录于专栏：程序员程序员

Python协程

创建协程函数

Python3.5引入了关键字async来定义协程函数

async def fun():
    """协程函数"""
    print(1)

协程函数和普通的函数不一样，不能直接执行。必须将协程对象（函数）放入事件循环中来执行。在Python3.4的时候，引入内置模块asyncio，该模块可以将协程对象加入到事件循环中执行。

import asyncio

from asyncio.events import get_event_loop

async def fun():    
    """协程函数"""
    print(1)

xc = fun()              # 生成一个协程对象，而非调用协程函数。

# 执行协程函数，必须将协程对象放入事件循环之中。

# Python3.7之前运行协程的方式
# loop = asyncio.get_event_loop()     # 获取一个事件循环
# loop.run_until_complete(xc)         # 将协程对象放入任务列表

# Python3.7之后，可以使用下面的方式运行协程函数。
asyncio.run(xc)

await

await也是Python3.5引入的新关键字。await的作用就是等待可等待对象。可等待对象包含协程对象，future对象，task对象。例如：

import asyncio

async def func1():
    print("start")
    await func2()     # 等待协程对象
    print("end")


async def func2():
    i = 0
    while True:
        if i < 10:
            await asyncio.sleep(1)
            print(i)
            i += 1
        else:
            break

asyncio.run(func1())

程序执行结果如下所示：

start
0
1
2
3
4
5
6
7
8
9
end

我们可以观察到，首先输出start，说明协程func1被执行，然后遇到await，转去执行协程函数func2，func2中有一个等待。可以观察到是每隔1s输出一个数字，等待func2执行完毕，然后func1接着在未执行完的地方继续执行。

task对象

如果我们希望这两个协程函数可以在await的时候，切换到另外一个协程函数继续执行，而不是等着。那么就需要task。例如：

import asyncio

async def func1():
    print("start")
    await asyncio.sleep(3)      # 注意，这里不在是等待func2对象，而是睡眠3s
    print("end")


async def func2():
    i = 0
    while True:
        if i < 10:
            await asyncio.sleep(1)
            print(i)
            i += 1
        else:
            break


task_list = [func1(), func2()]      # 构造一个任务列表
t1 = asyncio.wait(task_list)        # 构造成为task对象
asyncio.run(t1)                     # 执行

程序执行结果如下所示：

start
0
1
end
2
3
4
5
6
7
8
9

观察输出结果，可以看到，func1和func2是交替执行的。也就是说将多个协程放入同一个事件循环中，当一个协程执行到await的时候，会自动切到另一个协程执行。

asyncio.wait源码内部会对列表中的每个协程执行ensure_future从而封装为Task对象。

除了上面的方式之外，还可以通过asyncio.create_task(协程对象)的方式创建Task对象。asyncio.create_task() 函数在 Python 3.7 中被加入。在 Python 3.7 之前，可以改用低层级的 asyncio.ensure_future() 函数。

import asyncio


async def func1():
    print("start")
    await asyncio.sleep(3)      # 注意，这里不在是等待func2对象，而是睡眠3s
    print("end")


async def func2():
    i = 0
    while True:
        if i < 10:
            await asyncio.sleep(1)
            print(i)
            i += 1
        else:
            break


async def main():

    # 将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    c1 = asyncio.create_task(func1(), name='c1')
    c2 = asyncio.create_task(func2(), name='c2')

    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待相对应的协程全都执行完毕并获取结果，主要是为了方便观察。
    await c1    # 等待task对象
    await c2


asyncio.run(main())                     # 执行

这里的执行过程是非常有意思的。

mian()执行到await c1的时候，等待func1执行；
func1执行到await asyncio.sleep(3)时候，会自动跳转到事件循环中的其它协程函数，这里就是func2；
然后func2执行3次await asyncio.sleep(1)，等待3s，这时候func1等待结束，打印end
await c1返回，接着执行await c2; 而func2协程已经在执行了。所以会接着从2开始打印。直到func2执行结束。本次整个事件循环执行结束。

为什么要把asyncio.create_task封装在另一个协程函数内？ 这是因为asyncio.create_task将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，如果不封装在另一个协程函数内，直接执行asyncio.create_task，由于此时还未执行asyncio.run，事件循环不存在，将会导致错误。

一次事件循环中，每个协程只会被执行一次，协程遇到await将会阻塞，这时事件循环机制会调用其它的协程去执行。

由于func2的执行的时间要比func1长。因此，我们只等待func2和上面的效果将是一样的。

import asyncio


async def func1():
    print("start")
    await asyncio.sleep(3)      # 注意，这里不在是等待func2对象，而是睡眠3s
    print("end")


async def func2():
    i = 0
    while True:
        if i < 10:
            await asyncio.sleep(1)
            print(i)
            i += 1
        else:
            break



async def main():

    # 将协程封装到一个Task对象中并立即添加到事件循环的任务列表中，等待事件循环去执行（默认是就绪状态）。
    c1 = asyncio.create_task(func1(), name='c1')
    c2 = asyncio.create_task(func2(), name='c2')

    # 当执行某协程遇到IO操作时，会自动化切换执行其他任务。
    # 此处的await是等待相对应的协程全都执行完毕并获取结果，主要是为了方便观察。
    # await c1  
    await c2        # 只等待c2


asyncio.run(main())                     # 执行

执行结果如下所示：

start
0
1
end
2
3
4
5
6
7
8
9

可以看到执行结果是一模一样的。

还有一种看起来介于上面两者方式之间的方法。

import asyncio


async def func1():
    print("start")
    await asyncio.sleep(3)      # 注意，这里不在是等待func2对象，而是睡眠3s
    print("end")


async def func2():
    i = 0
    while True:
        if i < 10:
            await asyncio.sleep(1)
            print(i)
            i += 1
        else:
            break



async def main():

    task_list = [
        # name是协程任务起名字，如果不写，会自动生成默认值。
        asyncio.create_task(func1(), name='c1'),
        asyncio.create_task(func2(), name='c2')
    ]

    # 此处的await是等待所有协程执行完毕，并将所有协程的返回值保存到done
    # 如果设置了timeout值，则意味着此处最多等待timeout秒，完成的协程返回值写入到done中，未完成则写到pending中。没设置默认等待所有协程结束
    done, pending = await asyncio.wait(task_list, timeout=5)    # 等待5s
    print(done)
    print("--------------------------------")
    print(pending)


asyncio.run(main())                     # 执行

程序执行结果如下：

start
0
1
end
2
3
{<Task finished name='c1' coro=<func1() done, defined at 协程.py:144> result=None>}
--------------------------------
{<Task pending name='c2' coro=<func2() running at 协程.py:154> wait_for=<Future pending cb=[<TaskWakeupMethWrapper object at 0x7f99084f7ca0>()]>>}

输出的结果表示func1执行完成，而func2未完成。

future对象

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

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())

混用协程异步和线程（进程）异步

一般在程序开发中我们要么统一使用 asycio 的协程实现异步操作、要么都使用进程池和线程池实现异步操作。但如果协程的异步和进程池/线程池的异步混搭时，我们需要将线程池/进程池对象封装为asyncio.Future对象。例如:

import time
import asyncio


def func1():
    # 某个耗时操作
    time.sleep(2)
    return "NB"


async def main():
    loop = asyncio.get_running_loop()
    # 1. Run in the default loop's executor ( 默认ThreadPoolExecutor )
    # 第一步：内部会先调用 ThreadPoolExecutor 的 submit 方法去线程池中申请一个线程去执行func1函数，并返回一个concurrent.futures.Future对象
    # 第二步：调用asyncio.wrap_future将concurrent.futures.Future对象包装为asycio.Future对象。
    # 因为concurrent.futures.Future对象不支持await语法，所以需要包装为 asycio.Future对象 才能使用。
    fut = loop.run_in_executor(None, func1)
    result = await fut
    print('default thread pool', result)


asyncio.run(main())

当项目以协程式的异步编程开发时，如果要使用一个第三方模块，而第三方模块不支持协程方式异步编程时，就需要用到这个功能。

uvloop

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

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

安装uvloop

pip3 install uvloop

import asyncio
import uvloop       # 导入uvloop

async def func1():
    print("start")
    await asyncio.sleep(3)      # 注意，这里不在是等待func2对象，而是睡眠3s
    print("end")


async def func2():
    i = 0
    while True:
        if i < 10:
            await asyncio.sleep(1)
            print(i)
            i += 1
        else:
            break

# 只需要下面这句，使用uvloop的事件循环替代asyncio的事件循环即可。
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

task_list = [func1(), func2()] 
t1 = asyncio.wait(task_list)     
asyncio.run(t1)

一个闪电般快速的ASGI服务器Uvicor就是基于uvloop和httptools构建的。

参考资料 https://www.cnblogs.com/wupeiqi/p/12834355.html

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2021-12-09，如有侵权请联系 cloudcommunity@tencent.com 删除

node.js