此篇文章關鍵闡述了PythonAsyncio生產調度基本原理詳細信息，Python.Asyncio是1個專而精的庫，它包括一些功效，而跟關鍵生產調度有關的思路除開三類可在等待目標外，還有其他某些功效，他們各自坐落于runners.py，base_event.py，event.py3個文檔中

　　序言

　　在本文《PythonAsyncio中Coroutines,Tasks,Future可在等待對象關聯及功效》中闡述了Python的可在等待目標功效，尤其是Task目標在運行的時候也可以自我驅動，但一個Task目標只有推動1條實行鏈，如果想好幾條鏈實行（高并發），也是需要EventLoop來制定推動，下面將采取Python.Asyncio庫的源代碼去了解EventLoop是怎樣運轉的。

　　1.簡單介紹

　　Python.Asyncio是1個專而精的庫，它包括一些功效，而跟關鍵生產調度有關的思路除開三類可在等待目標外，還有其他某些功效，他們各自坐落于runners.py，base_event.py，event.py3個文檔中。

　　runners.py文件有個最主要的類--Runner，它工作職責是保證進到協同程序方式的事件循環直到復位工作中，及在撤出協同程序方式時清除仍在運行內存的協同程序，制作器等目標。

　　協同程序方式僅僅是為了能便于了解，對電子計算機來說，并沒那樣區別

　　event.py文件除開儲放著EventLoop對象插口及其得到和設定EventLoop的函數公式外，兩個EventLoop可生產調度對象，分別是Handler和TimerHandler，他們能夠稱之為EvnetLoop啟用其他對象器皿，用以連招待生產調度目標和事件循環之間的關系，但是它們完成比較簡單，對Handler，它源代碼如下所示:

　　#早已移除了個別不愿關的編碼
　　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)

　　通過源碼可以發現，Handle功能十分簡單，提供了可以被取消以及可以在自己所處的上下文執行的功能，而TimerHandle繼承于Handle比Handle多了一些和時間以及排序相關的參數，源碼如下:

　　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

　　通過代碼可以發現，這兩個對象十分簡單，而我們在使用Python.Asyncio時并不會直接使用到這兩個對象，而是通過loop.call_xxx系列方法來把調用封裝成Handle對象，然后等待EventLoop執行。所以loop.call_xxx系列方法可以認為是EventLoop的注冊操作，基本上所有非IO的異步操作都需要通過loop.call_xxx方法來把自己的調用注冊到EventLoop中，比如Task對象就在初始化后通過調用loop.call_soon方法來注冊到EventLoop中，loop.call_sonn的實現很簡單，

　　它的源碼如下：

　　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

　　可以看到call_soon真正相關的代碼只有10幾行，它負責把一個調用封裝成一個Handle，并添加到self._reday中，從而實現把調用注冊到事件循環之中。

　　loop.call_xxx系列函數除了loop.call_soon系列函數外，還有另外兩個方法--loop.call_at和loop.call_later，它們類似于loop.call_soon，不過多了一個時間參數，來告訴EventLoop在什么時間后才可以調用，同時通過loop.call_at和loop.call_later注冊的調用會通過Python的堆排序模塊headpq注冊到self._scheduled變量中，

　　具體代碼如下：

　　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

　　2.EventLoop的調度實現

　　在文章《Python Asyncio中Coroutines,Tasks,Future可等待對象的關系及作用》中已經分析到了runner會通過loop.run_until_complete來調用mainTask從而開啟EventLoop的調度，所以在分析EventLoop的調度時，應該先從loop.run_until_complete入手，

　　對應的源碼如下：

　　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)

　　這段源碼并不復雜，它的主要邏輯是通過把Corotinue轉為一個Task對象，然后通過Task對象初始化時調用loop.call_sonn方法把自己注冊到EventLoop中，最后再通過loop.run_forever中的循環代碼一直運行著，直到_stopping被標記為True:

　　while True:
　　#正真執行任務的邏輯
　　self._run_once()
　　if self._stopping:
　　break
　　可以看出，這段代碼是確保事件循環能一直執行著，自動循環結束，而真正調度的核心是_run_once函數，
　　它的源碼如下:
　　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為當前時間+一個時間單位，猜測是能多處理一些這段時間內產生的事件
　　end_time=self.time()+self._clock_resolution
　　while self._scheduled:
　　handle=self._scheduled[0]
　　if handle._when>=end_time:
　　break
　　handle=heapq.heappop(self._scheduled)
　　handle._scheduled=False
　　self._ready.append(handle)
　　################################################################################
　　#第四階段，遍歷所有準備調度的handle，并且通過handle的context來執行handle對應的callback#
　　################################################################################
　　ntodo=len(self._ready)
　　for i in range(ntodo):
　　handle=self._ready.popleft()
　　#如果handle已經被取消，則不調用
　　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()
　　handle=None#Needed to break cycles when an exception occurs.

