原生协程是基于async关键字的
# 定义一个简单的原生协程cor
async def cor():
print('enter cor')
print('exit cor')
print(type(cor)) # <class 'function'>
print(type(cor())) # <class 'coroutine'>
可以看到cor
的类型<class 'function'>
函数类型,说明async
关键字修饰的函数也是一个函数而已,跟普通函数在定义上没啥什么差别,差别在于被调用的时候,cor()
并不是执行函数体中的语句,而是生成一个<class 'coroutine'>
类型的对象,即协程对象,跟生成器类似,协程也有send()
方法。
c = cor()
try:
c.send(None)
except StopIteration as e:
print(e.value) # None
当c.send(None)时,才会执行cor中的语句,也就是执行print('enter cor')
和print('exit cor')
这两句,就执行完毕了,执行完毕时会抛出StopIteration
异常,这个异常对象的value
属性中保存着cor
的返回值,这里core
没有return
语句,其实默认返回None
。
原生协程中使用 await 关键字
async def cor():
print('enter cor')
rst = await cor1()
print('rst -->', rst)
print('exit cor')
async def cor1():
print('enter cor1')
print('exit cor1')
return 'cor1'
c = cor()
try:
c.send(None)
except StopIteration as e:
print('cor -->', e.value) # None
输出如下:
enter cor
enter cor1
exit cor1
rst --> cor1
exit cor
cor --> None
await
关键字后面必须是一个实现了__awwit__
方法的对象,不一定是协程对象,只要实现了这个方法,就会进入到此方法中,调用该方法中的逻辑。比如后面要介绍的Future
对象,它也是实现了__await__
方法的。
await
关键字的语义是表示挂起当前的cor
协程的执行,进入到cor1
协程中执行cor1
的逻辑,直到cor1
执行完毕,然后执行流程又回到cor
挂起的地方,继续执行cor
中await
后面的语句。直到最后抛出StopIteration
异常。
基于生成器的协程,也就是非原生协程
import types
async def cor():
print('enter cor')
rst = await cor2()
print('rst --> ', rst)
print('exit cor')
@types.coroutine
def cor2():
print('enter cor2')
yield
print('exit cor2')
return 'cor2'
c = cor()
try:
c.send(None)
c.send(None)
except StopIteration as e:
print('cor -->', e.value) # None
输出如下:
enter cor
enter cor2
exit cor2
rst --> cor2
exit cor
cor --> None
与上面的原生协程的嵌套不同的是,调用了两次c.send(None)
,执行第一次c.send(None)
时,会在cor2
的yield
关键字处挂起,第二次c.send(None)
则会在yield
挂起的地方,接着往下执行,然后core2
返回'cor2'
,赋值给rst
变量,继续执行cor
中await
后面的语句,直到最后抛出StopIteration
异常。
总结
async 是一个关键字,async def 定义的类型还是一个function类型,只有当它被调用时才返回一个协程对象
async def
跟def
定义的方法在没被调用时没有任何区别,不必看得很神秘,它也可以有return
语句,这点也正常,
因为python
中没有return
语句的函数实际上默认是返回None
的,所以只是显式的return
和隐式return None
的区别
对于协程的send(None)
方法,跟生成器的send(None)
类似,不同的是,原生协程的send方法被调用的时候,会一直执行到碰
到await
语句,但是不会停下来,会直接再进入到await EXPR
的EXPR
中,其实EXPR
是一个awaitable
对象,会调用该对象的
__await__()
执行该方法的里面的逻辑,如果该awaitable
对象是一个原生协程对象,那么它的__await__()
方法中的逻辑就
是在定义此协程时async def
下面的逻辑,执行完毕后,该协程对象就关闭了,执行流程就再次跳转到当前挂起的协程中,
执行该协程中余下的逻辑,最后执行完毕,抛出StopIteration
异常。
对于原生生协程来说,调用send()
方法时,会一直执行到出现StopIteration
异常为止,只有在 __await__()
方法中有yield
语句时才
会挂起在那里,如果__await__()
方法中没有yield
语句,不会挂起,会返回await的返回值,继续执行,直到抛出StopIteration
异常。
先抛出一个结论,在asyncio
中协程的流程的挂起操作,实际上还是是通过yield
关键字来实现的,并不是await
关键字, async
和await
关键字只不过是语法糖。
asyncio的基本流程分析
import asyncio
async def cor():
print('enter cor ...')
await asyncio.sleep(2)
print('exit cor ...')
return 'cor'
loop = asyncio.get_event_loop()
task = loop.create_task(cor())
rst = loop.run_until_complete(task)
print(rst)
从这个简单的例子入手,逐步分析协程在事件循环中的执行流程。
第1步
async def cor()
定义了一个cor
协程。第2步
loop = asyncio.get_event_loop()
得到一个事件循环对象loop
,这个loop
在一个线程中只有唯一的一个实例,只要在同一个线程中调用此方法,得到的都是同一个loop
对象。第3步
task = loop.create_task(cor())
把cor
协程包装成一个task
对象。第4步
rst = loop.run_until_complete(task)
把这个task
对象添加到事件循环中去。
首先看第3步的loop.create_task(cor())这个方法
class BaseEventLoop(events.AbstractEventLoop):
...
def __init__(self):
...
# 用来保存包裹task.step方法的handle对象的对端队列
self._ready = collections.deque()
# 用来保存包裹延迟回调函数的handle对象的二叉堆,是一个最小二叉堆
self._scheduled = []
...
def create_task(self, coro):
"""Schedule a coroutine object.
Return a task object.
"""
self._check_closed()
# self._task_factory 默认是None
if self._task_factory is None:
# 创建一个task对象
task = tasks.Task(coro, loop=self)
if task._source_traceback:
del task._source_traceback[-1]
else:
task = self._task_factory(self, coro)
# 返回这个task对象
return task
def call_soon(self, callback, *args):
self._check_closed()
if self._debug:
self._check_thread()
self._check_callback(callback, 'call_soon')
# 关键代码callback就是task._step方法,args是task._step的参数
handle = self._call_soon(callback, args)
if handle._source_traceback:
del handle._source_traceback[-1]
return handle
def _call_soon(self, callback, args):
# 1 handle是一个包裹了task._step方法和args参数的对象
handle = events.Handle(callback, args, self)
if handle._source_traceback:
del handle._source_traceback[-1]
# 2 关键代码,把handle添加到列表self._ready中
self._ready.append(handle)
return handle
loop.create_task(cor())
实际上是创建了一个Task
类的实例。再来看一下Task
这个类
class Task(futures.Future):
...
def __init__(self, coro, *, loop=None):
assert coroutines.iscoroutine(coro), repr(coro)
# 调用父类的__init__获得线程中唯一的loop对象
super().__init__(loop=loop)
if self._source_traceback:
del self._source_traceback[-1]
self._coro = coro
self._fut_waiter = None
self._must_cancel = False
# 关键代码,把_step方法注册到loop对象中去
self._loop.call_soon(self._step)
self.__class__._all_tasks.add(self)
def _step(self, exc=None):
...
result = coro.send(None)
...
task
实例化时,调用了self._loop.call_soon(self._step)
--> loop.call_soon(callback, *args)
--> loop._call_soon(callback, args)
,实际上是把handle(task._step)
这个对象放到了loop._ready
队列中,放在这个队列中有什么用呢?先告诉大家,_step
方法会在loop
对象的循环中被调用,也就是会执行coro.send(None)
这句。coro.send(None)
实际上就是执行上面定义的cor
协程的里面的语句。
也就是说到第3步
执行完时,loop
对象已经实例化,task
对象也实例化,并且task
对象的_step
方法被封装成handle
对象放入了loop
对象的_ready
队列中去了。
再来看第4步loop.run_until_complete(task)
class BaseEventLoop(events.AbstractEventLoop):
def run_until_complete(self, future):
...
# future就是task对象,下面2句是为了确保future是一个Future类实例对象
new_task = not futures.isfuture(future)
future = tasks.ensure_future(future, loop=self)
if new_task:
# An exception is raised if the future didn't complete, so there
# is no need to log the "destroy pending task" message
future._log_destroy_pending = False
# 添加回调方法_run_until_complete_cb到当前的task对象的callbacks列表中,_run_until_complete_cb就是最后
# 把loop的_stop属性设置为ture的,用来结束loop循环的
future.add_done_callback(_run_until_complete_cb)
try:
# 开启无线循环
self.run_forever()
except:
...
raise
finally:
...
# 执行完毕返回cor的返回值
return future.result()
def run_forever(self):
...
try:
events._set_running_loop(self)
while True:
# 每次运行一次循环,判断下_stopping是否为true,也就是是否结束循环
self._run_once()
if self._stopping:
break
finally:
...
def _run_once(self):
# loop的_scheduled是一个最小二叉堆,用来存放延迟执行的回调函数,根据延迟的大小,把这些回调函数构成一个最小堆,然后再每次从对顶弹出延迟最小的回调函数放入_ready双端队列中,
# loop的_ready是双端队列,所有注册到loop的回调函数,最终是要放入到这个队列中,依次取出然后执行的
# 1. self._scheduled是否为空
sched_count = len(self._scheduled)
if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and
self._timer_cancelled_count / sched_count >
_MIN_CANCELLED_TIMER_HANDLES_FRACTION):
# Remove delayed calls that were cancelled if their number
# is too high
new_scheduled = []
for handle in self._scheduled:
if handle._cancelled:
handle._scheduled = False
else:
new_scheduled.append(handle)
heapq.heapify(new_scheduled)
self._scheduled = new_scheduled
self._timer_cancelled_count = 0
else:
# Remove delayed calls that were cancelled from head of queue.
while self._scheduled and self._scheduled[0]._cancelled:
self._timer_cancelled_count -= 1
handle = heapq.heappop(self._scheduled)
handle._scheduled = False
# 2. 给timeout赋值,self._scheduled为空,timeout就为None
timeout = None
# 只要self._ready和self._scheduled中有一个不为空,timeout就为0
if self._ready or self._stopping:
timeout = 0
# 只要self._scheduled不为空
elif self._scheduled:
# Compute the desired timeout.
# 用堆顶的回调函数的延迟时间作为timeout的等待时间,也就是说用等待时间最短的回调函数的时间作为timeout的等待时间
when = self._scheduled[0]._when
timeout = max(0, when - self.time())
、
if self._debug and timeout != 0:
...
# 3. 关注else分支,这是关键代码
else:
# timeout=None --> 一直阻塞,只要有io事件产生,立马返回event_list事件列表,否则一直阻塞着
# timeout=0 --> 不阻塞,有io事件产生,就立马返回event_list事件列表,没有也返空列表
# timeout=2 --> 阻塞等待2s,在这2秒内只要有io事件产生,立马返回event_list事件列表,没有io事件就阻塞2s,然后返回空列表
event_list = self._selector.select(timeout)
# 用来处理真正的io事件的函数
self._process_events(event_list)
# Handle 'later' callbacks that are ready.
end_time = self.time() + self._clock_resolution
# 4. 依次取出堆顶的回调函数handle添加到_ready队列中
while self._scheduled:
handle = self._scheduled[0]
# 当_scheduled[]中有多个延迟回调时,通过handle._when >= end_time来阻止没有到时间的延迟函数被弹出,
# 也就是说,当有n个延迟回调时,会产生n个timeout,对应n次run_once循环的调用
if handle._when >= end_time:
break
# 从堆中弹出堆顶最小的回调函数,放入 _ready 队列中
handle = heapq.heappop(self._scheduled)
handle._scheduled = False
self._ready.append(handle)
# 5. 执行self._ready队列中所有的回调函数handle对象
ntodo = len(self._ready)
for i in range(ntodo):
handle = self._ready.popleft()
if handle._cancelled:
continue
if self._debug:
try:
self._current_handle = handle
t0 = self.time()
handle._run()
dt = self.time() - t0
if dt >= self.slow_callback_duration:
logger.warning('Executing %s took %.3f seconds',
_format_handle(handle), dt)
finally:
self._current_handle = None
else:
# handle._run()实际上就是执行task._step(),也就是执行cor.send(None)
handle._run()
handle = None # Needed to break cycles when an exception occurs.
执行run_until_complete
方法时future.add_done_callback(_run_until_complete_cb)
这一步其实是task.add_done_callback(_run_until_complete_cb)
,也就是把_run_until_complete_cb
回调函数注册到task
对象中去了,这个回调函数的作用是当cor协程执行完毕时,回调_run_until_complete_cb
把loop对象的 _stopping
属性设为True
,然后_run_once
执行完毕时,判断_stopping
为True
就跳出while
循环,run_until_complete
才能返回task.result
从上面的函数调用流程
run_until_complete()
--> run_forever()
--> _run_once()
,重点看_run_once
这个方法的执行流程。
此时:
cor
协程还未开始执行。loop._ready = [handle(task._step)]
,loop._scheduled = []
。
第一轮_run_once()
的调用执行开始
注意这里的第1,2,3,4,5步是在_run_once()
上标记的1,2,3,4,5注释
第1,2步的逻辑判断,
timeout = 0
。第3步
event_list = self._selector.select(0)
,此时立马返回空[]。第4步 由于
loop._scheduled = []
,不执行第4步中的语句。第5步 依次从
_ready
队列中取出回调函数handle,执行handle._run()
。
执行handle._run()
方法,也就是调用task._step()
,来看看task._step()
的执行逻辑:
class Task(futures.Future):
...
def _step(self, exc=None):
"""
_step方法可以看做是task包装的coroutine对象中的代码的直到yield的前半部分逻辑
"""
...
try:
if exc is None:
# 1.关键代码
result = coro.send(None)
else:
result = coro.throw(exc)
# 2. coro执行完毕会抛出StopIteration异常
except StopIteration as exc:
if self._must_cancel:
# Task is cancelled right before coro stops.
self._must_cancel = False
self.set_exception(futures.CancelledError())
else:
# result为None时,调用task的callbasks列表中的回调方法,在调用loop.run_until_complite,结束loop循环
self.set_result(exc.value)
except futures.CancelledError:
super().cancel() # I.e., Future.cancel(self).
except Exception as exc:
self.set_exception(exc)
except BaseException as exc:
self.set_exception(exc)
raise
# 3. result = coro.send(None)不抛出异常
else:
# 4. 查看result是否含有_asyncio_future_blocking属性
blocking = getattr(result, '_asyncio_future_blocking', None)
if blocking is not None:
# Yielded Future must come from Future.__iter__().
if result._loop is not self._loop:
self._loop.call_soon(
self._step,
RuntimeError(
'Task {!r} got Future {!r} attached to a '
'different loop'.format(self, result)))
elif blocking:
if result is self:
self._loop.call_soon(
self._step,
RuntimeError(
'Task cannot await on itself: {!r}'.format(
self)))
# 4.1. 如果result是一个future对象时,blocking会被设置成true
else:
result._asyncio_future_blocking = False
# 把_wakeup回调函数设置到此future对象中,当此future对象调用set_result()方法时,就会调用_wakeup方法
result.add_done_callback(self._wakeup)
self._fut_waiter = result
if self._must_cancel:
if self._fut_waiter.cancel():
self._must_cancel = False
else:
self._loop.call_soon(
self._step,
RuntimeError(
'yield was used instead of yield from '
'in task {!r} with {!r}'.format(self, result)))
# 5. 如果result是None,则注册task._step到loop对象中去,在下一轮_run_once中被回调
elif result is None:
# Bare yield relinquishes control for one event loop iteration.
self._loop.call_soon(self._step)
# --------下面的代码可以暂时不关注了--------
elif inspect.isgenerator(result):
# Yielding a generator is just wrong.
self._loop.call_soon(
self._step,
RuntimeError(
'yield was used instead of yield from for '
'generator in task {!r} with {}'.format(
self, result)))
else:
# Yielding something else is an error.
self._loop.call_soon(
self._step,
RuntimeError(
'Task got bad yield: {!r}'.format(result)))
finally:
self.__class__._current_tasks.pop(self._loop)
self = None # Needed to break cycles when an exception occurs.
def _wakeup(self, future):
try:
future.result()
except Exception as exc:
# This may also be a cancellation.
self._step(exc)
else:
# 这里是关键代码,上次的_step()执行到第一次碰到yield的地方挂住了,此时再次执行_step(),
# 也就是再次执行 result = coro.send(None) 这句代码,也就是从上次yield的地方继续执行yield后面的逻辑
self._step()
self = None # Needed to break cycles when an exception occurs.
当task._step()执行时,调用core.send(None),即调用:
async def cor():
# 1.
print('enter cor ...')
# 2.
await asyncio.sleep(2)
# 3.
print('exit cor ...')
# 4.
return 'cor'
注意这里的第1,2,3,4步是在cor
上标记的1,2,3,4注释
第1步
print('enter cor ...')
第2步
await asyncio.sleep(2)
,sleep
是一个非原生协程,前面介绍过,await
语句挂起当前的协程也就是cor
,然后会进入到sleep
协程中的。注意,此时执行流程已经在sleep
协程中了,我们来看一下sleep
协程的代码逻辑。
看一下sleep
协程实现
@coroutine
def sleep(delay, result=None, *, loop=None):
"""Coroutine that completes after a given time (in seconds)."""
if delay == 0:
yield
return result
if loop is None:
loop = events.get_event_loop()
# 1. 创建一个future对象,用来衔接前一个task
future = loop.create_future()
# 2. 添加一个延迟执行的回调函数futures._set_result_unless_cancelled 到当前loop对象的_scheduled二叉堆中,这个堆中的回调函数按照delay的大小,形成一个最小堆
h = future._loop.call_later(delay,
futures._set_result_unless_cancelled,
future, result)
try:
# 3. 执行 yield from future 语句,此时会调用future的 __iter__() 方法,然后在 yield future 处挂住,返回future本身,self._asyncio_future_blocking = True
return (yield from future)
finally:
h.cancel()
sleep
是一个非原生协程,delay=2
注意这里的第1,2,3步是在sleep
上标记的1,2,3注释
第1步 生成一个新
Future
对象,这个对象不同于跟task
是不同的对象,他们都是Future
类型的对象,因为Task
类继承自Future
类。第2步
loop
对象中注册了一个futures._set_result_unless_cancelled
的延迟回调函数handle
对象,前面介绍过,延迟回调函数handle
对象是放在loop._scheduled
这个最小二叉堆
中的,此时,loop
对象的_scheduled
最小堆中只有一个延迟回调函数handle
。到sleep
中的第2步完成为止,loop._scheduled=[handle(delay_2s__set_result_unless_cancelled)]
,loop._ready=[]
,注意正在执行的handle._run()
的流程还没走完,现在是进入到了sleep
协程中的第2步中。第3步 执行
(yield from future)
会调用future
的__iter__
方法,进入到__iter__
方法中,首先把self._asyncio_future_blocking
赋值为True
了,,然后yield self
,注意,此时cor
协程的执行流程就挂起在了yield
处,返回self
也就是Future
对象自己,也就是说执行result= core.send(None)
最终挂起在新的Future
对象的yield self
处,返回得到了一个Future
对象赋值给result
。即result
就是在sleep()
协程中新生成的一个Future
对象了。
我们看一下Future
对象的这个方法。
class Future:
...
def __iter__(self):
# self.done()返回False,
if not self.done():
self._asyncio_future_blocking = True
# 把Future对象自己返回出去
yield self # This tells Task to wait for completion.
assert self.done(), "yield from wasn't used with future"
return self.result() # May raise too.
if compat.PY35:
__await__ = __iter__ # make compatible with 'await' expression
到此为止,result = core.send(None)
的调用得到了一个Future
对象,然后执行流程继续往下走也就是由于result是Future对象,因此进入到_step
方法的第4.1步,这里看一下代码片段
# 4.1. 如果result是一个future对象时,blocking会被设置成true
else:
result._asyncio_future_blocking = False
# 把_wakeup回调函数设置到此future对象中,当此future对象调用set_result()方法时,就会调用_wakeup方法
result.add_done_callback(self._wakeup)
result.add_done_callback(self._wakeup)
实际上就是把task._wakeup
方法注册到了新Futrue
对象的回调方法列表_callbacks = [task._wakeup,]
中,到此为止,task._step
方法才彻底执行完毕。第一轮_run_once()
的调用执行结束了。此时 loop._stopping = Fasle
,然后继续执行下一轮的_run_once()
。
此时:
cor
协程的执行流程挂起在sleep
协程的中产生的新Future
对象的__iter__
方法的yield
处。cor
协程的执行了cor
中标记的第1,2步,第3,4步未执行。新
Future
对象的_callbacks = [task._wakeup,]
。loop._scheduled=[handle(delay_2s__set_result_unless_cancelled)]
,loop._ready=[]
。
第二轮_run_once()
的调用执行开始
进入到_run_once()
方法中,由于loop._scheduled=[handle(delay_2s__set_result_unless_cancelled)]
注意这里的第1,2,3,4,5步是在_run_once()
上标记的1,2,3,4,5注释
第1,2步的逻辑判断,
timeout = 2
第3步
event_list = self._selector.select(2)
,也就是说阻塞2s中,注意,此时因为我们编写的那个cor
协程是没有io事件的,是一个通过sleep
协程模拟耗时操作的,不涉及到真正的io事件,所以这个时候,selector.select(2)
会完整的阻塞2秒钟。第4步 依次取出
_scheduled
的延迟回调函数handle
,放入到_ready
队列中。第5步 依次从
_ready
队列中取出延迟回调函数handle
,执行handle._run()
。
第5步中的回调函数就是sleep
协程中注册到loop
对象的futures._set_result_unless_cancelled
函数
def _set_result_unless_cancelled(fut, result):
"""Helper setting the result only if the future was not cancelled."""
if fut.cancelled():
return
# fut就是sleep中新生成的Future对象,调用set_result()方法
fut.set_result(result)
Future
对象的set_result
方法
class Future:
...
def set_result(self, result):
if self._state != _PENDING:
raise InvalidStateError('{}: {!r}'.format(self._state, self))
self._result = result
self._state = _FINISHED
# 回调Future对象中添加的所有回调函数
self._schedule_callbacks()
def _schedule_callbacks(self):
callbacks = self._callbacks[:]
if not callbacks:
return
self._callbacks[:] = []
# 依次取出注册到Future对象中的所有回调函数,放入到loop._ready队列中去,等待下一轮 _run_onece的调用时,执行这些回调
for callback in callbacks:
self._loop.call_soon(callback, self)
fut.set_result(result)
--> fut._schedule_callbacks()
--> fut._loop.call_soon(callback, self)
, callback
实际上是新Future
对象的_callbacks
中的task._wakeup
方法,task._wakeup
又被添加到loop._ready
队列中去了。到此为止handle._run()
执行完毕,第二轮的_run_once()
执行完毕。
此时:
cor
协程的执行流程挂起在sleep
协程的中产生的新Future
对象的__iter__
方法的yield
处。新
Future
对象的_callbacks = []
。loop._ready = [handle(task._wakeup)]
,loop._scheduled=[]
。
第三轮_run_once()
的调用执行开始
注意这里的第1,2,3,4,5步是在_run_once()
上标记的1,2,3,4,5注释
第1,2步的逻辑判断,
timeout = 0
。第3步
event_list = self._selector.select(0)
,也就是说立即返回。第4步 由于
loop._scheduled=[]
,因此不执行第4步中的逻辑。第5步 依次从
_ready
队列中取出回调函数handle
,执行handle._run()
执行handle._run()
就是执行task._wakeup()
,又要回到task._wakeup()
代码中看看
class Task(futures.Future):
def _wakeup(self, future):
try:
# future为sleep协程中生成的新的Future对象
future.result()
except Exception as exc:
# This may also be a cancellation.
self._step(exc)
else:
# 这里是关键代码,上次的_step()执行到第一次碰到yield的地方挂住了,此时再次执行_step(),
# 也就是再次执行 result = coro.send(None) 这句代码,也就是从上次yield的地方继续执行yield后面的逻辑
self._step()
self = None # Needed to break cycles when an exception occurs.
调用task._wakeup()
实际上又是执行task._step()
也就是再次执行result = core.send(None)
这行代码,前面提到过,core
协程被挂起在Future
对象的__iter__
方法的yield
处,此时再次执行result = core.send(None)
,就是执行yield
后面的语句
class Future:
...
def __iter__(self):
# self.done()返回False,
if not self.done():
self._asyncio_future_blocking = True
# 把Future对象自己返回出去
yield self # This tells Task to wait for completion.
assert self.done(), "yield from wasn't used with future"
# 调用task._wakeup再次进入到core挂起的地方执行yield后面的语句
return self.result() # May raise too.
if compat.PY35:
__await__ = __iter__ # make compatible with 'await' expression
self.result()
返回的是None
,所以 cor
协程的await asyncio.sleep(2)
返回的是None
,到此为止cor
协程的第三步await asyncio.sleep(2)
才真正的执行完毕,也就是说sleep
协程执行完毕了,然后继续执行cor
协程await下面的语句print('exit cor ...')
最后返回'cor'
,到此为止cor
协程就完全执行完毕了。
async def cor():
print('enter cor ...')
await asyncio.sleep(2) # 上次在这里挂起
print('exit cor ...')
return 'cor'
前面介绍了,原生协程在执行结束时会抛出StopIteration
异常,并且把返回值存放在异常的的value
属性中,因此在task._step()
的第2步捕捉到StopIteration
异常
# 2. coro执行完毕会抛出StopIteration异常
except StopIteration as exc:
if self._must_cancel:
# Task is cancelled right before coro stops.
self._must_cancel = False
self.set_exception(futures.CancelledError())
else:
# exc.value是'cor'
# 调用task.set_result('cor')
self.set_result(exc.value)
task.set_result('cor')
其实就是把task
中的存放的回调函数又放到loop._ready
队列中去,task
中的回调函数就是前面介绍的_run_until_complete_cb
函数。到此为止第3轮的_run_once()
执行完毕。
此时:
cor
协程的执行完毕。新
Future
对象的_callbacks = []
。loop._ready = [handle(_run_until_complete_cb)]
,loop._scheduled=[]
。
第四轮_run_once()
开始执行
注意这里的第1,2,3,4,5步是在_run_once()
上标记的1,2,3,4,5注释
第1,2步的逻辑判断,
timeout = 0
。第3步
event_list = self._selector.select(0)
,也就是说立即返回。第4步 由于
loop._scheduled=[]
,因此不执行第4步中的逻辑。第5步 依次从
_ready
队列中取出回调函数handle
,执行handle._run()
执行handle._run()
就是执行_run_until_complete_cb(task)
def _run_until_complete_cb(fut):
exc = fut._exception
if (isinstance(exc, BaseException)
and not isinstance(exc, Exception)):
# Issue #22429: run_forever() already finished, no need to
# stop it.
return
# fut就是task对象,_loop.stop()就是把loop._stopping赋值为Ture
fut._loop.stop()
loop._stopping
为True
。第4轮_run_once()
执行完毕。
def run_forever(self):
...
try:
events._set_running_loop(self)
while True:
# 第4轮_run_once()结束
self._run_once()
# _stopping为true,跳出循环,run_forever()执行结束
if self._stopping:
break
finally:
...
跳出while
循环, run_forever()
执行结束,run_until_complete()
也就执行完毕了,最后把cor
协程的返回值'cor'返回出来赋值给rst
变量。
到此为止所有整个task
任务执行完毕,loop
循环关闭。
总结
loop._ready
和loop._scheduled
是loop
对象中非常重要的两个属性
loop._ready
是一个双端队列deque
,用来存放调用loop.call_soon
方法时中放入的回调函数。loop._scheduled
是一个最小堆,根据延迟时间的大小构建的最小堆,用来存放调用loop.call_later
时存放的延迟回调函数。loop._scheduled
中有延迟调用函数是,timeout
被赋值为堆顶的延迟函数的等待时间,这样会使得select(timeout)
阻塞等待timeout
秒。到时间后loop._scheduled
中的回调函数最终还是会被转移到loop._ready
队列中去执行。
每一个协程都会被封装到一个task
对象中,task
在初始化时就把task._step
方法添加到loop._ready
队列中,同时添加一个停止loop
的事件循环的回调函数到task._callback
列表中。
task._step
函数就是驱动协程的,里面执行cor.send(None)
,分两种情况:
-
第一种:
cor
中有await
语句cor
执行到await
处,await
的后面如果是另一个协程cor1
,则进入到cor1
中执行,cor1
执行完毕,如果cor1
返回的是None,
则task._step
方法会对result
进行判断,返回None
,执行task.set_result()
,这个方法里
面会调取task._callback
列表中的所有回调方法,依次执行,此时,task._callback
列表中只注册了一个停止loop
事件
循环的回调,此时就调用该回调函数,把loop._stopping
设置成Ture,使得loop停止。如果
cor1
返回的是一个future
对象,也就是task._step
函数执行到cor1
协程返回的一个cor1_future
时,则task._step
方
法会对result
进行判断,返回类型是future
则对cor1_future
添加一个回调函数task._wakeup
,当cor1_future.set_result()
在某一时刻被调用时(比如像sleep协程n秒后被调用,或者正在的IO事件触发的调用),会调用cor1_future
添加的回调函数
也就是task._wakeup
函数,task._wakeup
里面又会调用task._step
来驱动上次cor
停止的位置,也就是cor
的await
处,更准确的说是cor1_future.__iter__
方法的yield self
处,继续
执行yield
后面的语句,await cor1()
才算真正的执行完毕,然后接着执行await cor1()
下面的语句,直到cor
执行完毕时会抛出StopIteration
异常,cor
返回值会保存在StopIteration
异常对象的value
属性中,与上面的逻辑一样,task
会调用之前调用的停止loop
的回调函数,停止loop
循环 -
第二种:
cor
中没有await
语句如果没有await语句,
task._step
执行后,result = cor.send(None)
抛出StopIteration
异常,与
上面的逻辑一样,task
会调用之前调用的停止loop
的回调函数,停止loop
循环
本质上,在一次loop循环中会执行一次result = cor.send(None)
语句,也就是cor
中的用户书写的逻辑,碰到await expr
,再进
入到expr
语句,直到碰到future
的yield
语句,这一次循环的result = cor.send(None)
逻辑才算执行完成了。如果
await
后压根没有返回future
,result = cor.send(None)
则会直接执行完await expr
语句后再接着执行await expr
下面的语句,直到
抛出StopIteration
异常。
通过await future
或者yield from future
语句对cor
的执行逻辑进行分割,yield
之前的语句被封装到task._step
方法中,在一次loop
循环中被调用,yield
之后的逻辑封装在task._wakeup
方法中,在下一次的loop
循环中被执行,而这个task._wakeup
是由future.set_result()
把它注册到loop._ready
队列中的。
sleep
协程模拟耗时IO
操作,通过向loop中注册一个延迟回调函数,明确的控制select(timeout)
中的timeout
超时时间 + future
对象的延迟timeout
秒调用future.set_result()
函数来实现一个模拟耗时的操作。
这个其实每一个真实的耗时的IO
操作都会对应一个future
对象。只不过sleep
中的回调明确的传入了延迟回调的时间,而真实的IO
操作时的future.set_result()
的调用则是由真实的IO
事件,也就是select(timeout)
返回的socket
对象的可读或者可写事件来触发的,一旦有相应的事件产生,就会回调对应的可读可写事件的回调函数,而在这些回调函数中又会去触发future.set_result()
方法。
(上面的总结都是根据本人自己的理解所写,限于本人的表达能力有限,很多表达可能会产生一些歧义,还望各位看官一边看此文一边开启debug调试,来帮助自己理解。)