devlights · February 17, 2023 08:14
diff --git a/asyncio_evloop_thread.py b/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')
	# ------------------------------------------------
	# イベントループを別スレッドで動作させて
	# メインスレッドは生かすサンプル
	#
	# 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')