如何在Python中实现变量变更时跨文件通知主程序？

方案1：用观察者模式封装配置变量 #

把config.py里的offline改成带回调触发的属性，任何地方修改它时，都会自动执行提前注册好的回调函数（比如main.py里的doStuffWhenOfflineVarChanges）。

修改config.py：

_offline = True
# 存储需要触发的回调函数列表
offline_callbacks = []

def set_offline(new_state):
    global _offline
    if _offline != new_state:
        _offline = new_state
        # 遍历执行所有注册的回调
        for callback in offline_callbacks:
            callback(new_state)

def get_offline():
    return _offline

main.py里注册回调：

import config as cfg
import request as req

def doStuffWhenOfflineVarChanges(newState):
    print(f"离线状态变更为: {newState}")

# 把回调函数注册到config的回调列表里
cfg.offline_callbacks.append(doStuffWhenOfflineVarChanges)

request.py里修改状态时调用封装好的方法：

import config as cfg
import requests
from datetime import datetime

def logEntrance(barCode, noOfGuests, dt=datetime.now()):
    date = dt.strftime("%Y-%m-%d")
    time = dt.strftime("%H:%M:%S")
    headers = {'Content-type': 'application/json', 'Authorization': cfg.auth}

    url = 'https://xxxxx.xxxxxxxx.xxx/' + cfg.customerId + '/api.ashx'
    params = {"inputtype": "logentrances"}
    pl = [{"Barcode": barCode ,  "PoeID": cfg.poeId,  "UserID": cfg.userId,  "EntranceDate": date,  "EntranceTime": time,  "NoOfGuests": str(noOfGuests),  "OfflineMode": cfg.get_offline()}]

    print(pl)

    try:
        r = requests.post(url, json=pl, params=params, headers=headers)
        print(r.status_code)
        # 请求成功后可以考虑把状态改回False
        # cfg.set_offline(False)
    except:
        cfg.set_offline(True)
    return r

方案2：main中轮询检测状态变化 #

如果场景对实时性要求不高，直接在main.py里开个线程，定期检查offline的状态，发现变化就执行对应操作。

修改main.py：

import config as cfg
import request as req
import time
from threading import Thread

def doStuffWhenOfflineVarChanges(newState):
    print(f"离线状态变更为: {newState}")

def monitor_offline_state():
    last_state = cfg.offline
    while True:
        current_state = cfg.offline
        if current_state != last_state:
            doStuffWhenOfflineVarChanges(current_state)
            last_state = current_state
        time.sleep(1)  # 每秒检查一次，可根据需求调整间隔

# 启动后台监控线程
Thread(target=monitor_offline_state, daemon=True).start()

# 这里写你的主逻辑代码

这个方案改动最小，缺点是状态变更会有延迟，且占用少量系统资源。

方案3：用事件队列传递状态变更 #

借助Python的queue模块，main.py监听队列，request.py修改offline后往队列发送状态变更消息，main.py收到消息就执行操作。

修改main.py：

import config as cfg
import request as req
from queue import Queue
from threading import Thread

# 创建全局状态队列
state_queue = Queue()

def doStuffWhenOfflineVarChanges(newState):
    print(f"离线状态变更为: {newState}")

def handle_queue_messages():
    while True:
        new_state = state_queue.get()
        doStuffWhenOfflineVarChanges(new_state)
        state_queue.task_done()

# 启动队列处理线程
Thread(target=handle_queue_messages, daemon=True).start()

# 把队列传给request模块
req.state_queue = state_queue

修改request.py：

import config as cfg
import requests
from datetime import datetime

state_queue = None  # 由main.py初始化赋值

def logEntrance(barCode, noOfGuests, dt=datetime.now()):
    # ... 原有代码 ...
    try:
        r = requests.post(url, json=pl, params=params, headers=headers)
        print(r.status_code)
        # 请求成功恢复在线时发送消息
        # if cfg.offline:
        #     cfg.offline = False
        #     state_queue.put(False)
    except:
        # 只在状态从False变True时发送消息
        if not cfg.offline:
            cfg.offline = True
            state_queue.put(True)
    return r

这个方案解耦性更强，适合复杂的多模块交互场景。

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