Python 多进程管理器便捷的进程间数据共享multiprocessing.Manager 提供了一种高级的进程间数据共享方式。它通过代理对象在多个进程之间同步数据无需手动处理锁和共享内存。本文详解 Manager 的完整用法。一、Manager 基础创建和使用import multiprocessing as mpimport timefrom multiprocessing.managers import (BaseManager, BaseProxy, SyncManager)def manager_basics():Manager 的基本用法# 创建 Manager 对象启动管理器服务器进程manager mp.Manager()# 创建各种共享数据结构shared_list manager.list([1, 2, 3])shared_dict manager.dict({key: value})shared_set manager.set([1, 2, 3])shared_namespace manager.Namespace()# Namespace 可以通过属性访问shared_namespace.counter 0shared_namespace.config {host: localhost, port: 8080}# 在子进程中修改共享数据def worker(lst, dct, ns):子进程修改 Manager 管理的共享对象lst.append(4)lst.append(5)dct[new_key] new_valuens.counter 1print(f子进程: list{list(lst)}, dict{dict(dct)}, counter{ns.counter})processes []for i in range(3):p mp.Process(targetworker,args(shared_list, shared_dict, shared_namespace))p.start()processes.append(p)for p in processes:p.join()print(f主进程: list{list(shared_list)})print(f主进程: dict{dict(shared_dict)})print(f主进程: counter{shared_namespace.counter})# 使用完毕后必须 shutdownmanager.shutdown()二、Manager 提供的核心数据类型def manager_data_types():Manager 支持的所有数据类型manager mp.Manager()# 1. list: 类似 list 的代理lst manager.list([1, 2, 3])lst.append(4)lst.extend([5, 6])print(flist: {list(lst)}, 长度{len(lst)})# 2. dict: 类似 dict 的代理dct manager.dict({a: 1})dct[b] 2dct.update({c: 3, d: 4})print(fdict: {dict(dct)})# 3. set: 类似 set 的代理Python 3.6s manager.set([1, 2, 3, 3, 2]) # 去重s.add(4)s.discard(1)print(fset: {set(s)})# 4. Namespace: 属性访问的代理ns manager.Namespace()ns.x 10ns.y 20# Namespace 支持嵌套ns.sub manager.Namespace()ns.sub.value nestedprint(fNamespace: x{ns.x}, sub.value{ns.sub.value})# 5. Queue: 进程安全队列q manager.Queue()q.put(task-1)q.put(task-2)print(fQueue: get{q.get()})# 6. Lock / RLock / Semaphore / Event / Conditionlock manager.Lock()with lock:print(获取了 Manager 提供的锁)manager.shutdown()三、自定义 Manager 类型class MyCustomManager(BaseManager):自定义 Manager注册自定义类型。适用于需要共享自定义类的场景。class Counter:一个简单的计数器类def __init__(self, initial: int 0):self._count initialdef increment(self, amount: int 1):self._count amountdef get_count(self) - int:return self._countdef __repr__(self) - str:return fCounter({self._count})# 注册自定义类型到 ManagerMyCustomManager.register(Counter, Counter)def custom_manager_demo():使用自定义 Manager 类型manager MyCustomManager()manager.start() # 启动管理器服务器# 创建共享的 Counter 实例shared_counter manager.Counter(10)def worker(counter):子进程操作共享计数器for _ in range(100):counter.increment()processes [mp.Process(targetworker, args(shared_counter,))for _ in range(4)]for p in processes: p.start()for p in processes: p.join()print(f最终计数: {shared_counter.get_count()} (期望: 410))manager.shutdown()四、代理对象的工作机制def proxy_object_demo():代理对象是 Manager 的核心机制。当通过 Manager 创建对象时实际上是在 Manager 服务器进程中创建了一个真实对象而客户端进程获得的是代理对象。对代理对象的操作会被序列化并通过管道发送到服务器进程执行结果再反序列化返回。manager mp.Manager()shared_dict manager.dict({count: 0})def examine_proxy(proxy):检查代理对象的内部机制# 代理对象类型print(f代理类型: {type(proxy)})print(f代理 ID: {id(proxy)})# 代理对象的自动同步proxy[count] 1# 上面的操作等价于伪代码# 1. 序列化操作(__setitem__, (count, 1))# 2. 通过管道发送到 Manager 服务器# 3. 服务器执行真实字典的 __setitem__# 4. 返回确认examine_proxy(shared_dict)# 代理对象的局限性# 1. 某些原地操作不会自动同步如 list.sort()# 2. 嵌套修改需要重新赋值# 3. 属性访问较慢每次都有 IPC 开销manager.shutdown()五、Manager vs 共享内存性能对比def manager_vs_shared_memory():Manager 使用 IPC进程间通信每次操作都有序列化开销。共享内存使用零拷贝在大数据量场景下性能更优。选择依据- 小数据量、频繁读写Manager 更方便- 大数据量、批量操作共享内存更高效- 需要复杂数据结构Manager 更易用- 需要极致性能共享内存import arrayfrom multiprocessing import shared_memoryDATA_SIZE 100_000ITERATIONS 100# Manager 性能测试 manager mp.Manager()shared_list manager.list(range(DATA_SIZE))def manager_worker(lst):for i in range(len(lst)):lst[i] lst[i] * 2start time.perf_counter()processes [mp.Process(targetmanager_worker, args(shared_list,))for _ in range(4)]for p in processes: p.start()for p in processes: p.join()manager_time time.perf_counter() - startmanager.shutdown()# 共享内存性能测试 data array.array(i, range(DATA_SIZE))shm shared_memory.SharedMemory(createTrue, sizelen(data) * data.itemsize)shm.buf[:len(data) * data.itemsize] bytes(data)def shm_worker(shm_name, size):ex_shm shared_memory.SharedMemory(nameshm_name)arr array.array(i)arr.frombytes(ex_shm.buf[:size])for i in range(len(arr)):arr[i] * 2ex_shm.buf[:size] bytes(arr)ex_shm.close()start time.perf_counter()processes [mp.Process(targetshm_worker, args(shm.name, len(data) * data.itemsize))for _ in range(4)]for p in processes: p.start()for p in processes: p.join()shm_time time.perf_counter() - startshm.close()shm.unlink()print(f Manager vs 共享内存性能{DATA_SIZE} 元素 )print(fManager 操作: {manager_time:.3f}s)print(f共享内存: {shm_time:.3f}s)print(fManager 比共享内存慢 {manager_time / shm_time:.1f}x)六、分布式多进程Manager 支持网络访问。可以在一台机器上启动 Manager 服务器其他机器通过网络连接进行远程进程间通信。适用场景1. 跨机器的分布式任务队列2. 共享配置中心3. 分布式锁def remote_manager_server():启动可远程访问的 Manager 服务器manager mp.Manager()# 获取 Manager 服务器地址# 默认只监听本地可以配置为监听网络接口print(fManager 地址: {manager._address})print(fManager 进程 PID: {manager._process.pid})# 实际使用时需要通过 multiprocessing.managers 自定义return managerdef custom_remote_manager():创建可远程连接的自定义 Managerclass RemoteManager(BaseManager):pass# 注册需要共享的类型RemoteManager.register(get_dict, callablelambda: {})# 启动服务器监听指定地址manager RemoteManager(address(0.0.0.0, 50000), authkeybsecret)manager.start()print(f远程 Manager 服务器启动在 {manager._address})# 客户端连接其他机器上# client RemoteManager(address(server_ip, 50000), authkeybsecret)# client.connect()# shared_data client.get_dict()return manager七、总结# 1. Manager 提供了 list/dict/set/Namespace/Queue/Lock 等共享类型# 2. 内部使用代理模式操作被序列化后发送到 Manager 服务器执行# 3. 自定义 Manager 可以注册自定义类来共享# 4. 代理对象方便易用但每次操作都有 IPC 开销# 5. 大数据量场景共享内存性能更优# 6. Manager 支持网络分布式访问# 7. 使用后必须调用 shutdown() 清理资源# 8. 适合需要复杂数据结构的进程间通信if __name__ __main__:manager_basics()
