Python使用multiprocessing實(shí)現(xiàn)多進(jìn)程

更新時(shí)間：2025年07月21日 09:35:04 作者：黑鹿022

當(dāng)我們工作中涉及到處理大量數(shù)據(jù)、并行計(jì)算或并發(fā)任務(wù)時(shí),Python的multiprocessing模塊是一個(gè)強(qiáng)大而實(shí)用的工具,在本文中,我們將探索如何使用multiprocessing模塊實(shí)現(xiàn)多進(jìn)程編程,將介紹進(jìn)程池的概念和用法,需要的朋友可以參考下

一、進(jìn)程概念

進(jìn)程（Process）是資源分配的最小單位，它是操作系統(tǒng)進(jìn)行資源分配和調(diào)度運(yùn)行的基本單位，通俗理解：一個(gè)正在運(yùn)行的程序就是一個(gè)進(jìn)程。

注：一個(gè)程序運(yùn)行后至少有一個(gè)進(jìn)程

二、python多進(jìn)程

在python中使用多進(jìn)程我們需要導(dǎo)入 multiprocessing 模塊

import multiprocessing

實(shí)例演示：

我們先定義兩個(gè)函數(shù)，然后創(chuàng)建子進(jìn)程去調(diào)用它們

import time

def mission1(n):
    for i in range(1, n + 1):
        print(f'第{i}次執(zhí)行任務(wù)一。')
        time.sleep(0.2)


def mission2(n):
    for i in range(1, n + 1):
        print(f"第{i}次執(zhí)行任務(wù)二。")
        time.sleep(0.2)

每個(gè)函數(shù)每次循環(huán)都sleep 0.2秒目的是為了之后演示進(jìn)程之間是同步執(zhí)行的情況，否則程序運(yùn)行太快沒辦法體現(xiàn)。

我們先試著直接調(diào)用它們：

mission1(5)
mission2(5)
"""
結(jié)果：
第1次執(zhí)行任務(wù)一。
第2次執(zhí)行任務(wù)一。
第3次執(zhí)行任務(wù)一。
第4次執(zhí)行任務(wù)一。
第5次執(zhí)行任務(wù)一。
第1次執(zhí)行任務(wù)二。
第2次執(zhí)行任務(wù)二。
第3次執(zhí)行任務(wù)二。
第4次執(zhí)行任務(wù)二。
第5次執(zhí)行任務(wù)二。
"""

結(jié)果：先執(zhí)行了mission1 再執(zhí)行了mission2

子進(jìn)程依賴于父進(jìn)程創(chuàng)建，因而我們要?jiǎng)?chuàng)建兩個(gè)子進(jìn)程，那么必須在父進(jìn)程里面去創(chuàng)建。
python運(yùn)行文件的時(shí)候會(huì)創(chuàng)建一個(gè)主進(jìn)程也就是 MainProcess，因而我們必須在main里面去操作子進(jìn)程：

if __name__ == '__main__':
		pass

接著我們創(chuàng)建子進(jìn)程對(duì)象：

if __name__ == '__main__':
    p1 = multiprocessing.Process(target=mission1,  args=(5,))
    p2 = multiprocessing.Process(target=mission2,  kwargs={'n': 5})

multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}) ：
target—表示調(diào)用對(duì)象，即子進(jìn)程要執(zhí)行的任務(wù)（回調(diào)函數(shù)入口地址）
args—表示以元組的形式向子任務(wù)函數(shù)傳參，元組方式傳參一定要和參數(shù)的順序保持一致
kwargs—表示以字典的方式給子任務(wù)函數(shù)傳參，字典方式傳參字典中的key要和參數(shù)名保持一致
name—為子進(jìn)程的名稱

進(jìn)程運(yùn)行：

p1.start()
p2.start()

完整代碼：

import multiprocessing

import time

def mission1(n):
    for i in range(1, n + 1):
        print(f'第{i}次執(zhí)行任務(wù)一。')
        time.sleep(0.2)


