如何用asyncio实现异步可选用户输入？我的代码存在什么问题？

你的代码核心问题有两个：

1. 使用time.sleep阻塞事件循环 #

asyncio的事件循环基于单线程调度，time.sleep(1)会直接阻塞整个线程，导致skip_stop协程里的await aioconsole.ainput()完全没机会被调度执行——事件循环在sleep期间彻底停摆，无法处理用户输入。

2. 协程任务被过早终止 #

你创建skip_stop_task后，立刻调用阻塞的sleep，随后马上设置stop_event，这意味着skip_stop协程还没来得及等待用户输入，就因为stop_event.is_set()变为True而退出循环，自然不会触发User Entered日志。

修复后的代码 #

import aioconsole
import logging
import asyncio
# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


async def skip_stop(stop_event=None):
    try:
        # 等待用户输入，同时监听stop_event
        user_input = await aioconsole.ainput()
        logging.info(f"User Entered `{user_input}`")
        return 'Hey Stop please'
    except asyncio.CancelledError:
        logging.info("No User Input found, closing skip stop coroutine")


async def main():
    i = 0
    while i < 10:
        skip_stop_task = asyncio.create_task(skip_stop())
        
        # 使用asyncio.sleep替代time.sleep，避免阻塞事件循环
        await asyncio.sleep(1)
        
        # 检查任务是否完成：若未完成则取消任务
        if not skip_stop_task.done():
            skip_stop_task.cancel()
            try:
                await skip_stop_task
            except asyncio.CancelledError:
                pass
        else:
            # 任务已完成（用户输入内容），获取结果并退出
            result = await skip_stop_task
            if result == 'Hey Stop please':
                return
        
        i += 1

if __name__ == '__main__':
    asyncio.run(main())

关键修改点 #

• 把time.sleep(1)替换为await asyncio.sleep(1)，让事件循环在等待期间可以调度其他协程（比如skip_stop处理用户输入）。
• 移除了stop_event，改用任务的cancel()方法终止未完成的协程，逻辑更简洁。
• 增加了对CancelledError的捕获，处理协程被取消时的日志输出。

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