Asyncio之EventLoop筆記

使用事件循環

Python3.4 采用了一個強大的框架來支持代碼的並發執行: asyncio。這個框架使用事件循環來編排回調和異步任務。
事件循環位於事件循環策略的上下文中-這是 asyncio 所特有的概念。
下圖是協程，事件循環和策略之間的相互作用

協程可以被認為是可以在明確標記有某種語法元素的階段“暫停”的函數.
通過任務對象跟蹤協程的狀態,由相應的事件循環實例化。 事件循環跟蹤當前正在運行的任務，並將 CPU 時間從空閑協程委派給待處理協議。在本章中，我們將更多地了解事件循環接口及其生命周期。將討論事件循環策略-以及全局 asyncio API 對它們的影響。或者和其他異步工作單元(callbacks, promises/futures, and coroutines), 不同的事件循環，但是事件循環是區別於操作系統的。

定位當前正在運行的循環

存在問題

由於各種原因，並發框架必須能夠告訴您事件循環當前是否正在運行以及它是哪一個。例如，您的代碼可能必須斷言只有一個特定的循環實現正在運行您的任務。因此，只有一個任務可以改變某些共享資源或確保將調度您的回調

解決方案

使用全局 asyncio.get_event_loop 和 asyncio.get_running_loop 的 api。
代碼示例 1

import asyncio
loop = asyncio.get_event_loop()

輸出

<_UnixSelectorEventLoop running=False closed=False debug=False>
#windows 輸出
<_WindowsSelectorEventLoop running=False closed=False debug=False>

代碼示例 2

import asyncio
try:
    loop = asyncio.get_running_loop()
except RuntimeError:
    print("No loop running")

在 Python 3.7 中，有兩種有效的方法來獲取當前正在運行的循環實例。
我們可以調用 asyncio.get_event_loop 或 asyncio.get_running_loop
但 asyncio.get_event_loop 內部是做了什么？大概下面幾點
1.檢查在調用函數時是否有循環運行
2.返回其 pid 與當前進程 pid 匹配的運行循環（如果有）
3.如果沒有，獲取存儲在 asynci omodule 中的全局變量中的線程全局 LoopPolicy 實例。
4.如果沒有設置它，則使用鎖用 DefaultLoopPolicy 實例化它。(_init_event_loop_policy 方法)
5.注意，DefaultLoopPolicy 是依賴於操作系統的子類 BaseDefaultEventLoopPolicy，它提供了一個默認的循環實現。獲取被調用的事件循環
6.這是有個問題:僅在主線程上實例化循環並將其分配給線程局部變量時才會使用 loop_policy.get_event_loop 方法。
如果你不在主線程上並且沒有通過其他方式實例化運行循環，則會引發 RuntimeError

這個過程有一些問題

• get_event_loop 檢查是否存在並返回當前運行的循環
• 事件循環策略是全局存儲線程，而循環實例是本地存儲線程
• 如果你在主線程上，get_event_loop 方法將實例化該循環並在策略中本地保存實例線程。
• 如果你不在主線程上，它將引發 RuntimeError
  asyncio.get_running_loop 的工作方式不同。 如果有一個正在運行，它將始終返回當前正在運行的循環實例。 如果沒有，則會引發 RuntimeError。

創建一個新的循環實例

存在問題

由於 asyncio 中的循環與循環策略的概念緊密耦合，因此不建議通過循環構造函數創建循環實例。
否則，我們可能會遇到范圍問題，因為全局 asyncio.get_event_loop 函數只檢索自己創建的循環或通過 asyncio.set_event_loop 設置的循環。

解決方案

要創建一個新的事件循環實例，我們將使用 asyncio.new_event_loop 的 API
注意：此 api 不會更改當前安裝的事件循環，但會初始化（asyncio）全局事件循環策略 - 如果之前未初始化的話。
另一個問題是我們將新創建的循環附加到事件循環策略的觀察程序，以確保我們的事件循環監視 UNIX 系統上新生成的子進程的終止

import asyncio
import sys
loop = asyncio.new_event_loop()
print(loop)  # Print the loop
asyncio.set_event_loop(loop)
if sys.platform != "win32":
    watcher = asyncio.get_child_watcher()
    watcher.attach_loop(loop)

上面的代碼怎么運行的呢
如果從主線程調用，那么 asyncio.get_event_loop 應用程序接口僅實例化該循環
下面是一個循環綁定到線程的例子

import asyncio
import threading
from functools import partial


def _worker(worker, *args, **kwargs):
    # 循環存在於循環策略的上下文中。DefaultLoopPolicy 對每個線程的循環進行限定，
    # 不允許通過 asyncio.get_event_loop 在主線程之外創建循環
    # 因此，我們必須通過 asyncio.set_event_loop（asyncio.new_event_loop（））創建一個線程本地事件循環。
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(worker(*args, **kwargs))
    finally:
        loop.close()


