Python的進(jìn)程及進(jìn)程池詳解

更新時(shí)間：2021年11月25日 15:04:29 作者：程序員-夏天

這篇文章主要為大家介紹了Python的進(jìn)程及進(jìn)程池，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助

進(jìn)程

進(jìn)程是操作系統(tǒng)分配資源的基本單元，是程序隔離的邊界。

進(jìn)程和程序

程序只是一組指令的集合，它本身沒有任何運(yùn)行的含義，它是靜態(tài)的。

進(jìn)程程序的執(zhí)行實(shí)例，是動(dòng)態(tài)的，有自己的生命周期，有創(chuàng)建有撤銷，存在是暫時(shí)的。

進(jìn)程和程序不是一一對(duì)應(yīng)的，一個(gè)程序可以對(duì)應(yīng)多個(gè)進(jìn)程，一個(gè)進(jìn)程也可以執(zhí)行一個(gè)或者多個(gè)程序。

我們可以這樣理解：編寫完的代碼，沒有運(yùn)行時(shí)稱為程序，正在運(yùn)行的代碼，會(huì)啟動(dòng)一個(gè)（或多個(gè)）進(jìn)程。

進(jìn)程的狀態(tài)

在我們的操作系統(tǒng)?作時(shí)，任務(wù)數(shù)往往?于cpu核心數(shù)，即?定有?些任務(wù)正在執(zhí)?，?另外?些任務(wù)在等待cpu，因此導(dǎo)致了進(jìn)程有不同的狀態(tài)。

就緒狀態(tài)：已滿?運(yùn)?條件，等待cpu執(zhí)?
執(zhí)?狀態(tài)：cpu正在執(zhí)?
等待狀態(tài)：等待某些條件滿?，比如?個(gè)程序sleep了，此時(shí)就處于等待狀態(tài)

Python中的進(jìn)程

在Python中，進(jìn)程是通過multiprocessing多進(jìn)程模塊來創(chuàng)建的，multiprocessing模塊提供了?個(gè)Process類來創(chuàng)建進(jìn)程對(duì)象。

創(chuàng)建?進(jìn)程

Process語(yǔ)法結(jié)構(gòu)：

Process(group, target, name, args, kwargs)

group：指定進(jìn)程組，?多數(shù)情況下?不到
target：表示調(diào)用對(duì)象，即子進(jìn)程要執(zhí)行的任務(wù)
name：子進(jìn)程的名稱，可以不設(shè)定
args：給target指定的函數(shù)傳遞的參數(shù)，以元組的?式傳遞
kwargs：給target指定的函數(shù)傳遞命名參數(shù)

Process常用方法

p.start() 啟動(dòng)進(jìn)程，并調(diào)用該子進(jìn)程中的p.run()方法
p.join(timeout):主進(jìn)程等待?進(jìn)程執(zhí)?結(jié)束再結(jié)束，timeout是可選的超時(shí)時(shí)間
is_alive()：判斷進(jìn)程?進(jìn)程是否還存活
p.run() 進(jìn)程啟動(dòng)時(shí)運(yùn)行的方法，正是它去調(diào)用target指定的函數(shù)
p.terminate() ?即終??進(jìn)程

Process創(chuàng)建的實(shí)例對(duì)象的常?屬性

name：當(dāng)前進(jìn)程的別名，默認(rèn)為Process-N，N為從1開始遞增的整數(shù)

pid：當(dāng)前進(jìn)程的pid（進(jìn)程號(hào)）

import multiprocessing
import os
import time
def work(name):
    print("子進(jìn)程work正在運(yùn)行......")
    time.sleep(0.5)
    print(name)
    # 獲取進(jìn)程的名稱
    print("子進(jìn)程name", multiprocessing.current_process())
    # 獲取進(jìn)程的pid
    print("子進(jìn)程pid", multiprocessing.current_process().pid, os.getpid())
    # 獲取父進(jìn)程的pid
    print("父進(jìn)程pid", os.getppid())
    print("子進(jìn)程運(yùn)行結(jié)束......")
if __name__ == '__main__':
    print("主進(jìn)程啟動(dòng)")
    # 獲取進(jìn)程的名稱
    print("主進(jìn)程name", multiprocessing.current_process())
    # 獲取進(jìn)程的pid
    print("主進(jìn)程pid", multiprocessing.current_process().pid, os.getpid())
    # 創(chuàng)建進(jìn)程
    p = multiprocessing.Process(group=None, target=work, args=("tigeriaf", ))
    # 啟動(dòng)進(jìn)程
    p.start()
    print("主進(jìn)程結(jié)束")

