如何在异步任务中拥有舒适的(例如GNU-readline风格)输入行?

2022-03-25 00:00:00 python readline python-asyncio

问题描述

我有一个异步程序,有两个任务:

  • 任务%1做了一些工作
  • 任务2提供命令行界面(CLI),它从用户读取命令并将其发送到任务1进行处理

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

它能用,但不太舒服。问题是输入行没有提供编辑命令,除了在Linux终端级别处理的[backspace ctrl-H]和[ctrl-U],而不是在程序中。我至少需要[左]和[右]箭头以及[删除]。

我在导入readline的情况下尝试了内置input(),这给我带来了舒适的编辑,但它必须在单独的线程中运行,因为它正在阻塞(loop.run_in_executor)。

现在的问题是[ctrl-C]处理(KeyboardInterrupt)。任务已取消,但其自身线程中的input()仍在等待[Enter]键。只有在[Enter]键之后,应用程序才会退出。这太令人困惑了,而且对于这种方法来说,它是一个拦截器错误。遗憾的是,无法终止线程,因此我无法使[ctrl-C]键正常终止程序。

您是否知道如何使用input()解决问题,或者是否知道在异步任务中启用基本输入行编辑的替代方法?


解决方案

您是否考虑过将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的链接是否有可能.

相关文章