2. python协程/异步编程详解

       

目录

1. 简单的异步程序

2. 协程函数和协程对象

3. 事件循环

4. 任务对象Task及Future对象

        4.1 Task与Future的关系

        4.2 Future对象

        4.3 全局对象和循环事件对象

5. await关键字

6. 异步上下文管理

7.异步迭代器

8. asyncio的常用函数

        8.1 asyncio.run

        8.2 asyncio.get_event_loop

        8.3 asyncio.wait

        8.4 asyncio.wait_for

        8.5 asyncio.create_task和loop.create_task

        8.6 eventloop.call_later

        8.7 eventloop.close

---

        协程（coroutine）：类似一种轻量级的多线程程序，是在一个线程里面，实现多任务并行执行的编程方式。相比多线程，它在切换任务时，CPU开销更低，速度更快。

        协程实现多任务并行执行实际是通过各任务异步执行来实现的。

        asyncio 是python异步代码库，是几乎所有python异步实现的基础，因此，使用python的协程功能，得先掌握asyncio库的使用。

1. 简单的异步程序

import asyncio
async def fun1():print('a')await asyncio.sleep(.5)print('b')async def fun2():print('c')await asyncio.sleep(.5)print('d')asyncio.run(asyncio.wait([fun1(),fun2()]))

        运行结果： 

        可以看到，虽然fun1和fun2是依次放进事件列表中的，可实际执行，并不是abcd，而是cadb，先执行fun2后执行fun1，这里与列表在函数取参时LIFO（后入先出）有关。

        在执行过程中，fun2先打印c后，遇到sleep，并不是原地阻塞0.5秒，而是跳出来，执行fun1，打印a，在fun1中又遇到sleep，这里又跳出来执行fun2，fun2结束后又去fun1执行。这就是异步执行，它并不是顺序执行，而是可以暂停当前执行的流程，转到另一个程序中去执行，之后又回到暂停位置继续执行。

2. 协程函数和协程对象

        由async声明定义的函数，即是协程函数，它直接返回的就是协程对象。

        协程函数要被执行，须将协程对象交给事件循环来处理，或者通过await关键字执行，不能直接执行协程函数。

        注意，协程函数直接返回协程对象，指函数名()返回，并不是执行结果return返回：

import asyncio
async def fun():await asyncio.sleep(.5)print("This is a coroutine function")return "Hello world"async def main():t = fun()r = await tprint(f"Result:{r}")asyncio.run(main())##运行结果：
This is a coroutine function
Result:Hello world

        示例中函数名fun()返回的是一个协程对象，它被赋值给t，通过await执行后，fun最后的返回结果放在r中，fun的return返回Hello world，因此最后打印出的r就是Hello world 

3. 事件循环

        事件循环是asyncio运行的核心，是由循环调度器_event_loop_policy调度加入到Task队列中的Task对象进行执行。

        在上面调用asyncio.run函数的时候，该函数实际是创建一个循环调度器，并将将可Task对象放进循环中，然后等待对象执行完成，再退出。它相当于：

eventloop = asyncio.new_event_loop()
asyncio.set_event_loop(eventloop)
eventloop.run_until_complete(obj)

        asyncio.new_event_loop创建一个新的循环调度器，set_event_loop启用该调度器，run_until_complete将对象obj放入调度器的Tasks队列中，等待执行完成。

        事件循环的本质是将所有待执行的协程对象包装成任务对象（Task对象），将Task对象放入到事件循环队列中，调度器执行事件循环，当某个对象遇到阻塞，就挂起转而执行队列中的其它对象，当挂起时间到了，又回到该对象挂起位置继续执行。

4. 任务对象Task及Future对象

        4.1 Task与Future的关系

        asyncio.Future类是一个基类，asyncio.Task类继承自Future。事件循环调度器中执行的对象就是任务对象（Task对象），只有Future对象（或Task对象）才能放入进事件循环队列tasks中。

        因此，上面示例中，run_until_complete接收的参数应该是Future对象，当run_until_complete传入一个协程对象后，会先判断是否为Future对象，如果是协程，则被wrap成Task对象（Task(coro,eventloop)），再将对象注册（_register_task）到队列中。

        4.2 Future对象

        通常不直接用，它常用来保存最后结果，在Task对象中，当等待Future.result()时，也即任务结整返回，最后执行Future.set_result(value)：

