[python][asyncio] イベントループを別スレッドで動作させてメインスレッドは生かすサンプル

asyncio_evloop_thread.py

# ------------------------------------------------
# イベントループを別スレッドで動作させて
# メインスレッドは生かすサンプル
#
# asyncio をつかった処理にて、別スレッドから
# イベントスレッドに対して処理を行う場合
# イベントスレッド自体がスレッドセーフではないため
# 専用のメソッドを利用する必要がある
#
# - ev_loop.call_soon_threadsafe()
# - ev_loop.run_coroutine_threadsafe()
# ------------------------------------------------
import asyncio

from datetime import datetime
from random import Random
from threading import Thread
from time import sleep


async def show(id: int, secs: int):
    """イベントスレッド上で動作する関数 (コルーチン)

    自身に指定された情報を元に待機してから情報出力します。

    Arguments:
        id {int} -- 自分に対して割り当てられた数値
        secs {int} -- 待機する秒数
    """

    print(f'[{id}]: {secs} sec(s) sleep...')
    await asyncio.sleep(secs)
    print(f'[{id}]: {datetime.now().isoformat()}')


def begin_ev_thread(loop: asyncio.AbstractEventLoop):
    """イベントループを開始します

    Arguments:
        loop {asyncio.AbstractEventLoop} -- イベントループ
    """

    # ----------------------------------------------
    # 別スレッド上でイベントループを動かす場合
    # 最初に現在のスレッドに対してイベントループは「これだよ」
    # って教えてあげてから、開始する。
    #
    # ev_loop.run_forever() は ev_loop.stop() が
    # 呼ばれるまでずっと動く。
    #
    # python 3.6 から
    #   ev_loop.shutdown_asyncgens()
    # が追加されているため、close() する前に呼び出しておく
    #
    # 参考：
    #   http://bit.ly/2DAchXi
    # ----------------------------------------------
    asyncio.set_event_loop(loop)
    try:
        print('==> event loop start')
        loop.run_forever()
    finally:
        print('==> event loop close')
        loop.run_until_complete(loop.shutdown_asyncgens())
        loop.close()


def show_dot():
    """進捗状況代わりにドットを出力しつづけるだけの関数
    """

    while True:
        sleep(1)
        print('.')


def do_main_work(secs: int):
    """メイン処理の代わり

    実行すると重い処理が内部で走るイメージ。

    Arguments:
        secs {int} -- 処理にかかる秒数
    """

    print('==> main-thread proc begin')
    sleep(secs)
    print('==> main-thread proc end')


if __name__ == "__main__":
    # イベントループを取得
    ev_loop = asyncio.get_event_loop()

    # ---------------------------------------------------
    # イベントループを別スレッドで起動
    # ---------------------------------------------------
    ev_thread = Thread(
        target=begin_ev_thread, args=(ev_loop,), daemon=False)
    ev_thread.start()

    # ---------------------------------------------------
    # 別スレッドでさらに処理が走っているとする
    # ---------------------------------------------------
    t = Thread(target=show_dot, daemon=True)
    t.start()

    # ---------------------------------------------------
    # コルーチンを生成してイベントループ上で並列実行
    #
    # イベントループは別スレッド上で動作しているので
    # 別スレッドからタスクを投入する場合は
    #
    # - asyncio.run_coroutine_threadsafe()
    #
    # で送り込む。
    #
    # run_coroutine_threadsafe() で返ってくる オブジェクトは
    # Future であるが、asyncio.Future ではなく
    #   concurrent.futures.Future
    # であることに注意
    #
    # 参考：
    #   http://bit.ly/2DAu1Sv
    # ---------------------------------------------------
    rnd = Random()
    futures = [
        asyncio.run_coroutine_threadsafe(show(i, rnd.randint(1, 5)), ev_loop)
        for i in range(5)
    ]

    # ---------------------------------------------------
    # メイン処理を実行（単にsleepしながら出力しているだけだが）
    # ---------------------------------------------------
    do_main_work(10)

    # ---------------------------------------------------
    # イベントループを停止
    # ---------------------------------------------------
    print('==> event loop stop')
    ev_loop.call_soon_threadsafe(ev_loop.stop)
    print('==> main-thread end')