Python跨进程共享数据/对象

2023-05-16

1. 跨进程共享方式

在multiprocess库中，跨进程对象共享有三种方式：

（1）第一种仅适用于原生机器类型，即python.ctypes当中的类型，这种在mp库的文档当中称为shared memory方式，即通过共享内存共享对象

（2）另外一种称之为server process，即有一个服务器进程负责维护所有的对象，而其他进程连接到该进程，通过代理对象操作服务器进程当中的对象；

（3）最后一种在mp文档当中没有单独提出，但是在其中多次提到，而且是mp库当中最重要的一种共享方式，称为inheritance，即继承，对象在父进程当中创建，然后在父进程是通过multiprocessing.Process创建子进程之后，子进程自动继承了父进程当中的对象，并且子进程对这些对象的操作都是反映到了同一个对象。

2. Shared Memory模型

shared memory模型能共享ctypes当中的类型，通过RawValue，RawArray等包装类提供。通过查看multiprocess的源码可以看到支持的类型有：

Value、Array、Lock等，

def Event() -> synchronize.Event: ...
def Lock() -> synchronize.Lock: ...
def RLock() -> synchronize.RLock: ...
def Semaphore(value: int = ...) -> synchronize.Semaphore: ...
def Pipe(duplex: bool = ...) -> Tuple[connection.Connection, connection.Connection]: ...
def Pool(
    processes: Optional[int] = ...,
    initializer: Optional[Callable[..., Any]] = ...,
    initargs: Iterable[Any] = ...,
    maxtasksperchild: Optional[int] = ...,
) -> pool.Pool: ...

# Functions Array and Value are copied from context.pyi.
# See https://github.com/python/typeshed/blob/ac234f25927634e06d9c96df98d72d54dd80dfc4/stdlib/2and3/turtle.pyi#L284-L291
# for rationale
def Array(typecode_or_type: Any, size_or_initializer: Union[int, Sequence[Any]], *, lock: bool = ...) -> sharedctypes._Array: ...
def Value(typecode_or_type: Any, *args: Any, lock: bool = ...) -> sharedctypes._Value: ...

共享的使用方法示例如下：

import multiprocessing


def func(num):
    num.value=11.11  # 子进程改变数值的值，主进程跟着改变


if __name__=="__main__":
    # d表示数值,主进程与子进程共享这个value。(主进程与子进程都是用的同一个value)
    num=multiprocessing.Value("d", 10.0) 
    p=multiprocessing.Process(target=func,args=(num,))
    p.start()
    p.join()

    print(num.value)

3. Server Process模型

这个模式支持跨进程共享所有对象，也即是想要共享 “自定义对象”，只能使用这个方式！

在server process模型中，有一个manager进程（就是那个server进程），负责管理实际的对象，真正的对象也是在manager进程的内存空间中。所有需要访问该对象的进程都需要先连接到该管理进程，然后获取到对象的一个代理对象(Proxy object)。这个模型是一个典型的RPC(远程过程调用)的模型。因为每个客户进程实际上都是在访问manager进程当中的对象，因此完全可以通过这个实现对象共享。

（1）Manager支持类型

通过查看源码可以发现，Manage() 支持的类型有：

def BoundedSemaphore(self, value: Any = ...) -> threading.BoundedSemaphore: ...
def Condition(self, lock: Any = ...) -> threading.Condition: ...
def Event(self) -> threading.Event: ...
def Lock(self) -> threading.Lock: ...
def Namespace(self) -> _Namespace: ...
def Queue(self, maxsize: int = ...) -> queue.Queue[Any]: ...
def RLock(self) -> threading.RLock: ...
def Semaphore(self, value: Any = ...) -> threading.Semaphore: ...
def Array(self, typecode: Any, sequence: Sequence[_T]) -> Sequence[_T]: ...
def Value(self, typecode: Any, value: _T) -> ValueProxy[_T]: ...
def dict(self, sequence: Mapping[_KT, _VT] = ...) -> Dict[_KT, _VT]: ...
def list(self, sequence

使用示例如下：

import multiprocessing


def func(dict_in,list_in):
    # 跨进程共享, 子进程修改，主进程跟着改变
    dict_in["index1"]="xxx"

    list_in.append("xx")
    list_in.append("yy")


if __name__=="__main__":
with multiprocessing.Manager() as mg:
    # 创建主进程与子进程之间共享的dict/list
    mydict=multiprocessing.Manager().dict()
    mylist=multiprocessing.Manager().list(range(5))
    p=multiprocessing.Process(target=func,args=(mydict,mylist))
    p.start()
    p.join()
    
    print(mylist)
    print(mydict)

（2）共享自定义类

很多场景下，Manager自带的类并不能满足我们的需求，这时候就需要用到Manager对自定义类的支持。Server Process模型共享自定义对象的实现流程如下：

(1) 基于multiprocessing.managers 重写MyManager，类内部啥都不用实现：

class MyManager(managers.BaseManager):
    """
    自定义Manager
    """
    # Pass is really enough. Nothing needs to be done here.
    pass

(2) 注册自定义类，如：RedisService、MySQLService，

# RedisService/MySQLService是自定义类, 类内部分别包含Redis连接和MySQL连接，类定义此处省略
MyManager.register("RedisService", RedisService)
 MyManager.register("MySQLService", MySQLService)

(3) 构造MyManager的实例，并由它创建多进程共享的自定义对象，

manager = MyManager()
manager.start()

# 创建共享对象
self.redis_service = manager.RedisService(settings)
self.mysql_service = manager.MySQLService(settings)

(4) 该对象以参数形式传入到子进程中，子进程直接使用。

全流程的代码示例如下：

from concurrent.futures import ProcessPoolExecutor
from multiprocessing import managers


class MyManager(managers.BaseManager):
    """
    自定义Manager
    """
    # Pass is really enough. Nothing needs to be done here.
    pass

def proc_worker(redis_service, mysql_service, task_id):
    """
    TODO::工作进程
    """
    
    # 此处可直接使用进程池共享的redis和mongo服务
    rst = redis_service.get(task_id)
    rst = mysql_service.save_result(rst)
    ...

class ServerExecutor:
    """
    调度执行器
    """
    def __init__(self, settings):
        # 配置信息, 用于创建redis链接和mysql链接
        self.settings = settings

        # 在Manager中注册自定义类（RedisService/MySQLService是我的自定义类, 类内部分别包含Redis连接和MySQL连接）
        MyManager.register("RedisService", RedisService)
        MyManager.register("MySQLService", MySQLService)
        manager = MyManager()
        manager.start()

        # 创建共享对象
        self.redis_service = manager.RedisService(settings)
        self.mysql_service = manager.MySQLService(settings)

        # 这里不仅可以是ProcessPoolExecutor，也可以是多进程Process或者进程池Pool，各自用法略有不同
        self.executor = ProcessPoolExecutor(settings.executor_num)

    def submit(self, task_id):
        """
        提交任务
        """
        future = self.executor.submit(proc_work,
                                      self.redis_service,
                                      self.mysql_service,
                                      task_id)
        return future


# demo
...
executor = ServerExecutor(settings)
future = executor.submit(task_id)
future.add_done_callback(task_done_callback)
...

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

python

跨进程共享数据