Flask调用aioserial报There is no current event loop in thread错误如何解决?
问题原因
aioserial初始化时会自动创建asyncio.Lock,而asyncio.Lock默认需要绑定当前线程的事件循环:
- 单独运行测试脚本时,代码执行在Python主线程,默认自带初始化好的asyncio事件循环,所以运行正常
- Flask的
before_first_request钩子默认在独立的工作线程中执行,这个线程默认没有初始化asyncio事件循环,因此触发报错
解决方案
方案1:手动为工作线程绑定事件循环(保留异步串口方案)
修改startserial函数,在初始化aioserial前先给当前线程创建并设置事件循环:
import asyncio @app.before_first_request def startserial(): global aioserialinstance try: # 新增两行代码:创建并绑定事件循环到当前线程 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) portaddress = find_comport(PID_MICROBIT, VID_MICROBIT, 115200) print(portaddress) aioserialinstance = test1() except Exception as e: print(str(e)) print("device not connected or wrong port number")
后续在Flask路由中调用异步串口读写方法时,通过绑定的事件循环执行即可,示例:
@app.route('/get-sensor-data') def get_sensor_data(): loop = asyncio.get_event_loop() # 替换为你自己的异步读串口逻辑 sensor_data = loop.run_until_complete(aioserialinstance.read_async(100)) return {"data": sensor_data.decode().strip()}
方案2:改用同步串口库(适配成本更低)
如果不需要高并发的异步IO操作,对于Microbit的简单控制、数据采集场景,直接用同步的pyserial库完全能满足需求,不需要处理事件循环适配问题:
import serial # 同步初始化串口,直接在Flask钩子中调用无报错 def init_serial(): return serial.Serial(port='COM5', baudrate=115200, timeout=1)
注意事项
Flask如果采用多worker模式部署,全局串口实例会被多个worker重复初始化,导致端口被占用,建议采用单worker部署,或者将串口通信逻辑抽为独立进程,通过本地消息队列和Flask进程交互。
内容的提问来源于stack exchange,提问作者aditya singh
相关产品推荐
相关产品推荐

