如何在Flask应用内运行后台控制任务以满足密室逃脱游戏管控需求

直接采用「单进程内运行Flask + 后台线程跑游戏主循环 + 线程锁保护共享状态」的架构即可，不需要复杂的多进程通信，完全满足你期望的「路由直接访问后台游戏任务变量」的需求，逻辑非常简单易懂。

实现步骤 #

1. 安装依赖 #

只需要两个依赖，执行如下命令安装：
pip install flask pyserial

2. 核心代码示例 #

from flask import Flask, request, jsonify
import threading
import time
import serial
from typing import Dict, Any

# 初始化Flask应用
app = Flask(__name__)

# --------------------------
# 共享状态 + 线程锁（核心）
# --------------------------
# 存储所有游戏状态，路由和游戏线程均可读写
game_state: Dict[str, Any] = {
    "current_stage": 1,
    "sensors": {},
    "is_running": True,
    "manual_skip": None  # 存储来自前端的跳过指令
}
# 线程锁，所有读写game_state、操作串口的地方都要加锁，避免并发冲突
state_lock = threading.Lock()
# 初始化串口，按实际情况修改端口和波特率
ser = serial.Serial("/dev/ttyUSB0", 9600, timeout=1)

# --------------------------
# 游戏主循环（后台线程运行）
# --------------------------
def game_loop():
    global game_state, ser
    while True:
        with state_lock:
            if not game_state["is_running"]:
                break
            # 1. 处理前端下发的手动指令，比如关卡跳过
            if game_state["manual_skip"] is not None:
                target_stage = game_state["manual_skip"]
                # 此处写跳过逻辑，下发指令到串口
                ser.write(f"SKIP:{target_stage}\n".encode())
                game_state["current_stage"] = target_stage
                game_state["manual_skip"] = None  # 消费完指令清空
            
            # 2. 读取串口获取传感器状态
            if ser.in_waiting:
                sensor_data = ser.readline().decode().strip()
                # 解析传感器数据更新到game_state即可
                game_state["sensors"] = parse_sensor_data(sensor_data)
            
            # 3. 运行你原有游戏核心逻辑，判断进度、下发控制指令给设备
            process_game_logic(game_state, ser)
        
        # 每次循环休眠10ms，避免占满CPU，可按实际需求调整
        time.sleep(0.01)

# --------------------------
# Flask 接口定义
# --------------------------
@app.route("/api/status", methods=["GET"])
def get_status():
    with state_lock:
        # 直接返回共享的游戏状态
        return jsonify(game_state)

@app.route("/api/control", methods=["POST"])
def send_control():
    cmd = request.get_json()
    with state_lock:
        # 把指令写到共享状态里，游戏循环会自动消费
        if cmd.get("action") == "skip":
            game_state["manual_skip"] = cmd.get("target_stage")
        elif cmd.get("action") == "reset":
            game_state["current_stage"] = 1
            ser.write(b"RESET\n")
        return jsonify({"code": 0, "msg": "指令已下发"})

if __name__ == "__main__":
    # 启动游戏线程，daemon=True表示主线程退出时自动结束游戏线程
    game_thread = threading.Thread(target=game_loop, daemon=True)
    game_thread.start()
    # 启动Flask服务，生产环境可换用gunicorn等部署，测试阶段直接运行即可
    app.run(host="0.0.0.0", port=5000, debug=False)

注意事项 #

• 代码中的parse_sensor_data和process_game_logic直接复用你原有Windows程序里的逻辑即可，不需要额外修改。
• 所有涉及game_state读写、串口操作的代码，都用with state_lock包裹，不会出现多线程冲突问题。
• 这套方案没有引入额外复杂组件，逻辑直观，维护成本极低，完全满足你对简单性的要求，后续新增功能只需往game_state加字段即可。

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