如何从另一个Python程序读取mypy.py中循环递增的var_a变量值？

下面提供几种可行的方法，实现read.py读取运行中mypy.py的var_a变量值：

方法1：基于共享内存管理器（推荐） #

借助Python的multiprocessing.managers实现进程间变量共享，适合本地进程间高效通信。

修改后的mypy.py： #

# mypy.py
import time
from multiprocessing.managers import BaseManager
import threading

var_a = 0

# 定义供外部获取变量的方法
def get_var_a():
    return var_a

# 循环更新变量
def update_var():
    global var_a
    while True:
        var_a += 1
        time.sleep(3)

if __name__ == "__main__":
    # 注册共享管理器
    class VarManager(BaseManager):
        pass
    VarManager.register('get_var_a', callable=get_var_a)
    
    # 启动管理器服务，绑定本地端口和认证密钥
    manager = VarManager(address=('localhost', 50000), authkey=b'secret_key')
    server = manager.get_server()
    
    # 启动变量更新线程，避免阻塞管理器服务
    threading.Thread(target=update_var, daemon=True).start()
    
    # 运行管理器服务
    server.serve_forever()

read.py实现： #

# read.py
from multiprocessing.managers import BaseManager

def main():
    # 对接mypy.py的共享管理器
    class VarManager(BaseManager):
        pass
    VarManager.register('get_var_a')
    
    try:
        manager = VarManager(address=('localhost', 50000), authkey=b'secret_key')
        manager.connect()
        var_a = manager.get_var_a()
        print(f"当前var_a的值：{var_a}")
    except ConnectionRefusedError:
        print("错误：mypy.py未启动或服务未运行")

if __name__ == "__main__":
    main()

方法2：文件中转（最简方案） #

通过文本文件作为中间介质，mypy.py定期写入变量值，read.py读取文件内容。

修改后的mypy.py： #

# mypy.py
import time

def main():
    var_a = 0
    while True:
        var_a += 1
        # 覆盖写入变量值到文件
        with open('var_a_cache.txt', 'w', encoding='utf-8') as f:
            f.write(str(var_a))
        time.sleep(3)
    
if __name__ == "__main__":
    main()

read.py实现： #

# read.py
def main():
    try:
        with open('var_a_cache.txt', 'r', encoding='utf-8') as f:
            content = f.read().strip()
            var_a = int(content)
            print(f"当前var_a的值：{var_a}")
    except FileNotFoundError:
        print("错误：未找到缓存文件，请先启动mypy.py")
    except ValueError:
        print("错误：缓存文件内容无效")

if __name__ == "__main__":
    main()

方法3：Socket通信 #

让mypy.py启动一个TCP服务，read.py作为客户端连接后获取变量值。

修改后的mypy.py： #

# mypy.py
import time
import socket
import threading

var_a = 0

# 更新变量的循环
def update_var():
    global var_a
    while True:
        var_a += 1
        time.sleep(3)

# 处理客户端请求
def handle_client(conn):
    global var_a
    conn.send(str(var_a).encode('utf-8'))
    conn.close()

# 启动TCP服务
def start_server():
    host = 'localhost'
    port = 65432
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind((host, port))
        s.listen()
        while True:
            conn, _ = s.accept()
            # 用线程处理请求，避免阻塞更新循环
            threading.Thread(target=handle_client, args=(conn,), daemon=True).start()

if __name__ == "__main__":
    threading.Thread(target=update_var, daemon=True).start()
    start_server()

read.py实现： #

# read.py
import socket

def main():
    host = 'localhost'
    port = 65432
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.connect((host, port))
            data = s.recv(1024).decode('utf-8')
            var_a = int(data)
            print(f"当前var_a的值：{var_a}")
    except ConnectionRefusedError:
        print("错误：mypy.py未启动或服务未运行")

if __name__ == "__main__":
    main()

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