import asyncioasync def fun(fut:asyncio.Future):print("This is a coroutine")fut.set_result('ABCDE')async def main():eventloop = asyncio.get_event_loop()fut = eventloop.create_future()eventloop.create_task(fun(fut))r = await futprint(f'Result:{r}')asyncio.run(main())##运行结果：
This is a coroutine
Result:ABCDE

        分析：asyncio.get_event_loop返回当前执行的EventLoop对象，它由asyncio.run创建。eventloop.create_future()创建一个Future对象，它什么也不做，它被当作参数传入到协程中；evenloop.create_task()将协程wrap成Task对象，加入到事件循环调度器的任务队列Tasks中，await fut等待fut有结果，并将结果保存在r中。

        当协程执行完后，调用fut.set_result就会在Future对象中产生一个值（结果），这时协程返回，await fut就因为fut有结果了结束，将结果返回给r，即ABCD。

        4.3 全局对象和循环事件对象

t1 = asyncio.create_task(coro)
t2 = eventloop.create_task(coro)
fut1 = asyncio.create_future()
fut2 = eventloop.create_future()

        ayncio和eventloop都能创建Task对象，前者是个全局Task对象，不依赖特定的事件循环，而后者则依赖当前的eventloop事件循环

5. await关键字

        await 后面跟可等待对象（awaitable，包括协程对象，任务对象，Future对象），这些对象最终都会被wrap成Task对象执行。await主要作用是暂停当前协程，等待对象执行完成，返回执行结果（函数最后的return），才回到当前协程。