通過上述代碼我們發(fā)現(xiàn)，multiprocessing.Process幫我們創(chuàng)建一個(gè)子進(jìn)程，并且成功運(yùn)行，但是我們發(fā)現(xiàn)，在子進(jìn)程還沒執(zhí)行完的時(shí)候主進(jìn)程就已經(jīng)死了，那么這個(gè)子進(jìn)程在主進(jìn)程結(jié)束后就是一個(gè)孤兒進(jìn)程，那么我們可以讓主進(jìn)程等待子進(jìn)程結(jié)束后再結(jié)束嗎？答案是可以的。那就是通過p.join()，join()的作用是讓主進(jìn)程等子進(jìn)程執(zhí)行完再退出。

import multiprocessing
import os
import time
def work(name):
    print("子進(jìn)程work正在運(yùn)行......")
    time.sleep(0.5)
    print(name)
    # 獲取進(jìn)程的名稱
    print("子進(jìn)程name", multiprocessing.current_process())
    # 獲取進(jìn)程的pid
    print("子進(jìn)程pid", multiprocessing.current_process().pid, os.getpid())
    # 獲取父進(jìn)程的pid
    print("父進(jìn)程pid", os.getppid())
    print("子進(jìn)程運(yùn)行結(jié)束......")
if __name__ == '__main__':
    print("主進(jìn)程啟動(dòng)")
    # 獲取進(jìn)程的名稱
    print("主進(jìn)程name", multiprocessing.current_process())
    # 獲取進(jìn)程的pid
    print("主進(jìn)程pid", multiprocessing.current_process().pid, os.getpid())
    # 創(chuàng)建進(jìn)程
    p = multiprocessing.Process(group=None, target=work, args=("tigeriaf", ))
    # 啟動(dòng)進(jìn)程
    p.start()
    p.join()
    print("主進(jìn)程結(jié)束")

運(yùn)行結(jié)果：

可以看出，主進(jìn)程是在子進(jìn)程結(jié)束后才結(jié)束的。

全局變量問題

全局變量在多個(gè)進(jìn)程中不共享，進(jìn)程之間的數(shù)據(jù)是獨(dú)立的，默認(rèn)情況下互不影響。

import multiprocessing
# 定義全局變量
num = 99
def work1():
    print("work1正在運(yùn)行......")
    global num   # 在函數(shù)內(nèi)部聲明使?全局變量num
    num = num + 1  # 對(duì)num值進(jìn)?+1
    print("work1 num = {}".format(num))
def work2():
    print("work2正在運(yùn)行......")
    print("work2 num = {}".format(num))
if __name__ == '__main__':
    # 創(chuàng)建進(jìn)程p1
    p1 = multiprocessing.Process(group=None, target=work1)
    # 啟動(dòng)進(jìn)程p1
    p1.start()
    # 創(chuàng)建進(jìn)程p2
    p2 = multiprocessing.Process(group=None, target=work2)
    # 啟動(dòng)進(jìn)程p2
    p2.start()

運(yùn)行結(jié)果：

從運(yùn)?結(jié)果可以看出，work1()函數(shù)對(duì)全局變量num的修改，在work2中并沒有獲取到，?還是原來的99，所以，進(jìn)程之間是不夠共享變量的。

守護(hù)進(jìn)程

上面說到，可以使用p.join()讓主進(jìn)程等待子進(jìn)程結(jié)束后再結(jié)束，那么可不可以讓子進(jìn)程在主進(jìn)程結(jié)束的時(shí)候就結(jié)束呢？答案是肯定的。我們可以使用p.daemon = True或者p2.terminate()進(jìn)行設(shè)置：

import multiprocessing
import time
def work1():
    print("work1正在運(yùn)行......")
    time.sleep(4)
    print("work1運(yùn)行完畢")
def work2():
    print("work2正在運(yùn)行......")
    time.sleep(10)
    print("work2運(yùn)行完畢")
