如何在两个Python脚本间共享实时更新的变量值？

嘿，这个实时变量同步的需求我之前做3D物体模拟的时候也碰到过，给你几个实用的实现方案，你可以根据自己的场景来选：

如果两个脚本是在同一台机器上的不同进程，用multiprocessing模块的共享内存对象是最高效的——它直接在内存中共享数据，没有额外IO开销，还能保证数据的原子性，确保读取到的和生产者的当前值完全一致。

生成变量的脚本（producer.py）： #

from multiprocessing import Value, Process
import time

def update_zp(zp):
    while True:
        # 还原你的givalue逻辑
        if zp.value > 0.4:
            zp.value = 0.1
        else:
            zp.value += 0.05  # 模拟坐标实时变化
        print(f"更新zp值为: {zp.value:.2f}")
        time.sleep(0.1)  # 模拟更新频率

if __name__ == "__main__":
    # 创建共享浮点型变量，初始值0.0
    zp_shared = Value('d', 0.0)
    # 启动更新进程
    p = Process(target=update_zp, args=(zp_shared,))
    p.start()
    p.join()

读取变量的脚本（consumer.py）： #

from multiprocessing import Manager, Process
import time

def read_zp(zp):
    while True:
        # 直接读取共享内存中的实时值
        current_zp = zp.value
        print(f"读取到zp值为: {current_zp:.2f}")
        time.sleep(0.1)  # 和更新频率匹配

def update_zp(zp):
    while True:
        if zp.value > 0.4:
            zp.value = 0.1
        else:
            zp.value += 0.05
        print(f"更新zp值为: {zp.value:.2f}")
        time.sleep(0.1)

if __name__ == "__main__":
    # 用Manager实现跨脚本共享（分开运行两个脚本也能生效）
    manager = Manager()
    zp_shared = manager.Value('d', 0.0)
    
    # 启动生产者和消费者进程
    producer_process = Process(target=update_zp, args=(zp_shared,))
    consumer_process = Process(target=read_zp, args=(zp_shared,))
    
    producer_process.start()
    consumer_process.start()
    
    producer_process.join()
    consumer_process.join()

队列自带线程/进程安全的同步机制，每次读取的都是生产者最新提交的完整值，适合需要确保“每次读取都是有效状态”的场景——比如你的物体坐标更新是一个完整操作，不想读取到一半更新的数据。

生成变量的脚本（producer.py）： #

from multiprocessing import Queue, Process
import time

def update_zp(queue):
    zp = 0.0
    while True:
        if zp > 0.4:
            zp = 0.1
        else:
            zp += 0.05
        # 只保留最新值，先清空旧值再放入新值
        if not queue.empty():
            queue.get_nowait()
        queue.put(zp)
        print(f"更新zp值为: {zp:.2f}")
        time.sleep(0.1)

if __name__ == "__main__":
    zp_queue = Queue(maxsize=1)  # 队列容量设为1，只存最新值
    p = Process(target=update_zp, args=(zp_queue,))
    p.start()
    p.join()

读取变量的脚本（consumer.py）： #

from multiprocessing import Manager, Process
import time

def read_zp(queue):
    while True:
        if not queue.empty():
            current_zp = queue.get()
            print(f"读取到zp值为: {current_zp:.2f}")
        time.sleep(0.1)

def update_zp(queue):
    zp = 0.0
    while True:
        if zp > 0.4:
            zp = 0.1
        else:
            zp += 0.05
        if not queue.empty():
            queue.get_nowait()
        queue.put(zp)
        print(f"更新zp值为: {zp:.2f}")
        time.sleep(0.1)

if __name__ == "__main__":
    manager = Manager()
    zp_queue = manager.Queue(maxsize=1)
    
    producer_process = Process(target=update_zp, args=(zp_queue,))
    consumer_process = Process(target=read_zp, args=(zp_queue,))
    
    producer_process.start()
    consumer_process.start()
    
    producer_process.join()
    consumer_process.join()

如果不想用多进程库，也可以用本地文件存储变量，同时用文件锁确保读写互斥，避免两个脚本同时操作导致数据不一致。这个方法延迟比前两种高，适合更新频率不高的场景。

生成变量的脚本（producer.py）： #

import fcntl
import time

def update_zp():
    zp = 0.0
    while True:
        if zp > 0.4:
            zp = 0.1
        else:
            zp += 0.05
        # 加排他锁写入，阻止其他进程同时操作
        with open("zp_value.txt", "w") as f:
            fcntl.flock(f, fcntl.LOCK_EX)
            f.write(str(zp))
            fcntl.flock(f, fcntl.LOCK_UN)
        print(f"更新zp值为: {zp:.2f}")
        time.sleep(0.1)

if __name__ == "__main__":
    update_zp()

读取变量的脚本（consumer.py）： #

import fcntl
import time

def read_zp():
    while True:
        try:
            # 加共享锁读取，允许其他进程读但阻止写
            with open("zp_value.txt", "r") as f:
                fcntl.flock(f, fcntl.LOCK_SH)
                zp_str = f.read().strip()
                fcntl.flock(f, fcntl.LOCK_UN)
            if zp_str:
                current_zp = float(zp_str)
                print(f"读取到zp值为: {current_zp:.2f}")
        except FileNotFoundError:
            print("等待生产者初始化文件...")
        time.sleep(0.1)

if __name__ == "__main__":
    read_zp()

内容的提问来源于stack exchange，提问作者Muhammad Bilal Abbasi