import asyncioasync def fun1():print("This fun1")await asyncio.sleep(.5)print("Fun1 over")return "A fun return."
async def fun2():print("This fun2")await asyncio.sleep(.5)print("Fun2 over")
async def fun3():print("This fun3")await asyncio.sleep(.5)print("Fun3 over")async def main():eventloop = asyncio.get_event_loop()a = eventloop.create_task(fun1())b = asyncio.create_task(fun2())print(f'{await a}')await fun3()await basyncio.run(main())##运行结果：
This fun1
This fun2
Fun1 over
Fun2 over
A fun return.
This fun3
Fun3 over

        示例分析：eventloop.create_task返回的是协程wrap后的Task对象，分别赋值给变量a和b，await可以作用Task对象，因此，await a得以执行，它等待a执行，直到a返回结果(字符串A fun return，通过print打印出来）。

        从运行结果看，实际上await a返回时，fun2已经结束了，也就是b对象已经执行结束了，所以，await fun3执行结果并不会出现在a,b对象执行的过程中，后面的await b也不会出现执行结果，因为前面已经执行完了。

        一个特别的函数：

        await asyncio.sleep(n)

        它表示暂停当前任务的事件循环控制权，转而执行其它异步任务，时间到了后，再回到当前协程继续执行。

6. 异步上下文管理

        python中上下文管理关键字with，它可以自动创建资源，并在结束后自动回收资源。with作用于定义了__enter__和__exit__方法的对象。

        __enter__()：定义了进入with代码块执行的动作，返回值默认为None，也可以指定某个返回值，返回的接收者就是as后面的变量；

        __exit__(exc_type, exc_value, traceback)：定义了退出with代码块执行的动作      

class Person:def __init__(self):self.name:str = "YSZ"self.score:float = 93.7def __enter__(self):print("Person enter")return selfdef __exit__(self, exc_type, exc_value, traceback):print("Person exit.")with Person() as p:print(f"name:{p.name}")print(f"score:{p.score}")##运行结果：
Person enter
name:YSZ
score:93.7
Person exit.

        异步上下文管理器是在with前面增加async关键字的管理器，类似的，这里作用的对象是定义了__aenter__和__aexit__两个异步函数的对象：

import asyncioclass Person:def __init__(self):self.name:str = "YSZ"self.score:float = 93.7async def __aenter__(self):print("Person enter")return selfasync def __aexit__(self, exc_type, exc_value, traceback):print("Person exit.")async def main():async with Person() as p:print(f"name:{p.name}")print(f"score:{p.score}")asyncio.run(main())##运行结果
Person enter
name:YSZ
score:93.7
Person exit.

7.异步迭代器

        一般情况下，可以用for in迭代遍历的对象都是定义了__iter__，__next__函数的对象，其中：

        __iter__返回一个可迭代对象iterator

        __next__返回操作对象，赋给for后面的变量（如下例的i），for不停的迭代，直到引发一个StopIteration异常：

class Person:def __init__(self):self.index = 0self.names:tuple = ("YSZ","TMS","JYN","MOD","GXT","QES")self.name:str = Nonedef __iter__(self):return selfdef __next__(self):self.name = self.names[self.index]self.index += 1if self.index > 5:raise StopIterationreturn selffor i in Person():print(f"name:{i.name} index:{i.index}")##运行结果：
name:YSZ index:1
name:TMS index:2
name:JYN index:3
name:MOD index:4
name:GXT index:5

        类似的，异步迭代器的对象是定义了__aiter__和__anext__函数的对象，其中：

        __aiter__函数不能是异步的，它返回一个异步可迭代对象asynchronous iterator，赋给in后面的变量

        __anext__返回对象必须是可等待（awaitable）对象，因此它必须是一个协程函数。async for不停的迭代，直到引发一个StopAsyncIteration异常停止： 

class Person:def __init__(self):self.index = 0self.names:tuple = ("YSZ","TMS","JYN","MOD","GXT","QES")self.name:str = Nonedef __aiter__(self):return selfasync def __anext__(self):self.name = self.names[self.index]self.index += 1if self.index > 5:raise StopAsyncIterationawait asyncio.sleep(.2)return self.nameasync def main():person = Person()async for i in person:print(f"name:{i}")asyncio.run(main())##运行结果：
name:YSZ
name:TMS
name:JYN
name:MOD
name:GXT

        注意：async for只能在协程函数中使用

8. asyncio的常用函数

        8.1 asyncio.run

        在第3点已经说明了，它实际是创建了一个新的事件循环调度器，并启动这个调度器，然后等待传给它的对象执行完成，它的参数接收可等待对象（协程对象，Task对象，Future对象）

        8.2 asyncio.get_event_loop

        返回：事件循环调度器

        它会获取当前正在运行的事件循环调度器，相当于asyncio.get_running_loop()；

        如果当前没有正在运行的事件循环，则创建一个新的事件循环，相当于loop = asyncio.new_event_loop，同时运行这个事件循环：asyncio.set_event_loop(loop)，并将这个事件循环调度器返回。

        8.3 asyncio.wait

        参数：一组可等待对象列表（list）

        返回：Future对象

        作用：等待一组可等待对象完成，最后返回任务结束的结果

        例如：asyncio.run(asyncio.wait(fun1(),fun2(),fun3())，其中fun1()~fun3()是协程对象

        8.4 asyncio.wait_for

        参数：可等待对象，超时的时间timeout

        返回：可等待对象执行结束后的返回（return）

        作用：通常和关键字await连用，表示对象必须在设定时间内执行完毕，否则抛出一个超时异常asyncio.TimeoutError

import asyncioasync def fun():print("Step1")await asyncio.sleep(.5)print("Step2")await asyncio.sleep(.5)return "Fun back"async def main():try:r = await asyncio.wait_for(fun(),timeout=0.8)print(f"Result:{r}")except asyncio.TimeoutError:print("Object execute timeout.")asyncio.run(main())##运行结果：
Step1
Step2
Object execute timeout.

        示例中规定了0.8秒内要完成fun()对象，但是里面花了1.0秒，固而抛出超时异常

        8.5 asyncio.create_task和loop.create_task

        参数：协程对象

        返回：Task对象

        作用：创建一个Task对象，第4.3已经说明了它们的区别，其中loop.create_task同时将创建的对象注册到任务队列中了，事件循环会执行它，而asyncio.create_task则是创建一个全局对象，并没有加到某个事件循环的队列中，需要await执行

        8.6 eventloop.call_later

        参数：时间，回调函数，函数的参数

        返回：TimerHandle

        作用：布置一个任务，过了“时间”之后被执行

import asynciodef fun(name:str):print(f"name:{name}  Hello world")async def main():eventloop = asyncio.get_event_loop()eventloop.call_later(1,fun,'YSZ')await asyncio.sleep(3)print("Over!")asyncio.run(main())##运行结果：
name:YSZ  Hello world
Over!

        示例中布置了一个任务，注意，并非是协程，而是普通函数，并要求在1秒后执行，需要await asyncio.sleep(3)等待，否则循环事件结束了，这个回调还没执行。 

        8.7 eventloop.close

        作用：关闭事件循环，需要先检查事件循环是否正在运行eventloop.is_running()，如果正在运行，则关不了，会触发一个RuntimeError异常。