Python多进程共享数据：Manager机制解析与实现

云舒

发布时间：2026-10-13 | 3人浏览过

原创

本文介绍Python中利用Manager实现无关联进程间共享复杂数据结构的方法。通过启动ManagerServer监听Socket，Client进程连接后获取共享字典或锁等对象。需注意直接修改客户端获取的变量不会同步至服务端，必须调用服务端方法刷新数据，否则无法实现真正的跨进程状态同步。

多进程通信方案对比

在Python中实现多进程间通信存在多种方案，包括使用队列（Queue）或管道（Pipe）等机制。

然而，上述传统方式通常要求参与通信的多个进程必须源自同一个父进程，存在明显的适用范围限制。

当多个进程彼此独立、不共享父进程时，仅能依赖共享内存或信号量等底层机制。这些方式在处理Queue、dict、list等复杂数据结构时显得繁琐且缺乏灵活性。

Manager机制原理

Manager提供了一种更高级的多进程通信抽象，它能够支持Python中几乎所有的标准数据结构，极大简化了共享数据的操作。

其核心原理是启动一个阻塞运行的ManagerServer进程，该进程监听特定的Socket端口；其他进程作为ManagerClient通过Socket连接至Server，从而建立通信通道。

代码实现与使用步骤

具体的实现逻辑封装在manager.py文件中，主要包含Server和Client两个核心类的定义。


# encoding=utf-8
__author__ = 'kevinlu1010@qq.com'
'''
进程间通信
'''
from multiprocessing.managers import BaseManager
from multiprocessing import RLock

MANAGER_PORT = 6000
MANAGER_DOMAIN = '0.0.0.0'
MANAGER_AUTH_KEY = 'aaaaaaaaaaaaaaa'

#定义一个Manager类
class InfoManager(BaseManager): pass


class DictItem():
  def __init__(self, ):
    self.items = dict()

  def set(self, key, value):
    self.items[key] = value

  def get(self, key):
    return self.items.get(key)

  def __setitem__(self, key, value):
    self.set(key, value)

#为这个manager类注册存储容器，也就是通过这个manager类实现的共享的变量，
#这个变量最好是一个类实例，自己定义的或者python自动的类的实例都可以
#这里不能把d改成dict()，因为Client那边执行d['keyi']='value'的时候会报错：d这个变量不能修改
d = DictItem()
lock = RLock()
InfoManager.register('dict', callable=lambda: d)
InfoManager.register('open_qq_login_lock', callable=lambda: lock)


class ManagerServer():
  '''
  multiprocess Manager服务类
  '''

  def __init__(self, domain, port, auth_key):
    self.domain = domain
    self.port = port
    self.auth_key = auth_key

  def start_manager_server(self):
    self.queue_manager = InfoManager(address=('', self.port), authkey=self.auth_key)
    # self.dict = self.queue_manager.dict()
    self.server = self.queue_manager.get_server()

  def run(self):
    self.start_manager_server()
    self.server.serve_forever()

  def stop(self):
    self.server.shutdown()
    self.is_stop = 1


class ManagerClient():
  '''
  访问mutiprocess Manager的类
  '''

  def __init__(self, domain, port, auth_key):
    self.domain = domain
    self.port = port
    self.auth_key = auth_key
    # self.get_share_dict()
    self.info_manager = InfoManager(address=(self.domain, self.port), authkey=self.auth_key)
    self.info_manager.connect()

  def get_dict(self):
    # self.dict = m.dict()
    self.dict = self.info_manager.dict()
    return self.dict

  def get_open_qq_login_lock(self):
    self.open_qq_login_lock = self.info_manager.open_qq_login_lock()
    return self.open_qq_login_lock


if __name__ == '__main__':
  pass

以下是该机制的具体使用步骤与注意事项。

第一步，启动ManagerServer进程。该进程处于阻塞状态，持续监听指定端口以接受客户端连接。


import manager

def run():
  manager_server = manager.ManagerServer(manager.MANAGER_DOMAIN, manager.MANAGER_PORT, manager.MANAGER_AUTH_KEY)
  manager_server.run()
if __name__ == '__main__':
  run()

第二步，在客户端实例化ManagerClient，并通过连接获取共享变量的引用。


# 进程间共享变量
manager_client = manager.ManagerClient(manager.MANAGER_DOMAIN, manager.MANAGER_PORT, manager.MANAGER_AUTH_KEY)
share_dict = manager_client.get_dict()
open_qq_login_lock = manager_client.get_open_qq_login_lock()