如何在并发进程间同步频繁变化的变量？

问题分析 #

你当前的代码无法实现同步的核心原因是：三个独立进程的内存空间相互隔离，游戏进程导入的simon_says只是模块加载时的初始值，后续SimonsSubconsious生成的新决策无法同步到游戏进程；同时SimonsSubconsious仅自行打印决策，没有提供跨进程共享状态的机制。

解决方案 #

通过**进程间通信（IPC）**实现决策状态的共享，以下提供两种可行方案：

方案一：使用共享变量集中管理进程 #

通过multiprocessing.Manager创建跨进程共享的变量，由主脚本统一启动三个进程，确保决策状态实时同步。

创建start_all.py作为主启动脚本：

import multiprocessing
import time
import random

# 潜意识进程：负责生成随机决策
def simons_subconscious(shared_simon_says):
    while True:
        # 1=输出Simon says，0=不输出
        shared_simon_says.value = random.randint(0, 1)
        time.sleep(1)

# 游戏1进程：输出动作指令
def simons_game1(shared_simon_says):
    commands = ['jump', 'sit', 'clap']
    while True:
        for command in commands:
            prefix = "Simon says" if shared_simon_says.value else ""
            print(f"[Game1] {prefix} {command}".strip())
            time.sleep(1)

# 游戏2进程：输出动物叫声指令
def simons_game2(shared_simon_says):
    commands = ['bark', 'moo', 'meow']
    while True:
        for command in commands:
            prefix = "Simon says" if shared_simon_says.value else ""
            print(f"[Game2] {prefix} {command}".strip())
            time.sleep(1)

if __name__ == '__main__':
    with multiprocessing.Manager() as manager:
        # 创建共享整数变量，初始值为0
        shared_simon_says = manager.Value('i', 0)
        
        # 初始化三个进程
        p_subconscious = multiprocessing.Process(target=simons_subconscious, args=(shared_simon_says,))
        p_game1 = multiprocessing.Process(target=simons_game1, args=(shared_simon_says,))
        p_game2 = multiprocessing.Process(target=simons_game2, args=(shared_simon_says,))
        
        # 启动所有进程
        p_subconscious.start()
        p_game1.start()
        p_game2.start()
        
        # 持续运行直到手动中断
        p_subconscious.join()
        p_game1.join()
        p_game2.join()

运行方式：直接执行python start_all.py即可启动三个同步的进程。

方案二：使用文件共享决策状态 #

如果需要保持三个脚本独立运行，可以通过写入/读取本地文件实现状态共享，适合不需要集中管理的场景。

修改SimonsSubconsious.py：

import time
import random

def main():
    while True:
        simon_says = random.randint(0, 1)
        # 将决策写入临时文件
        with open("simon_decision.txt", "w") as f:
            f.write(str(simon_says))
        time.sleep(1)

if __name__ == '__main__':
    main()

修改SimonsGame1.py：

import time

commands = ['jump', 'sit', 'clap']
while True:
    # 读取共享文件中的决策，处理文件不存在或读取失败的情况
    try:
        with open("simon_decision.txt", "r") as f:
            simon_says = int(f.read().strip())
    except (FileNotFoundError, ValueError):
        simon_says = 0
    
    for command in commands:
        prefix = "Simon says" if simon_says else ""
        print(f"[Game1] {prefix} {command}".strip())
        time.sleep(1)

修改SimonsGame2.py：

import time

commands = ['bark', 'moo', 'meow']
while True:
    # 读取共享文件中的决策，处理文件不存在或读取失败的情况
    try:
        with open("simon_decision.txt", "r") as f:
            simon_says = int(f.read().strip())
    except (FileNotFoundError, ValueError):
        simon_says = 0
    
    for command in commands:
        prefix = "Simon says" if simon_says else ""
        print(f"[Game2] {prefix} {command}".strip())
        time.sleep(1)

运行方式：分别启动三个脚本即可，注意保持三个脚本在同一目录下，确保文件读写路径一致。

方案对比 #

• 共享变量方案：同步性好、效率高，适合需要集中控制进程的场景。
• 文件共享方案：无需统一启动脚本，进程完全独立，但存在微小的IO延迟，需处理异常情况。

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