如何在异步任务中拥有舒适的（例如 GNU-readline 样式）输入行？

Question

我有一个包含两个任务的异步程序：

• 任务 1 做了一些工作
• task 2 提供了一个命令行界面 (CLI)，它从用户那里读取命令并将它们发送给 task 1 进行处理

CLI 基本上是一个从连接到标准输入的异步流中读取行的循环。

它有效，但不是很舒服。问题是输入行没有提供除[BACKSPACE ctrl-H] 和[ctrl-U] 之外的编辑命令，它们是在Linux 终端级别而不是在程序中处理的。我至少想要 [LEFT] 和 [RIGHT] 箭头以及 [DEL]。

我尝试了内置 input() 和 readline 导入，这给了我一个舒适的编辑，但它必须在单独的线程中运行，因为它是阻塞的 (loop.run_in_executor)。

现在问题是 [ctrl-C] 处理 (KeyboardInterrupt)。任务被取消，但input() 在它自己的线程中仍然等待[ENTER] 键。只有在 [ENTER] 后，应用程序才会退出。这太令人困惑了，对于这种方法来说，它是一个“阻止程序错误”。不幸的是，不可能杀死一个线程，所以我无法使 [ctrl-C] 按键正常终止程序。

您知道如何解决input() 的问题，或者您知道如何在异步任务中启用基本输入行编辑的另一种方法？

Answer 1

您是否考虑过将input 部分保留在主线程中并在子线程中运行事件循环？ 这可能有点老套，肯定需要改进，但你可以使用你所拥有的：

import asyncio
from threading import Thread

streaming_queue = asyncio.Queue()


async def main_async():
    while True:
        chunk = await streaming_queue.get()
        print("got chunk: ", chunk)


def event_loop(loop):
    try:
        loop.run_until_complete(main_async())
    except asyncio.CancelledError:
        loop.close()
        print("loop closed")


def send_chunk(chunk):
    streaming_queue.put_nowait(chunk)


def cancel_all():
    for task in asyncio.all_tasks():
        task.cancel()


if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    event_thread = Thread(target=event_loop, args=(loop,))
    event_thread.start()
    try:
        while True:
            chunk = input()
            loop.call_soon_threadsafe(send_chunk, chunk)
    except KeyboardInterrupt:
        print("cancelling all tasks")
        loop.call_soon_threadsafe(cancel_all)
        print("joining thread")
        event_thread.join()
        print("done")

最好的当然是input 的异步实现。不确定是否有类似的东西，或者@furas 的链接是否有可能......