　　根據源碼分析，能夠非常明確了解生產調度邏輯性中首先要先整齊self._scheduled，在整齊的一個過程是采用希爾排序去進行的，由于希爾排序在生產調度的場景中高效率非常高，但是這一段整齊編碼分為二種，我猜測是當要關閉的總數太多立即賦值效率更高一些。在整齊self._scheduled后，就進入了第2步，該流程逐漸等候系統軟件事件循環回到相對應的事情，假如self._ready含有數據信息，就不一一等了，必須立刻至下三個步驟，便于能趕快分配生產調度。在獲得系統軟件事件循環所得到的事件之后，就進入了接著，該流程可以通過self._process_events方法解決相對應的事情，然后把事情相對應的調整儲存到self._ready中，接著再賦值self._ready中所有Handle并逐個實行(實行的時候可以覺得EventLoop把管控權回到給相對應的啟用邏輯性)，到此一套完整的生產調度邏輯性就沒有了，并進到下個生產調度邏輯性。

　　3.互聯網IO事件解決

　　注：導致系統事件循環限制，因此材料IO通常依然采用多核來完成，實際見：github.com/python/asyn…

　　在研究EventLoop生產調度完成時忽視了self._process_events的實際完成邏輯性，由于_process_events方法所屬asyncio.base_event.py文件里的BaseEventLoop類并沒有有實際達到的，由于互聯網IO有關的需求全面的事件循環過來幫忙解決，因此與系統軟件事件循環有關的思路都會asyncio.selector_events.py里的BaseSelectorEventLoop類中。BaseSelectorEventLoop類封裝形式了selector模塊與系統軟件事件循環互動，使調用者不用去考慮到sock的建立及其sock形成的文件描述符的監視與銷戶等行為，下邊以BaseSelectorEventLoop中內置的pipe為事例，剖析BaseSelectorEventLoop是如何開始互聯網IO事故處理的。

　　在分析之前，先看一個例子，代碼如下：

　　import asyncio
　　import threading
　　def task():
　　print("task")
　　def run_loop_inside_thread(loop):
　　loop.run_forever()
　　loop=asyncio.get_event_loop()
　　threading.Thread(target=run_loop_inside_thread,args=(loop,)).start()
　　loop.call_soon(task)
　　如果直接運行這個例子，它并不會輸出task（不過在IDE使用DEBUG模式下線程啟動會慢一點，所以會輸出的），因為在調用loop.run_forever后EventLoop會一直卡在這段邏輯中:
　　event_list=self._selector.select(timeout)
　　所以調用loop.call_soon并不會使EventLoop馬上安排調度，而如果把call_soon換成call_soon_threadsafe則可以正常輸出，這是因為call_soon_threadsafe中多了一個self._write_to_self的調用，它的源碼如下：
　　class BaseEventLoop:
　　...
　　def call_soon_threadsafe(self,callback,*args,context=None):
　　"""Like call_soon(),but thread-safe."""
　　self._check_closed()
　　handle=self._call_soon(callback,args,context)
　　self._write_to_self()
　　return handle

　　由于這個調用是涉及到IO相關的，所以需要到BaseSelectorEventLoop類查看，接下來以pipe相關的網絡IO操作來分析EventLoop是如何處理IO事件的(只演示reader對象，writer對象操作與reader類似)，

　　對應的源碼如下：

