Python Asyncio调度原理详情_python_

# 已经移除了一些不想关的代码 class Handle: def __init__(self, callback, args, loop, context=None): # 初始化上下文，确保执行的时候能找到Handle所在的上下文 if context is None: context = contextvars.copy_context() self._context = context self._loop = loop self._callback = callback self._args = args self._cancelled = False def cancel(self): # 设置当前Handle为取消状态 if not self._cancelled: self._cancelled = True self._callback = None self._args = None def cancelled(self): return self._cancelled def _run(self): # 用于执行真正的函数，且通过context.run方法来确保在自己的上下文内执行。 try: # 保持在自己持有的上下文中执行对应的回调 self._context.run(self._callback, *self._args) except (SystemExit, KeyboardInterrupt): raise except BaseException as exc: cb = format_helpers._format_callback_source( self._callback, self._args) msg = f'Exception in callback {cb}' context = { 'message': msg, 'exception': exc, 'handle': self, } self._loop.call_exception_handler(context) 

class TimerHandle(Handle): def __init__(self, when, callback, args, loop, context=None): super().__init__(callback, args, loop, context) self._when = when self._scheduled = False def __hash__(self): return hash(self._when) def __lt__(self, other): if isinstance(other, TimerHandle): return self._when < other._when return NotImplemented def __le__(self, other): if isinstance(other, TimerHandle): return self._when < other._when or self.__eq__(other) return NotImplemented def __gt__(self, other): if isinstance(other, TimerHandle): return self._when > other._when return NotImplemented def __ge__(self, other): if isinstance(other, TimerHandle): return self._when > other._when or self.__eq__(other) return NotImplemented def __eq__(self, other): if isinstance(other, TimerHandle): return (self._when == other._when and self._callback == other._callback and self._args == other._args and self._cancelled == other._cancelled) return NotImplemented def cancel(self): if not self._cancelled: # 用于通知事件循环当前Handle已经退出了 self._loop._timer_handle_cancelled(self) super().cancel() def when(self): return self._when 

class BaseEventLoop: ... def call_soon(self, callback, *args, context=None): # 检查是否事件循环是否关闭，如果是则直接抛出异常 self._check_closed() handle = self._call_soon(callback, args, context) return handle def _call_soon(self, callback, args, context): # 把调用封装成一个handle，这样方便被事件循环调用 handle = events.Handle(callback, args, self, context) # 添加一个handle到_ready，等待被调用 self._ready.append(handle) return handle

class BaseEventLoop: ... def call_later(self, delay, callback, *args, context=None): if delay is None: raise TypeError('delay must not be None') timer = self.call_at(self.time() + delay, callback, *args, context=context) return timer def call_at(self, when, callback, *args, context=None): if when is None: raise TypeError("when cannot be None") self._check_closed() # 创建一个timer handle，然后添加到事件循环的_scheduled中，等待被调用 timer = events.TimerHandle(when, callback, args, self, context) heapq.heappush(self._scheduled, timer) timer._scheduled = True return timer

class BaseEventLoop: def run_until_complete(self, future): ... new_task = not futures.isfuture(future) # 把coroutine转换成task，这样事件循环就可以调度了，事件循环的最小调度单位为task # 需要注意的是此时事件循环并没注册到全局变量中，所以需要显示的传进去， # 同时Task对象注册的时候，已经通过loop.call_soon把自己注册到事件循环中，等待调度 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 # 当该task完成时，意味着当前事件循环失去了调度对象，无法继续调度，所以需要关闭当前事件循环，程序会由协程模式返回到线程模式 future.add_done_callback(_run_until_complete_cb) try: # 事件循环开始运行 self.run_forever() except: if new_task and future.done() and not future.cancelled(): # The coroutine raised a BaseException. Consume the exception # to not log a warning, the caller doesn't have access to the # local task. future.exception() raise finally: future.remove_done_callback(_run_until_complete_cb) if not future.done(): raise RuntimeError('Event loop stopped before Future completed.') return future.result() def run_forever(self): # 进行一些初始化工作 self._check_closed() self._check_running() self._set_coroutine_origin_tracking(self._debug) self._thread_id = threading.get_ident() old_agen_hooks = sys.get_asyncgen_hooks() # 通过asyncgen钩子来自动关闭asyncgen函数，这样可以提醒用户生成器还未关闭 sys.set_asyncgen_hooks(firstiter=self._asyncgen_firstiter_hook, finalizer=self._asyncgen_finalizer_hook) try: # 设置当前在运行的事件循环到全局变量中，这样就可以在任一阶段获取到当前的事件循环了 events._set_running_loop(self) while True: # 正真执行任务的逻辑 self._run_once() if self._stopping: break finally: # 关闭循环, 并且清理一些资源 self._stopping = False self._thread_id = None events._set_running_loop(None) self._set_coroutine_origin_tracking(False) sys.set_asyncgen_hooks(*old_agen_hooks)

while True: # 正真执行任务的逻辑 self._run_once() if self._stopping: break

class BaseEventLoop: ... def _run_once(self): # self._scheduled是一个列表，它只存放TimerHandle sched_count = len(self._scheduled) ############################### # 第一阶段，整理self._scheduled # ############################### if (sched_count > _MIN_SCHEDULED_TIMER_HANDLES and self._timer_cancelled_count / sched_count > _MIN_CANCELLED_TIMER_HANDLES_FRACTION): # 当待调度的任务数量超过100且待取消的任务占总任务的50%时，才进入这个逻辑 # 把需要取消的任务移除 new_scheduled = [] for handle in self._scheduled: if handle._cancelled: # 设置handle的_cancelled为True，并且把handle从_scheduled中移除 handle._scheduled = False else: new_scheduled.append(handle) # 重新排列堆 heapq.heapify(new_scheduled) self._scheduled = new_scheduled self._timer_cancelled_count = 0 else: # 需要取消的handle不多，则只会走这个逻辑，这里会把堆顶的handle弹出，并标记为不可调度，但不会访问整个堆 while self._scheduled and self._scheduled[0]._cancelled: self._timer_cancelled_count -= 1 handle = heapq.heappop(self._scheduled) handle._scheduled = False ################################# # 第二阶段，计算超时值以及等待事件IO # ################################# timeout = None # 当有准备调度的handle或者是正在关闭时，不等待，方便尽快的调度 if self._ready or self._stopping: timeout = 0 elif self._scheduled: # Compute the desired timeout. # 如果堆有数据时，通过堆顶的handle计算最短的超时时间，但是最多不能超过MAXIMUM_SELECT_TIMEOUT，以免超过系统限制 when = self._scheduled[0]._when timeout = min(max(0, when - self.time()), MAXIMUM_SELECT_TIMEOUT) # 事件循环等待事件，直到有事件或者超时 event_list = self._selector.select(timeout) ################################################## # 第三阶段，把满足条件的TimeHandle放入到self._ready中 # ################################################## # 获取得到的事件的回调，然后装填到_ready self._process_events(event_list) # 把一些在self._scheduled且满足调度条件的handle放到_ready中，比如TimerHandle。 # end_time为当前时间+一个时间单位，猜测
                
提示：
                    本文由神整理自网络，如有侵权请联系本站删除！
                    

                    本站声明： 

                    1、本站所有资源均来源于互联网，不保证100%完整、不提供任何技术支持； 

                    2、本站所发布的文章以及附件仅限用于学习和研究目的;不得将用于商业或者非法用途；否则由此产生的法律后果，本站概不负责！
                
                

                

                                            
上一篇：python打印日志方法的使用教程(logging模块)_python_
                                                                
下一篇：python pdfplumber库批量提取pdf表格数据转换为excel_python_