如何在Python脚本间共享变量？线程间变量同步异常排查

你遇到的问题本质是Python模块导入变量的机制导致的副本隔离：当你用from shared_var import FOO时，Python会在当前模块的命名空间里创建一个FOO的本地副本，而不是直接引用shared_var模块里的原始变量。所以utils.py里的子线程修改的是自己模块内的FOO副本，main.py里的主线程检查的是自己模块内的另一个副本，两者完全独立，自然看不到对方的修改。

下面给你三个不同的解决方案，你可以根据场景选择：

方案1：直接导入模块，通过模块访问变量 #

这是最简单直接的修复方式，所有模块都通过shared_var.FOO来访问原始变量，避免创建本地副本。

修改utils.py：

import time
import shared_var  # 改为导入整个模块

def func_b():
    print('func_b initial FOO value: ', shared_var.FOO)
    for i in range(10):
        shared_var.FOO = str(i)  # 通过模块引用修改原始变量
        print('func_b FOO value: ', shared_var.FOO)
        time.sleep(1)
    shared_var.FOO = 'DONE'
    print('func_b FOO value: ', shared_var.FOO)

修改main.py：

import time
import shared_var  # 改为导入整个模块
from utils import func_b
import threading

def check_FOO_status():
    print('main current FOO: ', shared_var.FOO)
    if shared_var.FOO == 'DONE':
        return True
    else:
        return False

if __name__ == "__main__":
    print('main start FOO value: ', shared_var.FOO)
    t = threading.Thread(target=func_b)
    t.start()
    running = True
    while running:
        if check_FOO_status():
            break
        time.sleep(3)
    print('main end FOO value: ', shared_var.FOO)

这样所有模块操作的都是shared_var模块里的同一个FOO变量，子线程的修改会立即被主线程感知到。

方案2：用可变对象包裹共享变量 #

如果不想修改导入方式，可以把FOO放在一个可变对象（比如字典、列表）里。因为可变对象的引用在导入后不会改变，所有模块操作的都是同一个对象，修改内部值会同步。

修改shared_var.py：

FOO = {'value': 'INIT'}  # 用字典包裹原始值

修改utils.py：

import time
from shared_var import FOO

def func_b():
    print('func_b initial FOO value: ', FOO['value'])
    for i in range(10):
        FOO['value'] = str(i)  # 修改字典内部的value
        print('func_b FOO value: ', FOO['value'])
        time.sleep(1)
    FOO['value'] = 'DONE'
    print('func_b FOO value: ', FOO['value'])

修改main.py：

import time
from shared_var import FOO
from utils import func_b
import threading

def check_FOO_status():
    print('main current FOO: ', FOO['value'])
    if FOO['value'] == 'DONE':
        return True
    else:
        return False

if __name__ == "__main__":
    print('main start FOO value: ', FOO['value'])
    t = threading.Thread(target=func_b)
    t.start()
    running = True
    while running:
        if check_FOO_status():
            break
        time.sleep(3)
    print('main end FOO value: ', FOO['value'])

这种方式不需要改变导入逻辑，只需要把不可变的字符串换成可变容器即可。

方案3：使用线程专用的通信工具（更规范） #

如果你的场景后续会更复杂，推荐使用Python线程库提供的threading.Event或队列来做状态通知，这比手动检查全局变量更高效可靠。

修改shared_var.py：

import threading
FOO = 'INIT'
done_event = threading.Event()  # 创建一个事件对象，用于标记任务完成

修改utils.py：

import time
from shared_var import FOO, done_event

def func_b():
    global FOO
    print('func_b initial FOO value: ', FOO)
    for i in range(10):
        FOO = str(i)
        print('func_b FOO value: ', FOO)
        time.sleep(1)
    FOO = 'DONE'
    done_event.set()  # 设置事件，告诉主线程任务完成
    print('func_b FOO value: ', FOO)

修改main.py：

import time
from shared_var import FOO, done_event
from utils import func_b
import threading

def check_FOO_status():
    print('main current FOO: ', FOO)
    return done_event.is_set()  # 检查事件是否被触发

if __name__ == "__main__":
    print('main start FOO value: ', FOO)
    t = threading.Thread(target=func_b)
    t.start()
    # 两种等待方式：要么用wait()阻塞到事件触发，要么保留原循环检查
    # done_event.wait()  # 更高效的方式，无需轮询
    running = True
    while running:
        if check_FOO_status():
            break
        time.sleep(3)
    print('main end FOO value: ', FOO)

这种方式避免了轮询的开销，而且线程间的状态传递更安全，适合复杂的多线程场景。

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