You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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开发需求:

  1. 定时更新LCD:将print('timeout!')替换为LCD写入代码,调整timeout为10秒即可实现每10秒更新一次
  2. 异步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

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.06.18 21:33:16