Python多进程并发与同步机制详解：Process、同步原语与Manager

前文《多线程与同步》探讨了多线程及其局限性，而采用多进程方式可有效规避全局解释器锁（GIL）的限制。借助multiprocessing模块，程序能够充分利用多核CPU的计算资源。

多进程基础

在Python中，多进程功能由multiprocessing包提供。其中，multiprocessing.Process对象（与threading.Thread类似）用于创建一个新的进程实例。

• 在类UNIX系统平台上，必须对每个Process对象调用join()方法（其作用等同于wait），以防止子进程在父进程退出后变为僵尸进程。
• multiprocessing提供了threading模块所不具备的进程间通信（IPC）机制，如Pipe和Queue，且效率更高。建议优先使用Pipe和Queue进行通信，尽量避免使用Lock、Event、Semaphore或Condition等同步原语。
• 多进程编程中应尽量避免共享资源。若确需共享，可通过共享内存（Shared Memory）或Manager对象来实现资源共享。

僵尸进程处理

在Unix或类Unix系统中，当子进程结束时，若父进程未通过wait系统调用读取其退出状态，该子进程将保持僵尸状态，持续占用部分系统资源且无法释放。

清除僵尸进程主要有以下几种方法：

终止父进程：当父进程（通常为主进程）退出时，其所有的僵尸子进程也会被操作系统自动清理。

读取子进程退出状态：对于由multiprocessing.Process创建的进程，可通过以下方式处理：

• 调用join()方法等待子进程结束（该方法内部会自动执行wait系统调用）；
• 在父进程中处理SIGCHLD信号：在信号处理程序中调用wait系统调用，或者将SIGCHLD信号设置为SIG_IGN以忽略该信号，从而自动清理僵尸进程；

使进程成为孤儿进程：若父进程先于子进程退出，子进程将被init进程收养，由init自动处理其状态。

通过设定signal.signal(signal.SIGCHLD, signal.SIG_IGN)或调用join()方法，可以有效避免僵尸进程的产生。

def zombieProc():
    print("zombie running")
    time.sleep(5)
    print("zombie exit")
if __name__ == '__main__':
    signal.signal(signal.SIGCHLD, signal.SIG_IGN)
    proc = multiprocessing.Process(target=zombieProc)
    proc.start()
    # proc.join()
    time.sleep(30)

Process类详解

Process([group [, target [, name [, args [, kwargs]]]]])，实例化后得到的对象代表一个子进程任务：

• group参数目前未被使用，值始终为None；
• target表示调用的可调用对象，即子进程需要执行的具体任务；
• args表示传递给调用对象的位置参数元组，例如args=(1, 'test', ['one'])；
• kwargs表示传递给调用对象的字典参数，例如kwargs={'name':'mike','age':18}；
• name用于指定子进程的名称；

Process类包含以下主要属性与方法：

• start()：启动进程，并自动执行其run方法；
• run()：进程启动时执行的方法，若继承Process类，则必须实现此方法；
• terminate()：强制终止进程，不执行任何清理操作（若进程创建了子进程，子进程将变为僵尸进程）；如果进程持有锁等资源，这些资源也不会被释放，可能导致死锁；
• is_alive()：返回布尔值，指示进程是否仍在运行；
• join([timeout])：阻塞父进程，直到子进程终止；
• daemon：默认值为False。若设为True，表示该进程为守护进程（父进程终止时它也会终止，且不能创建新子进程），必须在start()之前设置；
• name：进程的名称；
• pid/ident：进程的ID；
• exitcode：进程运行时为None；若为-N，表示进程被信号N终止；
• authkey：进程的身份验证密钥（默认由os.urandom()生成的32字符字符串），在涉及网络连接的底层进程间通信中提供安全性保障；

也可通过os.getpid()获取当前进程的PID，通过os.getppid()获取父进程的PID。

函数方式创建进程

通过Process类直接运行函数示例如下：

def simpleRoutine(name, delay):
    print(f"routine {name} starting...")
    time.sleep(delay)
    print(f"routine {name} finished")
if __name__ == '__main__':
    thrOne = multiprocessing.Process(target=simpleRoutine, args=("First", 1))
    thrTwo = multiprocessing.Process(target=simpleRoutine, args=("Two", 2))
    thrOne.start()
    thrTwo.start()
    thrOne.join()
    thrTwo.join()

继承方式创建进程

通过继承Process类并实现run方法来启动进程：

class SimpleProcess(multiprocessing.Process):
    def __init__(self, name, delay):
        super().__init__()
        self.name = name
        self.delay = delay
    def run(self):
        print(f"Process {self.name} starting...")
        time.sleep(self.delay)
        print(f"Process {self.name} finished")
if __name__ == '__main__':
    thrOne = SimpleProcess("First", 2)
    thrTwo = SimpleProcess("Second", 1)
    thrOne.start()
    thrTwo.start()
    thrOne.join()
    thrTwo.join()

同步机制

进程间的同步机制与线程间类似（所有对象均位于multiprocessing模块中）：

• Lock/Rlock：通过acquire()和release()方法获取与释放锁；
• Event：事件信号，通过set()和clear()方法设定与清除信号；通过wait()方法等待信号；
• Condition：条件变量；通过wait()等待条件，通过notify/notify_all()通知等待该条件的进程（等待与通知前需先持有锁）；
• Semaphore：信号量；维护一个计数器以限制并发访问数量；
• Barrier：屏障；只有当等待进程数量达到指定数量时，才会同时执行屏障保护后的代码。

屏障使用示例：

def waitBarrier(name, barr: multiprocessing.Barrier):
    print(f"{name} waiting for open")
    try:
        barr.wait()
        print(f"{name} running")
        time.sleep(2)
    except multiprocessing.BrokenBarrierError:
        print(f"{name} exception")
    print(f"{name} finished")
def openFun():  # 屏障满足条件时，执行一次
    print("barrier opened")
if __name__ == '__main__':
    signal = multiprocessing.Barrier(5, openFun)
    for i in range(10):
        multiprocessing.Process(target=waitBarrier, args=(i, signal)).start()  
        time.sleep(1)  

当第5个进程启动时，前5个进程会同时继续执行（openFun函数执行一次）；当第10个进程启动时，后5个进程会同时继续执行一次（openFun函数再次执行）。

状态管理Manager

Managers提供了一种在多进程（包括跨机器进程共享）中创建共享数据的方式：

• multiprocessing.Manager()返回一个SyncManager对象；该对象对应一个管理者子进程（manager process）以及代理对象（其他子进程通过代理访问）；
• 它确保当一个进程修改了共享对象后，其他进程中的共享对象也能同步更新；
• 其支持的类型包括：list、dict、Namespace、Lock、RLock、Semaphore、BoundedSemaphore、Condition、Event、Queue、Value和Array。

多进程共享字典与列表（每个进程中都能看到其他进程修改后的内容）

def worker(dictContext: dict, lstContext: list, name):
    pid = os.getpid()
    dictContext[name] = pid
    lstContext.append(pid)
    print(f"{name} worker: {lstContext}")
def managerContext():
    mgr = multiprocessing.Manager()
    multiprocessing.managers
    dictContext = mgr.dict()
    lstContext = mgr.list()
    jobs = [multiprocessing.Process(target=worker, args=(dictContext, lstContext, i)) for i in range(10)]
    for j in jobs:
        j.start()
    for j in jobs:
        j.join()
    print('Results:', dictContext)

本文关于Python多进程并发与同步机制的详细讲解到此结束，更多相关Python多进程内容请搜索本站以前的文章或继续浏览下面的相关文章，希望大家以后多多支持本站！