def create_event_loop_thread(worker, *args, **kwargs):
    return threading.Thread(target=partial(_worker, worker), args=args, kwargs=kwargs)


async def print_coro(*args, **kwargs):
    print(f"Inside the print coro on {threading.get_ident()}:", (args, kwargs))


def start_threads(*threads):
    [t.start() for t in threads if isinstance(t, threading.Thread)]


def join_threads(*threads):
    [t.join() for t in threads if isinstance(t, threading.Thread)]


def main():
    workers = [create_event_loop_thread(print_coro) for i in range(10)]
    start_threads(*workers)
    join_threads(*workers)


if __name__ == '__main__':
    main()

將循環附加到進程

使用更高級的 multiprocessing 模塊，我們可以構建一個跨平台的解決方案，在流程本地事件循環中運行多個協程。
這樣我們就可以規避 GIL 強加的 CPython 限制，並利用 asyncio 來提高 I/O 密集型任務的單核 CPU 使用率。

###協程附加到進程
import asyncio
import os
import random
import typing
from multiprocessing import Process

processes = []


def cleanup():
    global processes
    while processes:
        proc = processes.pop()
        try:
            proc.join()
        except KeyboardInterrupt:
            # Ctrl+C 終止進程
            proc.terminate()


async def worker():
    random_delay = random.randint(0, 3)
    result = await asyncio.sleep(random_delay, result=f"Working in process: {os.getpid()}")
    print(result)


def process_main(coro_worker: typing.Callable, num_of_coroutines: int, ):
    """
    在單獨的進程中運行多個協程的進程類。將在每個進程中運行的函數
    建議使用 asyncio.run 而不是實例化自己的事件循環。
     此示例僅用於說明如何在不同進程中實例化事件循環！
    :param coro_worker:
    :param num_of_coroutines:
    :return:
    """
    loop = asyncio.new_event_loop()
    try:
        workers = [coro_worker() for _ in range(num_of_coroutines)]
        loop.run_until_complete(asyncio.gather(*workers, loop=loop))
    except KeyboardInterrupt:
        print(f"Stoping {os.getpid()}")
        loop.stop()
    finally:
        loop.close()


def main(processes, num_procs, num_coros, process_main):
    for _ in range(num_procs):
        proc = Process(target=process_main, args=(worker, num_coros))
        processes.append(proc)
        proc.start()


if __name__ == '__main__':
    try:
        main(processes, 10, 2, process_main, )
    except KeyboardInterrupt:
        print("Ctrl+C 停止運行")
    finally:
        cleanup()
        print("CleanUp finished")

此示例說明如何編寫使用多處理的應用程序。

運行異步代碼而不用擔心循環

如果不想費心修改循環策略和清理異步生成器之后的代碼(您將在下一章中了解它們)，請使用以下代碼。
如果你只有一個線程和進程，並且只有一個協程需要從頭到尾運行，這也很好。

import asyncio
async def main():    
    pass
asyncio.run(main())

在 Python3.6 你可以使用以下方法

import asyncio


async def main():
    pass


loop = asyncio.get_event_loop()
try:
    loop.run_until_complete(main())
finally:
    try:
        # 清理任何沒有完全消耗的異步生成器。
        loop.run_until_complete(loop.shutdown_asyncgens())
    finally:
        loop.close()

如果代碼可能運行在線程中，需要使用下面的方式

import asyncio
import sys


async def main():
    pass


loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
if sys.platform != "win32":
    # 返回當前策略的當前子監視器。
    watcher = asyncio.get_child_watcher()
    # 給一個事件循環綁定監視器。
    # 如果監視器之前已綁定另一個事件循環，那么在綁定新循環前會先解綁原來的事件循環。
    watcher.attach_loop(loop)
    try:
        loop.run_forever()
    finally:
        try:
            loop.run_until_complete(loop.shutdown_asyncgens())
        finally:
            loop.close()

判斷是否只有一個事件循環

import asyncio
async def main(loop):
    assert loop == asyncio.get_running_loop()
    print("ok")

loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))

或者下面這種

import asyncio

async def main():
    pass

loop = asyncio.get_event_loop()
# 通過使用 loop.create_task API，可確保協程將在特定循環上運行。
task = loop.create_task(main())
task.add_done_callback(lambda fut: loop.stop())
loop.run_forever()

停止和關閉循環

import asyncio
import functools


async def main(loop):
    print("Print in main")


def stop_loop(fut, *, loop):
    loop.call_soon_threadsafe(loop.stop)


loop = asyncio.get_event_loop()
tasks = [loop.create_task(main(loop)) for _ in range(10)]
# 為了能夠正確地停止循環，我們需要確保已經消耗了所有任務，因此我們通過調用 asyncio.gather 來包裝它們並向其 add_done_callback，這將關閉我們的循環。
asyncio.gather(*tasks).add_done_callback(functools.partial(stop_loop, loop=loop))
try:
    loop.run_forever()
