为何这份Python多进程示例程序无法正常运行？

问题描述 #

现有一个极简Python多进程程序，预期在一个进程中设置sentinelFlag标记，另一个进程读取并响应该标记：

• generator进程调用ClassA实例的SetSentinel方法，定期将全局StatusHolder实例的sentinelFlag设为True
• listener进程定期检查该标记，若为True则打印信息并重置为False

实际运行中，SetSentinel确实会将标记设为True，但listener始终检测不到该值变化，调试时两个进程中gStatusHolder的对象地址看似一致，却无法共享状态。

原程序代码：

import time
from multiprocessing import Process, Lock


class StatusHolder:
    sentinelFlag = False

gStatusHolder = StatusHolder()

class ClassA:
    def SetSentinel(cls):
        while True:
            time.sleep(10)
            gStatusHolder.sentinelFlag = True
            print("Did you hear that?")


def generator(Alist):
    Alist.SetSentinel()


def listener():
    while True:
        if gStatusHolder.sentinelFlag:
            print("I heard!")
            gStatusHolder.sentinelFlag = False
        time.sleep(1)


def main():
    listenerThread = Process(target=listener)
    listenerThread.start()

    A = ClassA()
    generatorThread = Process(target=generator, args=(A,))
    generatorThread.start()

    while(True):
        pass

if __name__ == "__main__":
    main()

核心原因 #

Python的multiprocessing模块创建子进程时，会采用复制内存空间的方式（Unix下是fork，Windows下是重新导入模块并复制）。每个子进程都拥有独立的内存空间，全局变量gStatusHolder会被复制一份到子进程中：

• generator进程修改的是自己内存空间里的gStatusHolder副本
• listener进程检查的是自己内存空间里的另一个gStatusHolder副本
• 两个副本的对象地址在各自进程的地址空间中看似相同，但实际是完全独立的实例，修改其中一个不会影响另一个

解决方案 #

要实现跨进程状态共享，必须使用Python多进程提供的进程间通信（IPC）机制，以下是两种常用方案：

方案1：使用multiprocessing.Value共享简单变量 #

Value可以创建一个受锁保护的共享值，支持跨进程访问和修改：

import time
from multiprocessing import Process, Value


class ClassA:
    def SetSentinel(self, sentinel_flag):
        while True:
            time.sleep(10)
            sentinel_flag.value = True
            print("Did you hear that?")


def generator(alist, sentinel_flag):
    alist.SetSentinel(sentinel_flag)


def listener(sentinel_flag):
    while True:
        if sentinel_flag.value:
            print("I heard!")
            sentinel_flag.value = False
        time.sleep(1)


def main():
    # 创建共享布尔值，'b'表示类型为布尔
    sentinel_flag = Value('b', False)
    
    listener_process = Process(target=listener, args=(sentinel_flag,))
    listener_process.start()

    a_instance = ClassA()
    generator_process = Process(target=generator, args=(a_instance, sentinel_flag))
    generator_process.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        listener_process.terminate()
        generator_process.terminate()

if __name__ == "__main__":
    main()

方案2：使用multiprocessing.Manager共享复杂对象 #

如果需要共享更复杂的状态（比如自定义类实例），可以使用Manager创建可跨进程访问的对象：

import time
from multiprocessing import Process, Manager


class StatusHolder:
    def __init__(self):
        self.sentinelFlag = False

class ClassA:
    def SetSentinel(self, status_holder):
        while True:
            time.sleep(10)
            status_holder.sentinelFlag = True
            print("Did you hear that?")


def generator(alist, status_holder):
    alist.SetSentinel(status_holder)


def listener(status_holder):
    while True:
        if status_holder.sentinelFlag:
            print("I heard!")
            status_holder.sentinelFlag = False
        time.sleep(1)


def main():
    # 使用Manager创建可共享的StatusHolder实例
    with Manager() as manager:
        status_holder = manager.Namespace()
        status_holder.sentinelFlag = False

        listener_process = Process(target=listener, args=(status_holder,))
        listener_process.start()

        a_instance = ClassA()
        generator_process = Process(target=generator, args=(a_instance, status_holder))
        generator_process.start()

        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            listener_process.terminate()
            generator_process.terminate()

if __name__ == "__main__":
    main()

注意事项 #

• 进程间共享状态必须使用官方提供的IPC机制，不能依赖普通全局变量
• 共享状态的修改需要考虑线程/进程安全，必要时需配合锁（Lock）使用
• Windows系统下，子进程会重新导入主模块，因此全局变量的初始化逻辑需放在if __name__ == "__main__":块中，避免重复创建子进程

内容的提问来源于stack exchange，提问作者Drew S.