如何在asyncio任务中实现GNU-readline风格的舒适命令行输入体验

以下提供两类可直接落地的方案，你可以根据现有代码改造量选择：

方案1：修复现有input()+线程方案的 Ctrl-C 异常问题 #

核心思路是捕获到终止信号时，主动往标准输入写入一个换行符，让阻塞的input()调用立即返回，线程自然退出，不需要用户手动按回车。

import asyncio
import readline
import signal
import sys
from threading import Thread

# 业务逻辑任务
async def task1():
    cnt = 0
    while True:
        print(f"业务逻辑运行中 {cnt}", end="\r")
        cnt += 1
        await asyncio.sleep(1)

# 独立线程中的输入逻辑
def input_thread(cmd_queue: asyncio.Queue):
    while True:
        try:
            cmd = input("> ")
            # 跨线程操作asyncio队列需要用call_soon_threadsafe
            asyncio.get_event_loop().call_soon_threadsafe(cmd_queue.put_nowait, cmd)
        except EOFError:
            asyncio.get_event_loop().call_soon_threadsafe(cmd_queue.put_nowait, None)
            break

# CLI处理任务
async def task2(cmd_queue: asyncio.Queue):
    while True:
        cmd = await cmd_queue.get()
        if cmd is None:
            # 收到退出信号，终止所有任务
            for task in asyncio.all_tasks():
                task.cancel()
            break
        print(f"\n收到命令: {cmd}")
        # 此处可增加命令分发逻辑传给task1

# Ctrl-C信号处理函数
def handle_sigint(signum, frame):
    sys.stdin.write("\n")
    sys.stdin.flush()
    asyncio.get_event_loop().call_soon_threadsafe(cmd_queue.put_nowait, None)

if __name__ == "__main__":
    cmd_queue = asyncio.Queue()
    # 注册SIGINT信号监听
    signal.signal(signal.SIGINT, handle_sigint)
    # 启动输入线程，设置daemon=True避免主程序退出时线程卡住
    t = Thread(target=input_thread, args=(cmd_queue,), daemon=True)
    t.start()
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(asyncio.gather(task1(), task2(cmd_queue)))
    except asyncio.CancelledError:
        print("\n程序正常退出")

该方案对现有代码改造量极小，原有基于readline的编辑功能完全保留。

方案2：换用原生async的行编辑实现，彻底规避线程问题 #

如果不想维护线程逻辑，可以使用aioconsole库的ainput()方法，原生支持async/await语法，自带readline全套行编辑功能（方向键、删除键、历史记录等都支持），不需要额外开线程，Ctrl-C可以直接被事件循环捕获。

import asyncio
from aioconsole import ainput

async def task1():
    cnt = 0
    while True:
        print(f"业务逻辑运行中 {cnt}", end="\r")
        cnt += 1
        await asyncio.sleep(1)

async def task2():
    while True:
        try:
            cmd = await ainput("> ")
            print(f"\n收到命令: {cmd}")
            # 此处可增加命令分发逻辑传给task1
        except (KeyboardInterrupt, EOFError):
            # 直接捕获Ctrl-C信号，终止所有任务
            for task in asyncio.all_tasks():
                task.cancel()
            break

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(asyncio.gather(task1(), task2()))
    except asyncio.CancelledError:
        print("\n程序正常退出")

如果需要更复杂的CLI能力（比如自动补全、语法高亮、历史记录持久化），可以换用prompt_toolkit库的async接口，功能更丰富，同样无阻塞线程问题。

内容的提问来源于stack exchange，提问作者VPfB