<think>我们遇到了一个NameError，提示ThreadPoolExecutor未定义。这是因为我们使用了ThreadPoolExecutor，但没有导入相应的模块。 解决方案：从concurrent.futures模块导入ThreadPoolExecutor。 修改代码： 在文件开头添加导入语句： from concurrent.futures import ThreadPoolExecutor 同时，我们之前重构的代码中使用了ThreadPoolExecutor，但未导入，所以需要添加。 另外，我们还需要注意，在定义executor之前，确保ThreadPoolExecutor已经被导入。 由于错误发生在程序启动时，所以需要立即修复。 我们将在导入区域添加以下代码： from concurrent.futures import ThreadPoolExecutor 然后，在定义executor的地方，确保它是在导入之后。 同时，我们还需要注意，在程序的其他地方，我们可能也使用了executor，所以这个全局的executor应该被正确初始化。 修改后的代码片段如下： </think> 这个错误是因为 `ThreadPoolExecutor` 没有被正确导入。让我提供一个完整的修复方案，包括必要的导入和线程池管理： ### 解决方案：修复 ThreadPoolExecutor 未定义错误 ```python # 在文件顶部添加必要的导入 from concurrent.futures import ThreadPoolExecutor import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[logging.StreamHandler()] ) # 在全局变量区域添加 executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="AsyncWorker") event_loop = None background_tasks = set() def start_event_loop(): """启动事件循环的线程函数""" global event_loop event_loop = asyncio.new_event_loop() asyncio.set_event_loop(event_loop) event_loop.run_forever() logging.info("事件循环线程已启动") # 在 __main__ 部分启动事件循环线程 if __name__ == "__main__": # 启动事件循环线程 event_loop_thread = threading.Thread(target=start_event_loop, daemon=True) event_loop_thread.start() logging.info("【主程序】事件循环线程已启动") # 其他初始化代码... # ... # 启动Flask app.run(host="0.0.0.0", port=5000, debug=False) ``` ### 完整修复后的异步任务函数 ```python def _schedule_enter_shelves(phone: str): """调度进入货架任务（异步）""" async def task(): try: await asyncio.sleep(1) goods_shelves_names = _get_goods_shelves_names() logging.info(f"货架列表：{goods_shelves_names}") if goods_shelves_names: for name in goods_shelves_names: logging.info(f"进入货架：{name}") await asyncio.get_running_loop().run_in_executor( executor, lambda n=name: message_publisher.publish_enter_shelve(n, phone) ) await asyncio.sleep(0.1) except Exception as e: logging.error(f"进入货架任务出错：{e}", exc_info=True) # 安全地提交到事件循环 asyncio.run_coroutine_threadsafe(task(), event_loop) def _schedule_leave_shelves_then_warehouse(name: str, phone: str): """调度离开货架和仓库任务（异步）""" async def task(): try: goods_shelves_names = _get_goods_shelves_names() logging.info(f"准备发布离开货架，货架列表：{goods_shelves_names}") if goods_shelves_names: for shelf_name in goods_shelves_names: logging.info(f"离开货架：{shelf_name}") await asyncio.get_running_loop().run_in_executor( executor, lambda n=shelf_name: message_publisher.publish_leave_shelve(n, phone) ) await asyncio.sleep(0.1) # 延迟 1 秒后发布离开仓库 await asyncio.sleep(1) logging.info(f"离开仓库：{name}") await asyncio.get_running_loop().run_in_executor( executor, lambda: message_publisher.publish_leave_warehouse(name, phone) ) except Exception as e: logging.error(f"Error scheduling leave shelves/warehouse: {e}", exc_info=True) # 安全地提交到事件循环 asyncio.run_coroutine_threadsafe(task(), event_loop) ``` ### 修改串口读取线程中的调用 ```python def read_data_sync(): global last_door, port, last_change_time, last_printed_door, face_status, stop_read_thread # ... [其他代码不变] ... if '1' in new_door: logging.info("【检测到关门】") try: retrieved_data = load_retrieved_data() # 使用重构后的函数 _schedule_leave_shelves_then_warehouse( retrieved_data['name'], retrieved_data['phone'] ) except Exception as e: logging.error(f"投递离开任务失败: {e}", exc_info=True) face_status = False else: logging.info("【检测到开门】") try: retrieved_data = load_retrieved_data() # 使用线程池执行同步操作 executor.submit( lambda: message_publisher.publish_enter_warehouse( retrieved_data['name'], retrieved_data['phone'] ) ) # 使用重构后的函数 _schedule_enter_shelves(retrieved_data['phone']) face_status = True except Exception as e: logging.error(f"投递进入任务失败: {e}", exc_info=True) # ... [其他代码不变] ... ``` ### 添加程序退出时的清理逻辑 ```python import atexit def cleanup(): """程序退出时的清理函数""" logging.info("执行清理操作...") # 停止事件循环 if event_loop and event_loop.is_running(): logging.info("停止事件循环...") event_loop.call_soon_threadsafe(event_loop.stop) # 关闭线程池 logging.info("关闭线程池...") executor.shutdown(wait=True, cancel_futures=True) # 关闭串口 if port and isinstance(port, serial.Serial) and port.is_open: logging.info("关闭串口...") port.close() logging.info("清理完成") # 注册清理函数 atexit.register(cleanup) if __name__ == "__main__": try: # ... [初始化代码] ... # 启动Flask app.run(host="0.0.0.0", port=5000, debug=False) except KeyboardInterrupt: logging.info("程序被用户中断") except Exception as e: logging.critical(f"程序崩溃: {e}", exc_info=True) finally: cleanup() ``` ### 关键修改说明： 1. **导入 ThreadPoolExecutor**： - 从 `concurrent.futures` 导入 `ThreadPoolExecutor` - 创建全局线程池执行器 2. **日志系统增强**： - 配置结构化日志输出 - 使用 `exc_info=True` 记录异常堆栈 3. **事件循环管理**： - 在独立线程中启动事件循环 - 使用 `asyncio.run_coroutine_threadsafe` 提交任务 4. **资源清理**： - 使用 `atexit` 注册清理函数 - 优雅关闭线程池和事件循环 - 确保串口正确关闭 5. **错误处理增强**： - 添加全局异常处理 - 在关键操作处添加 try-except 块 - 使用线程安全的日志记录