　　class BaseSelectorEventLoop(base_events.BaseEventLoop):
　　#######
　　#創建#
　　#######
　　def __init__(self,selector=None):
　　super().__init__()
　　if selector is None:
　　#獲取最優的selector
　　selector=selectors.DefaultSelector()
　　self._selector=selector
　　#創建pipe
　　self._make_self_pipe()
　　self._transports=weakref.WeakValueDictionary()
　　def _make_self_pipe(self):
　　#創建Pipe對應的sock
　　self._ssock,self._csock=socket.socketpair()
　　#設置sock為非阻塞
　　self._ssock.setblocking(False)
　　self._csock.setblocking(False)
　　self._internal_fds+=1
　　#阻塞服務端sock讀事件對應的回調
　　self._add_reader(self._ssock.fileno(),self._read_from_self)
　　def _add_reader(self,fd,callback,*args):
　　#檢查事件循環是否關閉
　　self._check_closed()
　　#封裝回調為handle對象
　　handle=events.Handle(callback,args,self,None)
　　try:
　　key=self._selector.get_key(fd)
　　except KeyError:
　　#如果沒有注冊到系統的事件循環，則注冊
　　self._selector.register(fd,selectors.EVENT_READ,
　　(handle,None))
　　else:
　　#如果已經注冊過，則更新
　　mask,(reader,writer)=key.events,key.data
　　self._selector.modify(fd,mask|selectors.EVENT_READ,
　　(handle,writer))
　　if reader is not None:
　　reader.cancel()
　　return handle
　　def _read_from_self(self):
　　#負責消費sock數據
　　while True:
　　try:
　　data=self._ssock.recv(4096)
　　if not data:
　　break
　　self._process_self_data(data)
　　except InterruptedError:
　　continue
　　except BlockingIOError:
　　break
　　#######
　　#刪除#
　　#######
　　def _close_self_pipe(self):
　　#注銷Pipe對應的描述符
　　self._remove_reader(self._ssock.fileno())
　　#關閉sock
　　self._ssock.close()
　　self._ssock=None
　　self._csock.close()
　　self._csock=None
　　self._internal_fds-=1
　　def _remove_reader(self,fd):
　　#如果事件循環已經關閉了，就不用操作了
　　if self.is_closed():
　　return False
　　try:
　　#查詢文件描述符是否在selector中
　　key=self._selector.get_key(fd)
　　except KeyError:
　　#不存在則返回
　　return False
　　else:
　　#存在則進入移除的工作
　　mask,(reader,writer)=key.events,key.data
　　#通過事件掩碼判斷是否有其它事件
　　mask&=~selectors.EVENT_READ
　　if not mask:
　　#移除已經注冊到selector的文件描述符
　　self._selector.unregister(fd)
　　else:
　　#移除已經注冊到selector的文件描述符，并注冊新的事件
　　self._selector.modify(fd,mask,(None,writer))
　　#如果reader不為空，則取消reader
　　if reader is not None:
　　reader.cancel()
　　return True
　　else:
　　return False
　　通過源碼中的創建部分可以看到，EventLoop在啟動的時候會創建一對建立通信的sock，并設置為非阻塞，然后把對應的回調封裝成一個Handle對象并注冊到系統事件循環中（刪除則進行對應的反向操作），之后系統事件循環就會一直監聽對應的事件，也就是EventLoop的執行邏輯會阻塞在下面的調用中，等待事件響應：
　　event_list=self._selector.select(timeout)這時如果執行loop.call_soon_threadsafe，那么會通過write_to_self寫入一點信息：
　　def _write_to_self(self):
　　csock=self._csock
　　if csock is None:
　　return
　　try:
　　csock.send(b'')
　　except OSError:
　　if self._debug:
　　logger.debug("Fail to write a null byte into the self-pipe socket",exc_info=True)

　　由于csock被寫入了數據，那么它對應的ssock就會收到一個讀事件，系統事件循環在收到這個事件通知后就會把數據返回，然后EventLoop就會獲得到對應的數據，并交給process_events方法進行處理，

　　它的相關代碼如下:

　　class BaseSelectorEventLoop:
　　def _process_events(self,event_list):
　　for key,mask in event_list:
　　#從回調事件中獲取到對應的數據，key.data在注冊時是一個元祖，所以這里要對元祖進行解包
　　fileobj,(reader,writer)=key.fileobj,key.data
　　if mask&selectors.EVENT_READ and reader is not None:
　　#得到reader handle，如果是被標記為取消，就移除對應的文件描述符
　　if reader._cancelled:
　　self._remove_reader(fileobj)
　　else:
　　#如果沒被標記為取消，則安排到self._ready中
　　self._add_callback(reader)
　　if mask&selectors.EVENT_WRITE and writer is not None:
　　#對于寫對象，也是同樣的道理。
　　if writer._cancelled:
　　self._remove_writer(fileobj)
　　else:
　　self._add_callback(writer)
　　def _add_callback(self,handle):
　　#把回調的handle添加到_ready中
　　assert isinstance(handle,events.Handle),'A Handle is required here'
　　if handle._cancelled:
　　return
　　assert not isinstance(handle,events.TimerHandle)
　　self._ready.append(handle)
　　def _remove_reader(self,fd):
　　#如果事件循環已經關閉了，就不用操作了
　　if self.is_closed():
　　return False
　　try:
　　#查詢文件描述符是否在selector中
　　key=self._selector.get_key(fd)
　　except KeyError:
　　#不存在則返回
　　return False
　　else:
　　#存在則進入移除的工作
　　mask,(reader,writer)=key.events,key.data
　　mask&=~selectors.EVENT_READ
　　if not mask:
　　#移除已經注冊到selector的文件描述符
　　self._selector.unregister(fd)
　　else:
　　self._selector.modify(fd,mask,(None,writer))
　　if reader is not None:
　　reader.cancel()
　　return True
　　else:
　　return False

　　從編碼中可以看到_process_events會讓事情相對應的文件描述符予以處理，并且從事情調整中掌握到相對應的Handle對象添加到self._ready中，由EventLoop在下面賦值self._ready并實施。

　　能夠看見互聯網IO事件解決不復雜，由于系統軟件事件循環早已給我們做了許多生活了，可是客戶全部與互聯網IO有關實際操作都必須有1個相似的實際操作，那樣是十分繁瑣復雜，幸好asyncio庫已為我們做了封裝形式，我們只需要啟用就行了，便捷許多。

　　綜上所述，這篇文章就給大家介紹到這里了，希望可以為大家帶來幫助。