def mission2(n):
    for i in range(1, n + 1):
        print(f"第{i}次執(zhí)行任務(wù)二。")
        time.sleep(0.2)


if __name__ == '__main__':
    p1 = multiprocessing.Process(target=mission1,  args=(5,)) # 注意單元素的元組格式
    p2 = multiprocessing.Process(target=mission2,  kwargs={'n': 5})

    p1.start()
    p2.start()
 
"""
結(jié)果：
第1次執(zhí)行任務(wù)一。
第1次執(zhí)行任務(wù)二。
第2次執(zhí)行任務(wù)一。
第2次執(zhí)行任務(wù)二。
第3次執(zhí)行任務(wù)一。
第3次執(zhí)行任務(wù)二。
第4次執(zhí)行任務(wù)一。
第4次執(zhí)行任務(wù)二。
第5次執(zhí)行任務(wù)一。
第5次執(zhí)行任務(wù)二。
"""

結(jié)果顯示：兩個(gè)函數(shù)在兩個(gè)子進(jìn)程中同時(shí)執(zhí)行。（每次執(zhí)行順序可能有略微差別）

獲取進(jìn)程信息

current_process()

獲取當(dāng)前進(jìn)程的信息：name(進(jìn)程名稱) 和 pid (進(jìn)程ID)

def mission1(user, n):
    print(f"mission-1子進(jìn)程：{multiprocessing.current_process().name}")
    print(f"mission-1子進(jìn)程ID：{multiprocessing.current_process().pid}")
    for i in range(1, n + 1):
        print(f"用戶{user}正在第{i}次執(zhí)行任務(wù)一。")
        time.sleep(0.2)

OS 模塊

可以使用os模塊中的getpid()獲取當(dāng)前進(jìn)程的ID 以及 getppid()獲取當(dāng)前進(jìn)程的父進(jìn)程ID。

import os
def mission2(user, n):
    print(f"mission-2子進(jìn)程：{multiprocessing.current_process().name}")
    print(f"mission-2子進(jìn)程ID：{os.getpid()}")
    print(f"mission-2子進(jìn)程的父進(jìn)程為：{os.getppid()}")
    for i in range(1, n + 1):
        print(f"用戶{user}正在第{i}次執(zhí)行任務(wù)二。")
        time.sleep(0.2)

完整代碼：

import multiprocessing
import time
import os


def mission1(user, n):
    print(f"mission-1子進(jìn)程：{multiprocessing.current_process().name}")
    print(f"mission-1子進(jìn)程ID：{multiprocessing.current_process().pid}")
    print(f"mission-1子進(jìn)程的父進(jìn)程為：{os.getppid()}")
    for i in range(1, n + 1):
        print(f"用戶{user}正在第{i}次執(zhí)行任務(wù)一。")
        time.sleep(0.2)


def mission2(user, n):
    print(f"mission-2子進(jìn)程：{multiprocessing.current_process().name}")
    print(f"mission-2子進(jìn)程ID：{os.getpid()}")
    print(f"mission-2子進(jìn)程的父進(jìn)程為：{os.getppid()}")
    for i in range(1, n + 1):
        print(f"用戶{user}正在第{i}次執(zhí)行任務(wù)二。")
        time.sleep(0.2)

if __name__ == '__main__':
    print(f"主進(jìn)程的ID為：{multiprocessing.current_process().pid}")
    print(f"主進(jìn)程的父進(jìn)程ID為：{os.getppid()}")
    print(f'--------主進(jìn)程{multiprocessing.current_process().name}開始執(zhí)行--------')

    p1=multiprocessing.Process(target=mission1,name='mission-1',args=('張三',5))
    p2=multiprocessing.Process(target=mission2,name='mission-2',args=('李四',5))

    p1.start()
    p2.start()
