在《多線程與同步》中介紹了多線程及存在的問題，而通過使用多進(jìn)程而非線程可有效地繞過全局解釋器鎖。 因此，通過multiprocessing模塊可充分地利用多核CPU的資源。

多進(jìn)程

多進(jìn)程是通過multiprocessing包來實(shí)現(xiàn)的，multiprocessing.Process對(duì)象（和多線程的threading.Thread類似）用來創(chuàng)建一個(gè)進(jìn)程對(duì)象：

• 在類UNIX平臺(tái)上，需要對(duì)每個(gè)Process對(duì)象調(diào)用join()方法 (實(shí)際上等同于wait)避免其成為僵尸進(jìn)程。
• multiprocessing提供了threading包中沒有的IPC(比如Pipe和Queue)，效率上更高。應(yīng)優(yōu)先考慮Pipe和Queue，避免使用Lock/Event/Semaphore/Condition等同步方式。
• 多進(jìn)程應(yīng)盡量避免共享資源。必要時(shí)可以通過共享內(nèi)存和Manager的方法來共享資源。

僵尸進(jìn)程

在unix或unix-like系統(tǒng)中，當(dāng)一個(gè)子進(jìn)程退出后，它就會(huì)變成一個(gè)僵尸進(jìn)程，如果父進(jìn)程沒有通過wait系統(tǒng)調(diào)用來讀取這個(gè)子進(jìn)程的退出狀態(tài)的話，這個(gè)子進(jìn)程就會(huì)一直維持僵尸進(jìn)程狀態(tài)（占據(jù)部分系統(tǒng)資源，無法釋放）。

要清除僵尸進(jìn)程，有：

結(jié)束父進(jìn)程（一般是主進(jìn)程）：當(dāng)父進(jìn)程退出的時(shí)候僵尸進(jìn)程也會(huì)被隨之清除。

讀取子進(jìn)程退出狀態(tài)：如通過multiprocessing.Process產(chǎn)出的進(jìn)程可以：

• 調(diào)用join()來等待子進(jìn)程的方法來（內(nèi)部會(huì)wait子進(jìn)程）；
• 在父進(jìn)程中處理SIGCHLD信號(hào)：在處理程序中調(diào)用wait系統(tǒng)調(diào)用或者直接設(shè)置為SIG_IGN來清除僵尸進(jìn)程；

把進(jìn)程變成孤兒進(jìn)程，這樣進(jìn)程就會(huì)自動(dòng)交由init進(jìn)程來自動(dòng)處理。

通過設(shè)定signal.signal(signal.SIGCHLD, signal.SIG_IGN)或join進(jìn)程可避免僵尸進(jìn)程的產(chǎn)生

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]]]]])，實(shí)例化得到的對(duì)象，表示一個(gè)子進(jìn)程任務(wù)：

• group參數(shù)未使用，值始終為None；
• target表示調(diào)用對(duì)象，即子進(jìn)程要執(zhí)行的任務(wù)；
• args表示調(diào)用對(duì)象的位置參數(shù)元組，args=(1, ‘test’, [‘one’])；
• kwargs表示調(diào)用對(duì)象的字典參數(shù)，kwargs={‘name’:‘mike’,‘age’:18}；
• name為子進(jìn)程的名稱；

Process類的屬性與方法：

• start()：啟動(dòng)進(jìn)程，并執(zhí)行其run方法；
• run()：進(jìn)程啟動(dòng)時(shí)運(yùn)行的方法，繼承Process類時(shí)必須要實(shí)現(xiàn)方法；
• terminate()：強(qiáng)制終止進(jìn)程，不會(huì)進(jìn)行任何清理操作（若p創(chuàng)建了子進(jìn)程，則子進(jìn)程就成了僵尸進(jìn)程）;如進(jìn)程還持有鎖等，那么也不會(huì)被釋放，進(jìn)而導(dǎo)致死鎖；
• is_alive()：返回進(jìn)程是否在運(yùn)行狀態(tài)；
• join([timeout])：等待進(jìn)程終止；
• daemon：默認(rèn)值為False，如果設(shè)為True，代表為守護(hù)進(jìn)程（當(dāng)父進(jìn)程終止時(shí)，隨之終止；并且不能創(chuàng)建自己的新進(jìn)程），必須在start()之前設(shè)置；
• name：進(jìn)程的名稱；
• pid/ident：進(jìn)程的pid；
• exitcode：進(jìn)程在運(yùn)行時(shí)為None、如果為–N，表示被信號(hào)N結(jié)束；
• authkey：進(jìn)程的身份驗(yàn)證碼（默認(rèn)是由os.urandom()隨機(jī)生成的32字符的字符串），在涉及網(wǎng)絡(luò)連接的底層進(jìn)程間通信時(shí)提供安全性；

也可通過os.getpid()獲取進(jìn)程的PID，os.getppid()獲取父進(jìn)程的PID。

函數(shù)方式

通過Process類直接運(yùn)行函數(shù)：

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類，并實(shí)現(xiàn)run方法來啟動(dòng)進(jìn)程：

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()

同步機(jī)制

進(jìn)程間同步與線程間同步類似（只是所有對(duì)象都在multiprocessing模塊中）：

• Lock/Rlock: 通過acquire()和release()來獲取與釋放鎖；
• Event: 事件信號(hào)，通過set()和clear()來設(shè)定與清楚信號(hào)；通過wait()來等待信號(hào)；
• Condition: 條件變量；通過wait()用來等待條件，通過notify/notify_all()來通知等待此條件的進(jìn)程（等待與通知前，都需先持有鎖）；
• Semaphore: 信號(hào)量；維護(hù)一個(gè)計(jì)數(shù)器；
• Barrier: 屏障；只有等待進(jìn)程數(shù)量達(dá)到要求數(shù)量，才會(huì)同時(shí)開始執(zhí)行屏障保護(hù)后的代碼。

屏障示例：

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():  # 屏障滿足條件時(shí)，執(zhí)行一次
    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)  

當(dāng)?shù)?個(gè)進(jìn)程啟動(dòng)時(shí)，前面5個(gè)進(jìn)程會(huì)同時(shí)開始執(zhí)行（openFun函數(shù)會(huì)執(zhí)行一次）；當(dāng)?shù)?0個(gè)進(jìn)程啟動(dòng)時(shí)，后面5個(gè)進(jìn)程會(huì)同時(shí)開始執(zhí)行一次（openFun函數(shù)又會(huì)執(zhí)行一次）。

狀態(tài)管理Managers

Managers提供了一種創(chuàng)建由多進(jìn)程（包括跨機(jī)器間進(jìn)程共享）共享的數(shù)據(jù)的方式：

• multiprocessing.Manager()返回一個(gè)SyncManager對(duì)象；此對(duì)象對(duì)應(yīng)著一個(gè)管理者子進(jìn)程（manager process）以及代理（其他子進(jìn)程使用）；
• 它確保當(dāng)某一進(jìn)程修改了共享對(duì)象之后，其他進(jìn)程中的共享對(duì)象也會(huì)得到更新；
• 其支持的類型有：list、dict、Namespace、Lock、RLock、Semaphore、BoundedSemaphore、Condition、Event、Queue、Value和Array。

多進(jìn)程進(jìn)共享字典與列表（每個(gè)進(jìn)程中都能看到其他進(jìn)程修改過的內(nèi)容）

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)

到此這篇關(guān)于Python多進(jìn)程并發(fā)與同步機(jī)制超詳細(xì)講解的文章就介紹到這了,更多相關(guān)Python多進(jìn)程并發(fā)內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論