如何让独立循环的Python脚本与Flask应用实现数据交互？

嘿，我来帮你搞定这个问题！首先得指出你原来代码里的一个小问题：Flask的路由函数hello_world不能直接定义参数myData——Flask的路由函数默认只能接收URL路径里的参数，所以得换个方式共享数据。下面给你几种实用的方案，按需选择：

方案1：整合到Flask后台线程（最简单） #

如果不需要两个完全独立的脚本，直接把数据生成逻辑放到Flask的后台线程里是最省心的方式，用全局变量共享数据即可：

from flask import Flask
import threading
import time

app = Flask(__name__)
myData = 0  # 全局共享变量，存储最新生成的数据

def get_data():
    # 这里替换成你实际的数据生成逻辑，比如调用外部接口、读取传感器等
    return int(time.time()) % 100  # 模拟生成数据

def generate_data_loop():
    global myData
    while True:
        myData = get_data()
        time.sleep(1)  # 控制数据生成的频率，根据实际需求调整

@app.route('/')
def hello_world():
    # 直接返回最新的myData
    return f"当前数据：{str(myData)}"

if __name__ == '__main__':
    # 启动后台数据生成线程，daemon=True表示线程随主程序退出而结束
    data_thread = threading.Thread(target=generate_data_loop, daemon=True)
    data_thread.start()
    app.run(host='0.0.0.0', port=5000)

注意事项： #

• 如果get_data()是耗时操作，或者有多个线程同时修改myData，建议加线程锁保证安全：

  from threading import Lock
  data_lock = Lock()
  
  # 在生成数据时：
  with data_lock:
      myData = get_data()
  
  # 在Flask路由里读取时：
  with data_lock:
      current_data = myData
  return f"当前数据：{str(current_data)}"
  

方案2：用Redis消息队列（适合完全独立的脚本） #

如果你必须保持两个脚本完全独立（比如数据生成脚本和Flask运行在不同进程/机器），用Redis作为中间存储是个靠谱的选择：

第一步：安装Redis #

先安装Redis服务（根据你的操作系统安装，比如Ubuntu用sudo apt install redis-server，Windows可以下载官方安装包），然后安装Python的Redis客户端：

pip install redis

数据生成脚本（独立运行） #

import redis
import time

# 连接本地Redis，根据实际地址修改
r = redis.Redis(host='localhost', port=6379, db=0)

def get_data():
    # 替换成你的实际数据生成逻辑
    return int(time.time()) % 100

while True:
    myData = get_data()
    r.set('latest_data', myData)  # 将数据存入Redis的键值对
    time.sleep(1)

Flask应用脚本（独立运行） #

from flask import Flask
import redis

app = Flask(__name__)
r = redis.Redis(host='localhost', port=6379, db=0)

@app.route('/')
def hello_world():
    # 从Redis读取最新数据
    myData = r.get('latest_data')
    if myData:
        return f"当前数据：{myData.decode('utf-8')}"
    else:
        return "暂无数据"

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

方案3：用文件共享数据（轻量简单场景） #

如果不想依赖额外服务，用本地文件作为数据中转也是可行的，适合低并发的简单场景：

数据生成脚本（独立运行） #

import time
import os

def get_data():
    # 替换成你的实际数据生成逻辑
    return int(time.time()) % 100

while True:
    myData = get_data()
    # 先写入临时文件，再重命名避免读写冲突
    with open('data_temp.txt', 'w') as f:
        f.write(str(myData))
    # 重命名操作是原子性的，不会出现Flask读到半写状态的文件
    os.replace('data_temp.txt', 'data.txt')
    time.sleep(1)

Flask应用脚本（独立运行） #

from flask import Flask
import os

app = Flask(__name__)

@app.route('/')
def hello_world():
    data_file = 'data.txt'
    if os.path.exists(data_file):
        with open(data_file, 'r') as f:
            myData = f.read().strip()
        return f"当前数据：{myData}"
    else:
        return "暂无数据"

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

注意事项： #

• 用文件共享时，一定要用原子性的文件替换操作（比如os.replace），避免Flask读取到不完整的数据。
• 高并发场景下不建议用这种方式，性能和可靠性不如Redis。

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