前言
今天开始聊一聊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._ready 与 self._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._callback 与 self._args |
|||
self._callback |
待执行函数主体 | |||
self._args |
待执行函数参数 | |||
_run |
待执行函数执行 | |||
get_event_loop |
获取当前线程的事件循环 | |||
_complete_eventloop |
将事件循环的_stopping 标志置位True |
tasks.py
类/函数 | 方法 | 对象 | 作用 | 描述 |
---|---|---|---|---|
Task | 继承自Future,主要用于整个协程运行的周期 | |||
__init__ |
初始化对象 self._coro ,并且call_soon 将self.__step 加入self._ready 队列 |
|||
self._coro |
用户定义的函数主体 | |||
__step |
Task类的核心函数 |
futures.py
类/函数 | 方法 | 对象 | 作用 | 描述 |
---|---|---|---|---|
Future | 主要负责与用户函数进行交互 | |||
__init__ |
初始化两个重要对象 self._loop 与 self._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的源代码,裁剪而来
● 本文中代码:代码
至此,本文结束
在下才疏学浅,有撒汤漏水的,请各位不吝赐教...