async await 闲谈

async await 闲谈

yield 和 yield from

在该例子中,首先创建了一个生成器对象t=f1(),然后不断使用t.send(None)一步步执行,在执行过程中进入sub()生成器,执行完后又返回到t中.

def sub():
    yield 'sub-1'
    yield 'sub-2'
    yield 'sub-3'

def f1():
    print('---enter---')
    # 通过遍历的方式执行子生成器
    for i in sub():
        yield i
    print('---exit---')

if __name__ == "__main__":
    # 创建生成器
    t = f1()
    # 一步步执行生成器
    while True:
        try:
            step_res = t.send(None)
            print(step_res)
        except StopIteration:
            break

执行结果:

---enter---
sub-1
sub-2
sub-3
---exit---

上面的例子中使用 for 循环执行子生成器,当子生成器很多时,这无疑是件很繁琐的事情, yield from的作用正是为了解放遍历子生成器的工作.
改造后代码如下:

def sub():
    yield 'sub-1'
    yield 'sub-2'
    yield 'sub-3'

def f1():
    print('---enter---')
    # 通过yield from执行子生成器
    yield from sub()
    print('---exit---')

if __name__ == "__main__":
    # 创建生成器
    t = f1()
    # 一步步执行生成器
    while True:
        try:
            step_res = t.send(None)
            print(step_res)
        except StopIteration:
            break

async await 和 yield from

await 和 yield from 实现的功能相似. 可以简单认为,async 和 await 是 yield from 语法糖

class Sub:
    def __await__(self):
        yield 'sub-1'
        yield 'sub-2'
        yield 'sub-3'


async def f3():
    print('---enter---')
    await Sub()
    print('---exit---')



if __name__ == "__main__":
    t = f1()
    while True:
        try:
            step_res = t.send(None)
            print(step_res)
        except StopIteration:
            break

如果仅仅是为了替代 yield from,那 async 和 await 存在的意义是什么?
使用 async await 最大的优点在于不用自己一步一步执行生成器对象,asyncio 实现了一套事件循环机制去执行 coroutine(即调用 async 函数返回的对象).

future 对象

future 可以类比 ES6 的 Promise 对象,await 一个 future 时会将当前执行的协程挂起,future 有结果时将结果返回,再从挂起处恢复执行.

• future 对象分两部分执行,await 一个 future 时执行 future 前半部分,调用 set_result 后执行后半部分; 具体的调度是由 event_loop 和 Task 完成的.

class Future:
    def __init__(self, *, loop=None):
        self._result = None
        self._callbacks = []
        self._loop = loop

    def set_result(self, result):
        self._result = result
        callbacks = self._callbacks[:]
        self._callbacks = []
        for callback in callbacks:
            loop._ready.append(callback)

    def add_callback(self, callback):
        self._callbacks.append(callback)

    def __iter__(self):
        # 前半部分
        yield self
        # 后半部分
        return 'future'

    __await__ = __iter__

使用 future 对象实现自己的异步 sleep 方法

import asyncio
import time


async def my_sleep(time):
    ev = asyncio.get_event_loop()

    # 创建一个future对象
    fut = ev.create_future()

    def set_result():
        fut.set_result(None)

    # 高度event_loop,time秒后给future设置结果.
    ev.call_later(time, set_result)
    # 等待future对象,等其有结果后恢复执行.
    await fut


async def tt():
    print('__start__', time.perf_counter())
    await my_sleep(2)
    print('__end__', time.perf_counter())


def creak_tasks():
    ev = asyncio.get_event_loop()
    ev.create_task(tt())
    ev.create_task(tt())
    ev.create_task(tt())


if __name__ == '__main__':
    ev = asyncio.get_event_loop()
    creak_tasks()
    ev.run_forever()

__start__ 0.177308003
__start__ 0.177351966
__start__ 0.177368488
__end__ 2.179307751
__end__ 2.179349455
__end__ 2.17936377

coroutine 对象