finally:
    try:
        loop.run_until_complete(loop.shutdown_asyncgens())
    finally:
        loop.close()

添加循環信號處理程序

我們通過 loop.add_signal_handler 添加一個新的信號處理程序。添加信號處理器。它類似於信號 API, 在這種情況下，我們決定在每個處理程序結束時停止循環。
如果要為示例添加另一個處理程序，只需將信號名稱添加到 SIGNAL_NAMES 以及以此方式命名的相應處理程序.

import asyncio
import functools
import os
import signal

SIGNAL_NAMES = ('SIGINT', 'SIGTERM')
SIGNAL_NAME_MESSAGE = " or ".join(SIGNAL_NAMES)


def sigint_handler(signame, *, loop, ):
    print(f"Stopped loop because of {signame}")
    loop.stop()


def sigterm_handler(signame, *, loop, ):
    print(f"Stopped loop because of {signame}")
    loop.stop()


loop = asyncio.get_event_loop()

for signame in SIGNAL_NAMES:
    loop.add_signal_handler(getattr(signal, signame),
                            functools.partial(locals()[f"{signame.lower()}_handler"], signame, loop=loop))

print("Event loop running forever, press Ctrl+C to interrupt.")
print(f"pid {os.getpid()}: send {SIGNAL_NAME_MESSAGE} to exit.")
try:
    loop.run_forever()
finally:
    loop.close()  # optional

為什么不直接使用 signal API 在循環迭代過程中檢查添加到循環中的信號處理程序呢？因為，當它關閉時，不可能向循環添加信號處理程序.另一個好處是，當循環關閉時，信號處理程序會為您清理。

從循環生成子進程

異步生成子流程並在單獨的部分中有效地分割創建和狀態管理是使用循環生成子流程的原因之一。
下面的解決方案對於異步子流程 api 的大多數非交互式使用已經足夠了。
通過在 Windows 系統上設置適當的事件循環策略，它具有跨平台的優點。

import asyncio
import shutil
import sys
from typing import Tuple, Union


async def invoke_command_async(*command, loop, encoding="UTF-8", decode=True) -> Tuple[
    Union[str, bytes], Union[str, bytes], int]:
    """
    Invoke a command asynchronously and return the stdout, stderr and the process return code.
    :param command:
    :param loop:
    :param encoding:
    :param decode:
    :return:
    """
    if sys.platform != 'win32':
        # 如果不是 windows 系統,防止有線程的使用
        asyncio.get_child_watcher().attach_loop(loop)
    process = await asyncio.create_subprocess_exec(*command,
                                                   stdout=asyncio.subprocess.PIPE,
                                                   stderr=asyncio.subprocess.PIPE,
                                                   loop=loop)
    out, err = await process.communicate()

    ret_code = process.returncode

    if not decode:
        return out, err, ret_code

    output_decoded, err_decoded = out.decode(encoding) if out else None, \
                                  err.decode(encoding) if err else None

    return output_decoded, err_decoded, ret_code


async def main(loop):
    # shutil 返回路徑 cmd 里可執行文件的路徑。

    out, err, ret_code = await invoke_command_async(shutil.which("ping"), "-c", "1", "8.8.8.8", loop=loop)
    print(out, err, ret_code)


if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())

loop = asyncio.get_event_loop()
loop.run_until_complete(main(loop))

等待子進程終止

為了確保我們可以在 Windows 下等待子進程的終止，我們將輪詢子進程以獲得進程返回代碼，該代碼指示已終止的子進程。

import asyncio

# Quote from https://docs.python.org/3/library/asyncio-subprocess.html:
# 在從其他線程執行子進程之前，必須在主線程中實例化子監視器
# 調用主線程中的 get_child_watcher（）函數來實例化子監視器
import functools
import shutil
import sys

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())


def stop_loop(*args, loop, **kwargs):
    loop.stop()


async def is_windows_process_alive(process, delay=0.5):
    """
    On windows the signal API is very sparse, meaning we don't have SIGCHILD.
    So we just check if we have a return code on our process object.
    :param process:
    :param delay:
    :return:
    """
    while process.returncode is None:
        await asyncio.sleep(delay)


async def main(process_coro, *, loop):
    process = await process_coro
    print(process)
    if sys.platform != "win32":
        child_watcher: asyncio.AbstractChildWatcher = asyncio.get_child_watcher()
        # 觀察者連接到循環並方便地為我們調用 watcher.add_child_handler
        # 注冊一個新的子處理回調函數。

        child_watcher.add_child_handler(process.pid, functools.partial(stop_loop, loop=loop))
    else:
        await is_windows_process_alive(process)
        loop.stop()


loop = asyncio.get_event_loop()

process_coro = asyncio.create_subprocess_exec(shutil.which("ping"), "-c", "1", "127.0.0.1",
                                              stdout=asyncio.subprocess.DEVNULL,
                                              stderr=asyncio.subprocess.DEVNULL)

loop.create_task(main(process_coro, loop=loop))
loop.run_forever()