ESP32 MicroPython Asyncio阻塞问题求助:定时任务与Web服务共存
问题分析与解决方案
问题根源
你的模拟代码无法正常运行的核心原因是:input()是同步阻塞函数,调用它会直接卡住整个asyncio事件循环,导致asyncio.wait_for的定时器无法触发,自然不会输出timeout!。asyncio的事件循环依赖异步可等待对象主动让出控制权,而同步阻塞函数不会做这件事。
解决方案
要实现“定时输出+等待输入”的异步效果,需要把同步的input()替换为异步读取输入的逻辑。下面是修改后的代码,同时适配MicroPython环境:
异步输入实现(兼容标准Python与MicroPython)
import asyncio import sys async def async_input(prompt): # 异步读取标准输入,避免阻塞事件循环 print(prompt, end='', flush=True) reader = asyncio.StreamReader() protocol = asyncio.StreamReaderProtocol(reader) await asyncio.get_event_loop().connect_read_pipe(lambda: protocol, sys.stdin) return (await reader.readline()).decode().strip() async def eternity(): # 用异步输入替代同步input() x = await async_input("Press enter..") print(f"You entered: {x}") async def main(): while True: try: await asyncio.wait_for(eternity(), timeout=1.0) # 输入完成后可执行后续逻辑,示例中直接退出循环 break except asyncio.TimeoutError: print('timeout!') asyncio.run(main())
ESP32 MicroPython 适配提示
针对你原本的ESP32开发需求:
- 定时更新LCD:将
print('timeout!')替换为LCD写入代码,调整timeout为10秒即可实现每10秒更新一次 - 异步Web服务器:使用异步socket或MicroPython的
aioserver库处理客户端请求,避免同步accept()调用阻塞事件循环,确保LCD更新和Web请求响应能同时进行
关键注意事项
- 所有可能阻塞的操作(网络请求、外设读写、IO操作)都要封装为异步函数,保证事件循环能正常调度任务
- MicroPython的
asyncio实现与标准Python有细微差异,若asyncio.run()不可用,可改用loop = asyncio.get_event_loop(); loop.run_until_complete(main())
内容的提问来源于stack exchange,提问作者focaazul
相关产品推荐
相关产品推荐