"""
主進(jìn)程的ID為：7500
主進(jìn)程的父進(jìn)程ID為：39284
--------主進(jìn)程MainProcess開始執(zhí)行--------
mission-2子進(jìn)程：mission-2
mission-2子進(jìn)程ID：40552
mission-2子進(jìn)程的父進(jìn)程為：7500
用戶李四正在第1次執(zhí)行任務(wù)二。
mission-1子進(jìn)程：mission-1
mission-1子進(jìn)程ID：46296
mission-1子進(jìn)程的父進(jìn)程為：7500
用戶張三正在第1次執(zhí)行任務(wù)一。
用戶李四正在第2次執(zhí)行任務(wù)二。
用戶張三正在第2次執(zhí)行任務(wù)一。
用戶李四正在第3次執(zhí)行任務(wù)二。
用戶張三正在第3次執(zhí)行任務(wù)一。
用戶李四正在第4次執(zhí)行任務(wù)二。
用戶張三正在第4次執(zhí)行任務(wù)一。
用戶李四正在第5次執(zhí)行任務(wù)二。
用戶張三正在第5次執(zhí)行任務(wù)一。

"""

兩個(gè)子進(jìn)程的父ID是相同的就是MainProcess這個(gè)主進(jìn)程的ID

三、進(jìn)程特點(diǎn)

多進(jìn)程之間不會(huì)共享全局變量
主進(jìn)程會(huì)默認(rèn)等待子進(jìn)程結(jié)束而結(jié)束但可以通過設(shè)置來讓子進(jìn)程跟著主進(jìn)程結(jié)束而結(jié)束

正常情況：

import multiprocessing
import time

def mission1(user, n):
    for i in range(1, n + 1):
        print(f"用戶{user}正在第{i}次執(zhí)行任務(wù)一。")
        time.sleep(0.1)

if __name__ == '__main__':
    print(f'--------主進(jìn)程{multiprocessing.current_process().name}開始執(zhí)行--------')
    p1 = multiprocessing.Process(target=mission1, name='mission-1', args=('張三', 5))
    p1.start()
    time.sleep(0.4) # 讓主進(jìn)程多執(zhí)行一會(huì)兒
    print(f'----------主進(jìn)程:{multiprocessing.current_process().name}最后一行代碼執(zhí)行完畢---------')

"""
--------主進(jìn)程MainProcess開始執(zhí)行--------
用戶張三正在第1次執(zhí)行任務(wù)一。
用戶張三正在第2次執(zhí)行任務(wù)一。
----------主進(jìn)程:MainProcess最后一行代碼執(zhí)行完畢---------
用戶張三正在第3次執(zhí)行任務(wù)一。
用戶張三正在第4次執(zhí)行任務(wù)一。
用戶張三正在第5次執(zhí)行任務(wù)一。
"""

正常情況下，主進(jìn)程（最后一行代碼執(zhí)行完了）仍然會(huì)等待子進(jìn)程(p1 mission-1)執(zhí)行完畢.

使用守護(hù)進(jìn)程或者終止進(jìn)程

import multiprocessing
import time


def mission1(user, n):
    for i in range(1, n + 1):
        print(f"用戶{user}正在第{i}次執(zhí)行任務(wù)一。")
        time.sleep(0.1)


if __name__ == '__main__':
    print(f'--------主進(jìn)程{multiprocessing.current_process().name}開始執(zhí)行--------')

    p1 = multiprocessing.Process(target=mission1, name='mission-1', args=('張三', 5))
    # 方式1: 開啟子進(jìn)程前提前設(shè)置守護(hù)主進(jìn)程
    p1.daemon = True
    p1.start()
    time.sleep(0.4)  # 讓主進(jìn)程多執(zhí)行一會(huì)兒
    # p1.terminate()
    print(f'----------主進(jìn)程:{multiprocessing.current_process().name}最后一行代碼執(zhí)行完畢---------')

"""
--------主進(jìn)程MainProcess開始執(zhí)行--------
用戶張三正在第1次執(zhí)行任務(wù)一。
用戶張三正在第2次執(zhí)行任務(wù)一。
----------主進(jìn)程:MainProcess最后一行代碼執(zhí)行完畢---------
"""

我們發(fā)現(xiàn)子進(jìn)程還沒執(zhí)行結(jié)束就被終止了。