多进程队列Worker类属性调用未按预期运行问题排查

问题根源 #

你遇到的问题本质是多进程内存空间相互隔离导致的：

• 你在主进程中创建的Worker实例，调用start()之后，操作系统会复制一份完整的进程内存空间给子进程，后续子进程中对self.task_type的修改，只会作用于子进程自己的内存副本，主进程里的原Worker实例的属性完全不会同步更新，所以你在主进程调用get_type()永远只能拿到初始化时的空字符串。
• 另外你的代码本身还有两个语法错误：
  • get_type()方法定义没有加self参数，调用会直接报错
  • 导入类错误，multiprocessing没有Worker类，应该导入Process

修复方案 #

使用multiprocessing提供的跨进程共享变量来存储task_type即可，推荐用Manager来创建共享字符串，修改后的完整代码如下：

from multiprocessing import Process, Queue, Manager

class Worker(Process):
    def __init__(self, queue):
        super(Worker, self).__init__()
        self.queue = queue
        # 创建跨进程共享的字符串变量
        manager = Manager()
        self.task_type = manager.Value(str, '')
    
    def get_type(self):
        print(self.task_type.value)
        return self.task_type.value

    def run(self):
        print('Worker started')
        # 初始化操作
        print('Computing things!')
        for data in iter(self.queue.get, None):
            # 对共享变量赋值需要操作.value属性
            self.task_type.value = data['type']
            print(self.task_type.value)  # 打印test
            # 处理数据逻辑

request_queue = Queue()
workers = []
for i in range(4):
    workers.append(Worker(request_queue))

for i in workers:
    i.start()

# 这里the_real_source为你的任务生成逻辑，示例用测试代码代替
the_real_source = [{'type':'test'}, {'type':'normal'}, {'type':'test'}, {'type':'normal'}]
for data in the_real_source:
    request_queue.put(data)
# 哨兵值，每个进程一个用来优雅退出
for i in range(4):
    request_queue.put(None)

# 终止匹配task_type的进程
for i in workers:
    if i.get_type() == 'test':
        i.terminate()
        i.join() # 回收进程资源，避免产生僵尸进程

额外注意点 #

• 共享变量的读写默认是进程安全的，无需额外加锁
• 如果是高频率更新task_type的场景，也可以用multiprocessing.Array来存储字符串，性能比Manager更好

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