从头造轮子:python3 asyncio之 run_until_complete (1)

发布于 2022年 01月 12日 08:44

前言

今天开始聊一聊python3的asyncio。关于asyncio,大家肯定都有自己的理解,并且网上大神也把基础概念也解释的比较透彻。
本文写作的初衷,主要是理解asyncio的原理并且实现一遍。
话不多说,我们开始!

一、知识准备

● 理解进程、线程、协程。简单来说,这三个都是为了解决多任务同时进行的问题
  1)进程是操作资源分配的最小单位,多任务的实现主要是极快地在进程间来回切换,而进程切换消耗时间最长(系统调用)
  2)线程依赖于进程,多个线程共享了父进程的一部分资源,线程切换时间相对于进程来说消耗时间大大减少,但是由于python gil的存在,并不存在多线程(系统调用)
  3)协程依赖于线程,由于进程/线程切换都是系统调用,开销是巨大的。而协程是在用户空间内完成任务切换,不会切换到操作系统资源(寄存器、信号量、堆栈等),所以这种方式开销最小。python的协程核心在于,遇到等待事件,就交出cpu控制权,转而让其他协程执行


● 理解python生成器,yield/yield from
  这里就不班门弄斧了,直接推荐大佬的blog


● 理解关键字async/await,async/await是3.5之后的语法,和yield/yield from异曲同工


二、环境准备

组件 版本
python 3.7.7

三、run_until_complete的实现

先来看下官方asyncio的使用方法:

|># more main.py
import asyncio
async def hello():
    print('enter hello ...')
    return 'world'

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
    
|># python3 main.py
enter hello ...
world

来看下造的轮子的使用方式:

▶ more main.py
from wilsonasyncio import get_event_loop
async def hello():
    print('enter hello ...')
    return 'world'


if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
    
▶ python3 main.py
enter hello ...
world

自己造的轮子也很好的运行了,下面我们来看下轮子的代码

四、代码解析

轮子代码

1)代码组成

|># tree
.
├── eventloops.py 
├── futures.py
├── main.py
├── tasks.py
├── wilsonasyncio.py
文件 作用
eventloops.py 事件循环
futures.py futures对象
tasks.py tasks对象
wilsonasyncio.py 可调用方法集合
main.py 入口

2)代码概览:

eventloops.py

类/函数 方法 对象 作用 描述
Eventloop 事件循环,一个线程只有运行一个
__init__ 初始化两个重要对象 self._readyself._stopping
self._ready 所有的待执行任务都是从这个队列取出来,非常重要
self._stopping 事件循环完成的标志
call_soon 调用该方法会立即将任务添加到待执行队列
run_once run_forever调用,从self._ready队列里面取出任务执行
run_forever 死循环,若self._stopping则退出循环
run_until_complete 非常重要的函数,任务的起点和终点(后面详细介绍)
create_task 将传入的函数封装成task对象,这个操作会将task.__step添加到__ready队列
Handle 所有的任务进入待执行队列(Eventloop.call_soon)之前都会封装成Handle对象
__init__ 初始化两个重要对象 self._callbackself._args
self._callback 待执行函数主体
self._args 待执行函数参数
_run 待执行函数执行
get_event_loop 获取当前线程的事件循环
_complete_eventloop 将事件循环的_stopping标志置位True

tasks.py

类/函数 方法 对象 作用 描述
Task 继承自Future,主要用于整个协程运行的周期
__init__ 初始化对象 self._coro ,并且call_soonself.__step加入self._ready队列
self._coro 用户定义的函数主体
__step Task类的核心函数

futures.py

类/函数 方法 对象 作用 描述
Future 主要负责与用户函数进行交互
__init__ 初始化两个重要对象 self._loopself._callbacks
self._loop 事件循环
self._callbacks 回调队列,任务暂存队列,等待时机成熟(状态不是PENDING),就会进入_ready队列
add_done_callback 添加任务回调函数,状态_PENDING,就虎进入_callbacks队列,否则进入_ready队列
set_result 获取任务执行结果并存储至_result,将状态置位_FINISH,调用__schedule_callbacks
__schedule_callbacks 将回调函数放入_ready,等待执行
result 获取返回值

3)执行过程

3.1)入口函数

main.py

async def hello():
    print('enter hello ...')
    return 'world'

if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
  • loop = get_event_loop()获取事件循环
  • task = loop.create_task(hello())将用户函数hello()封装成协程,我们看下create_task的源码
    def create_task(self, coro):
        task = tasks.Task(coro, loop=self)
        return task

    初始化一个Task对象,从代码概览得知,初始化对象之后会立即将__step添加到_ready队列等待执行

  • rst = loop.run_until_complete(task)开始执行事件循环的第一个函数run_until_complete

3.2)事件循环启动

eventloops.py

    def run_until_complete(self, future):
        future.add_done_callback(_complete_eventloop, future)
        self.run_forever()
        return future.result()
  • future.add_done_callback(_complete_eventloop, future)future添加回调函数(future就是task对象,而task对象里的任务就是hello() ),回调函数是_complete_eventloop 。就是future执行完成之后执行_complete_eventloop
  • self.run_forever()启动事件循环

3.3)第一次循环run_forever --> run_once

eventloops.py

    def run_once(self):
        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()
  • _ready队列的内容(task.__step)取出来执行

tasks.py

    def __step(self, exc=None):
        coro = self._coro
        try:
            if exc is None:
                coro.send(None)
            else:
                coro.throw(exc)
        except StopIteration as exc:
            super().set_result(exc.value)
        finally:
            self = None
  • coro.send(None)核心代码,跳转回到用户函数hello()

main.py

async def hello():
    print('enter hello ...')
    return 'world'
  • 用户函数非常简单,打印一行数据,以及返回一个字符串world,执行完成之后回到task.__step()
  • super().set_result(exc.value)由于用户函数执行完成,会抛出异常StopIteration,捕获之后执行set_result
  • 由代码概览得知set_result 的作用在于将任务状态置位_FINISHED,并且将回调函数(_complete_eventloop )写入_ready队列

3.4)第二次循环run_forever --> run_once

eventloops.py

    def run_once(self):
        ntodo = len(self._ready)
        for _ in range(ntodo):
            handle = self._ready.popleft()
            handle._run()
  • 继续循环,handle封装了_complete_eventloop
def _complete_eventloop(fut):
    fut._loop.stop()
  • 调用stop,设置停止标志

3.5)第三次循环run_forever

    def run_forever(self):
        while True:
            self.run_once() 
            if self._stopping:
                break
  • 跳出事件循环,回到run_until_complete
    def run_until_complete(self, future):
        future.add_done_callback(_complete_eventloop, future)
        self.run_forever()
        return future.result()

3.6)回到主函数,获取返回值

if __name__ == "__main__":
    loop = get_event_loop()
    task = loop.create_task(hello())
    rst = loop.run_until_complete(task)
    print(rst)
  • rst = loop.run_until_complete(task)获取返回值

3.7)执行结果

▶ python3 main.py
enter hello ...
return world ...

五、流程总结

六、小结

● task对象与future有什么区别?主要用于整个协程运行的周期,主要负责与用户函数进行交互
● 本文从asyncio的第一个函数run_until_complete,介绍了asyncio的基本流程:用户函数并不是立即执行,而是进入队列,然后根据eventloop在合适的时机进行统一调度
● 本文中的代码,参考了python 3.7.7中asyncio的源代码,裁剪而来
● 本文中代码:代码



至此,本文结束
在下才疏学浅,有撒汤漏水的,请各位不吝赐教...

推荐文章