如何在执行器任务仍在等待控制台输入的情况下干净地退出 python 异步应用程序

Question

我正在尝试向异步应用程序添加控制台输入侦听器，但遇到了一个有趣的问题。

以下代码不会在应终止时终止，而是在脚本终止后继续等待输入：

import asyncio


loop = asyncio.get_event_loop()

async def quit_after(timeout):
    await asyncio.sleep(timeout)
    print('main function terminating now') # this could simulate a crash, or a proper exit

async def print_input_loop():
    while True:
        # executor thread is daemon - should terminate when script ends, but doesn't
        text = await loop.run_in_executor(None, input, '> ')
        print(text)

if __name__ == '__main__':
    loop.create_task(print_input_loop())

    try:
        loop.run_until_complete(quit_after(5))
    finally:
        loop.close()

    print('reached end of file - script should now terminate')

我研究了其他异步获取输入的方法，例如在sys.stdin上使用add_reader()，但这在 Windows上不起作用，我需要我的解决方案独立于平台。

有没有人知道一种更好的异步获取控制台输入的方法，当应用程序停止时会优雅地停止，或者一种修复我现有代码以便它干净地退出的方法？

Answer 1

一种可能的解决方法是禁用并发中线程的退出功能。futures ：

def disable_exit_for_threadpool_executor():
    import atexit
    import concurrent.futures
    atexit.unregister(concurrent.futures.thread._python_exit)

此外，aioconsole还提供了跨平台功能，用于处理 asyncio 中的控制台输入：

import aioconsole

async def echo_loop():
    while True:
        text = await aioconsole.ainput('> ')
        print(text.strip())