如何在Python的while循环中使用可变变量值（Flask场景）

你的核心需求是轮询Flask服务器获取active状态，当状态为True时执行业务脚本，执行期间需实时检查状态，一旦变为False立即停止脚本，之后继续轮询。原代码存在my_func无法感知实时状态变化、轮询逻辑不严谨的问题，以下是修正后的实现：

1. 实现服务器请求函数 #

先完善read_active_data，用requests库向Flask服务器发起请求并解析JSON响应：

import requests

def read_active_data():
    try:
        # 替换为你的Flask服务器实际地址
        response = requests.get("http://your-flask-server/check-active")
        response.raise_for_status()  # 捕获HTTP请求错误
        data = response.json()
        return data.get("active", False)  # 解析失败时默认返回False
    except requests.exceptions.RequestException as e:
        print(f"请求服务器出错: {e}")
        return False

2. 修改业务函数my_func #

让my_func在执行过程中实时拉取服务器状态，一旦active变为False立即终止：

import time

def my_func():
    while True:
        # 先检查最新状态，再执行业务逻辑
        active = read_active_data()
        if not active:
            print("检测到active为False，停止业务脚本")
            break
        
        # 这里替换为你的实际业务代码
        print("my_func正在运行...")
        time.sleep(1)  # 业务逻辑执行间隔，可按需调整

3. 优化主轮询逻辑 #

主循环负责定期轮询服务器，触发或终止业务脚本：

def main():
    while True:
        active = read_active_data()
        if active:
            print("检测到active为True，启动业务脚本")
            my_func()
        else:
            print("当前无新订单，等待下一次轮询...")
        time.sleep(5)  # 轮询间隔，可按需调整

if __name__ == "__main__":
    main()

关键修改说明 #

• 实时状态感知：my_func内部每次循环都会重新请求服务器获取最新状态，确保能及时响应状态变化。
• 容错处理：在请求函数中加入异常捕获，避免服务器故障导致程序直接崩溃。
• 逻辑分离：主循环负责轮询触发，my_func专注业务执行与停止判断，代码结构更清晰。

多线程优化方案（适合长耗时业务） #

如果你的业务逻辑执行时间较长，不想等待业务完成再检查状态，可以用多线程实现：主线程持续轮询状态，子线程执行业务，状态变化时终止子线程。

import threading
import requests
import time

def read_active_data():
    try:
        response = requests.get("http://your-flask-server/check-active")
        response.raise_for_status()
        return response.json().get("active", False)
    except requests.exceptions.RequestException as e:
        print(f"请求失败: {e}")
        return False

# 业务函数通过事件对象接收停止信号
def my_func(stop_event):
    while not stop_event.is_set():
        # 替换为你的业务逻辑
        print("my_func正在运行...")
        time.sleep(1)

def main():
    stop_event = threading.Event()
    while True:
        active = read_active_data()
        if active:
            if not stop_event.is_set():
                print("启动业务脚本")
                thread = threading.Thread(target=my_func, args=(stop_event,))
                thread.start()
        else:
            if not stop_event.is_set():
                print("检测到active为False，停止业务脚本")
                stop_event.set()
                thread.join()
                stop_event.clear()  # 重置事件，准备下一次启动
        time.sleep(2)  # 轮询间隔

if __name__ == "__main__":
    main()

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