亲宝软件园·资讯

展开

Python多进程并发与同步机制超详细讲解

alwaysrun 人气:0

在《多线程与同步》中介绍了多线程及存在的问题,而通过使用多进程而非线程可有效地绕过全局解释器锁。 因此,通过multiprocessing模块可充分地利用多核CPU的资源。

多进程

多进程是通过multiprocessing包来实现的,multiprocessing.Process对象(和多线程的threading.Thread类似)用来创建一个进程对象:

僵尸进程

在unix或unix-like系统中,当一个子进程退出后,它就会变成一个僵尸进程,如果父进程没有通过wait系统调用来读取这个子进程的退出状态的话,这个子进程就会一直维持僵尸进程状态(占据部分系统资源,无法释放)。

要清除僵尸进程,有:

结束父进程(一般是主进程):当父进程退出的时候僵尸进程也会被随之清除。

读取子进程退出状态:如通过multiprocessing.Process产出的进程可以:

把进程变成孤儿进程,这样进程就会自动交由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]]]]]),实例化得到的对象,表示一个子进程任务:

Process类的属性与方法:

也可通过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模块中):

屏障示例:

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函数又会执行一次)。

状态管理Managers

Managers提供了一种创建由多进程(包括跨机器间进程共享)共享的数据的方式:

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

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)

加载全部内容

相关教程
猜你喜欢
用户评论