我们知道,调用一个含有 yield 的函数会返回一个 generator 对象,类似的,调用一个 async 函数会返回一个 coroutine 对象;
coroutine 对象可以认为是包装过的 generator 对象.

async def f1():
    pass


def f2():
    yield


print(f1())
print(f2())

输出:

<coroutine object f1 at 0x1054f75c8>
<generator object f2 at 0x10576a390>

Task 对象

Task 对象被用来在事件循环中运行协程。如果一个协程在等待一个 Future 对象，Task 对象会挂起该协程的执行并等待该 Future 对象完成。当该 Future 对象 完成，被打包的协程将恢复执行。

在前面的章节中介绍到,要执行一个 coroutine 对象需要调用其 next()方法.
Task 对象可以认为是 coroutine 对象的执行器. 通过 event_loop 调度 Tasks 可以自动执行 coroutine.

event loop 对象

event loop 是 asyncio 的核心,用于 Tasks 的调度执行.
asyncio.get_event_loop() 会返回一个单例.

event loop 核心 api 其实很少,理解它们对应理解 asyncio 具有很大的帮助.

call_soon 和 call_later

• call_soon(fn) 在下一轮 loop 中执行 fn,fn 就是一个普通函数;
• call_soon(time,fn) 在 time 秒后执行 fn;

看下面例子:

import asyncio
import time


def ct():
    return time.perf_counter()


ev = asyncio.get_event_loop()
ev.call_later(3, lambda: print(f'call after 3 sec at {ct()}'))
ev.call_later(2, lambda: print(f'call after 2 sec at {ct()}'))
ev.call_later(1, lambda: print(f'call after 1 sec at {ct()}'))
ev.call_soon(lambda: print(f'call soon 1 at {ct()}'))
ev.call_soon(lambda: print(f'call soon 2 at {ct()}'))
ev.call_soon(lambda: print(f'call soon 3 at {ct()}'))
ev.run_forever()

输出:

call soon 1 at 0.215582999
call soon 2 at 0.215620497
call soon 3 at 0.215626632
call after 1 sec at 1.218513569
call after 2 sec at 2.216406517
call after 3 sec at 3.217328712

creat_task

接收一个 coroutine 对象,创建一个 Task 对象,并返回.

这里是代码段的截取

class BaseEventLoop(events.AbstractEventLoop):
    def create_task(self, coro):
        """Schedule a coroutine object.

        Return a task object.
        """
        self._check_closed()
        if self._task_factory is None:
            task = tasks.Task(coro, loop=self)
            if task._source_traceback:
                del task._source_traceback[-1]
        else:
            task = self._task_factory(self, coro)
        return task

task 对象在创建完成之后就会在下一轮 loop 循环中执行 coroutine 对象,Task 的init方法中调用了 loop.call_soon();

class Task(futures._PyFuture):
    def __init__(self, coro, *, loop=None):
        super().__init__(loop=loop)
        # ...
        if not coroutines.iscoroutine(coro):
            self._log_destroy_pending = False
            raise TypeError(f"a coroutine was expected, got {coro!r}")

        # ...
        self._coro = coro
        self._loop.call_soon(self.__step, context=self._context)

使用举例:

import asyncio
import time


def ct():
    return time.perf_counter()


async def af():
    print(f'call async function at {ct()}')


ev = asyncio.get_event_loop()
ev.call_later(3, lambda: print(f'call after 3 sec at {ct()}'))
ev.call_later(2, lambda: print(f'call after 2 sec at {ct()}'))
ev.call_later(1, lambda: print(f'call after 1 sec at {ct()}'))
ev.call_soon(lambda: print(f'call soon 1 at {ct()}'))
ev.call_soon(lambda: print(f'call soon 2 at {ct()}'))
ev.call_soon(lambda: print(f'call soon 3 at {ct()}'))

ev.create_task(af())


ev.run_forever()

输出:

call soon 1 at 0.308287723
call soon 2 at 0.308325895
call soon 3 at 0.308332114
call async function at 0.308339678
call after 1 sec at 1.312385017
call after 2 sec at 2.308573672
call after 3 sec at 3.308438894