如何将while循环中的动态数据传递至其他Python模块？

全局变量确实不是最佳实践，会让代码耦合度高、难以调试，尤其是在多线程/多进程场景下还可能引发并发问题。下面提供两种更合理的实现方式：

方案一：用类封装状态 #

把数据点列表和获取数据的逻辑封装到类中，通过实例属性来维护数据，其他模块只需持有这个类的实例就能访问最新数据。

示例代码 #

import time

class DataFetcher:
    def __init__(self, api_data, max_points=5):
        self.api_data = api_data
        self.max_points = max_points
        self.my_DataPoints = []  # 用实例属性代替全局变量

    def start_fetching(self):
        while True:
            start = time.time()
            # 注意：这里应该调用api_data的方法获取最新数据，而不是直接赋值实例
            # 假设api_data有一个get_latest()方法返回最新数据点
            latest_data = self.api_data.get_latest()
            print(latest_data.rTime)
            self.my_DataPoints.append(latest_data)
            
            # 保持列表长度不超过max_points
            if len(self.my_DataPoints) > self.max_points:
                self.my_DataPoints.pop(0)
            
            print(self.my_DataPoints)
            
            end = time.time()
            if (end - start) < 1:
                time.sleep(1 - (end - start))

在其他模块/函数中使用 #

# 假设在另一个模块或函数中
from your_module import DataFetcher
from api_module import Api_data  # 导入你的Api_data类

# 初始化API实例和数据获取器
api_instance = Api_data()
fetcher = DataFetcher(api_instance)

# 启动数据获取（可以用线程避免阻塞主程序）
import threading
threading.Thread(target=fetcher.start_fetching, daemon=True).start()

# 其他地方需要访问数据时
def generate_json():
    # 获取最新的5个数据点
    latest_points = fetcher.my_DataPoints
    # 生成JSON逻辑
    import json
    return json.dumps([{"rTime": p.rTime} for p in latest_points])

方案二：传递可变容器对象 #

如果不想用类，可以把数据点列表作为参数传递给函数（因为列表是可变对象，函数内的修改会影响外部），这样无需全局变量就能共享数据。

示例代码 #

import time

def get_data(api_data, data_points):
    max_points = 5
    while True:
        start = time.time()
        latest_data = api_data.get_latest()  # 同样需要调用方法获取最新数据
        print(latest_data.rTime)
        data_points.append(latest_data)
        
        if len(data_points) > max_points:
            data_points.pop(0)
        
        print(data_points)
        
        end = time.time()
        if (end - start) < 1:
            time.sleep(1 - (end - start))

在其他模块/函数中使用 #

from your_module import get_data
from api_module import Api_data
import threading

# 初始化共享的列表
my_DataPoints = []
api_instance = Api_data()

# 启动数据获取线程
threading.Thread(target=get_data, args=(api_instance, my_DataPoints), daemon=True).start()

# 其他函数中访问数据
def generate_json():
    import json
    return json.dumps([{"rTime": p.rTime} for p in my_DataPoints])

额外注意事项 #

• 并发安全：如果程序涉及多线程访问my_DataPoints，需要加锁（比如threading.Lock）来避免数据竞争，修改或读取列表时都要加锁。
• 数据更新逻辑：原代码中myData = api_data只是赋值了Api_data的实例，并没有获取最新数据，你需要确保调用正确的方法来获取实时数据（比如api_data.fetch()或api_data.get_latest()）。
• 线程守护：用线程启动数据获取循环时，设置daemon=True可以让主线程退出时自动终止该线程，避免程序无法正常退出。

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