if __name__ == '__main__':
    # 創(chuàng)建進(jìn)程p1
    p1 = multiprocessing.Process(group=None, target=work1)
    # 啟動(dòng)進(jìn)程p1
    p1.start()
    # 創(chuàng)建進(jìn)程p2
    p2 = multiprocessing.Process(group=None, target=work2)
    # 設(shè)置p2守護(hù)主進(jìn)程
    # 第?種?式
    # p2.daemon = True  在start()之前設(shè)置，不然會(huì)拋異常
    # 啟動(dòng)進(jìn)程p2
    p2.start()
    time.sleep(2)
    print("主進(jìn)程運(yùn)行完畢！")
    # 第?種?式 
    p2.terminate()

執(zhí)行結(jié)果如下：

由于p2設(shè)置了守護(hù)主進(jìn)程，所以主進(jìn)程運(yùn)行完畢后，p2子進(jìn)程也隨之結(jié)束，work2任務(wù)停止，而work1繼續(xù)運(yùn)行至結(jié)束。

進(jìn)程池

當(dāng)需要?jiǎng)?chuàng)建的?進(jìn)程數(shù)量不多時(shí)，可以直接利?multiprocessing.Process動(dòng)態(tài)生成多個(gè)進(jìn)程，但如果要?jiǎng)?chuàng)建很多進(jìn)程時(shí)，?動(dòng)創(chuàng)建的話?作量會(huì)非常大，此時(shí)就可以?到multiprocessing模塊提供的Pool去創(chuàng)建一個(gè)進(jìn)程池。

multiprocessing.Pool常?函數(shù)：

apply_async(func, args, kwds)：使??阻塞?式調(diào)?func（任務(wù)并?執(zhí)?)，args為傳遞給func的參數(shù)列表，kwds為傳遞給func的關(guān)鍵字參數(shù)列表
apply(func, args, kwds)：使?阻塞?式調(diào)?func，必須等待上?個(gè)進(jìn)程執(zhí)行完任務(wù)后才能執(zhí)?下?個(gè)進(jìn)程，了解即可，幾乎不用
close()：關(guān)閉Pool,使其不再接受新的任務(wù)
terminate()：不管任務(wù)是否完成，?即終?
join()：主進(jìn)程阻塞，等待?進(jìn)程的退出，必須在close或terminate之后使?

初始化Pool時(shí),可以指定?個(gè)最?進(jìn)程數(shù)，當(dāng)有新的任務(wù)提交到Pool中時(shí)，如果進(jìn)程池還沒有滿，那么就會(huì)創(chuàng)建?個(gè)新的進(jìn)程?來執(zhí)?該任務(wù)，但如果進(jìn)程池已滿（池中的進(jìn)程數(shù)已經(jīng)達(dá)到指定的最?值），那么該任務(wù)就會(huì)等待，直到池中有進(jìn)程結(jié)束才會(huì)創(chuàng)建新的進(jìn)程來執(zhí)?。

from multiprocessing import Pool
import time
def work(i):
    print("work'{}'執(zhí)行中......".format(i), multiprocessing.current_process().name, multiprocessing.current_process().pid)
    time.sleep(2)
    print("work'{}'執(zhí)行完畢......".format(i))
if __name__ == '__main__':
    # 創(chuàng)建進(jìn)程池
    # Pool(3) 表示創(chuàng)建容量為3個(gè)進(jìn)程的進(jìn)程池
    pool = Pool(3)
    for i in range(10):
        # 利?進(jìn)程池同步執(zhí)?work任務(wù)，進(jìn)程池中的進(jìn)程會(huì)等待上?個(gè)進(jìn)程執(zhí)行完任務(wù)后才能執(zhí)?下?個(gè)進(jìn)程
        # pool.apply(work, (i, ))
        # 使?異步?式執(zhí)?work任務(wù)
        pool.apply_async(work, (i, ))
    # 進(jìn)程池關(guān)閉之后不再接受新的請(qǐng)求
    pool.close()
    # 等待po中所有子進(jìn)程結(jié)束，必須放在close()后面, 如果使?異步?式執(zhí)?work任務(wù)，主線程不再等待?線程執(zhí)?完畢再退出！
    pool.join()

執(zhí)行結(jié)果為：

從結(jié)果我們可以看出，只有3個(gè)子進(jìn)程在執(zhí)行任務(wù)，此處我們使用的是異步?式（pool.apply_async(work, (i, ))）執(zhí)?work任務(wù)，如果是以同步方式（pool.apply(work, (i, ))）執(zhí)行，進(jìn)程池中的進(jìn)程會(huì)等待上?個(gè)進(jìn)程執(zhí)行完任務(wù)后才能執(zhí)?下?個(gè)進(jìn)程。