本篇文章給大家?guī)?lái)了關(guān)于python的相關(guān)知識(shí)，其中主要介紹了關(guān)于多進(jìn)程的相關(guān)內(nèi)容，包括了什么是多進(jìn)程、進(jìn)程的創(chuàng)建、進(jìn)程間同步、進(jìn)程池等等，下面一起來(lái)看一下，希望對(duì)大家有幫助。

推薦學(xué)習(xí)：python視頻教程

一、什么是多進(jìn)程？

1. 進(jìn)程

程序：例如xxx.py這是程序，是一個(gè)靜態(tài)的

進(jìn)程：一個(gè)程序運(yùn)行起來(lái)后，代碼+用到的資源 稱(chēng)之為進(jìn)程，它是操作系統(tǒng)分配資源的基本單元。不僅可以通過(guò)線(xiàn)程完成多任務(wù)，進(jìn)程也是可以的

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

工作中，任務(wù)數(shù)往往大于cpu的核數(shù)，即一定有一些任務(wù)正在執(zhí)行，而另外一些任務(wù)在等待cpu進(jìn)行執(zhí)行，因此導(dǎo)致了有了不同的狀態(tài)

• 就緒態(tài)：運(yùn)行的條件都已經(jīng)慢去，正在等在cpu執(zhí)行
• 執(zhí)行態(tài)：cpu正在執(zhí)行其功能
• 等待態(tài)：等待某些條件滿(mǎn)足，例如一個(gè)程序sleep了，此時(shí)就處于等待態(tài)

二、進(jìn)程的創(chuàng)建-multiprocessing

1. Process類(lèi)語(yǔ)法說(shuō)明

multiprocessing 模塊通過(guò)創(chuàng)建一個(gè) Process 對(duì)象然后調(diào)用它的 start()方法來(lái)生成進(jìn)程，Process 與 threading.Thread API 相同。

語(yǔ)法格式：multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)

參數(shù)說(shuō)明：

• group：指定進(jìn)程組，大多數(shù)情況下用不到
• target：如果傳遞了函數(shù)的引用，可以任務(wù)這個(gè)子進(jìn)程就執(zhí)行這里的代碼
• name：給進(jìn)程設(shè)定一個(gè)名字，可以不設(shè)定
• args：給target指定的函數(shù)傳遞的參數(shù)，以元組的方式傳遞
• kwargs：給target指定的函數(shù)傳遞命名參數(shù)

multiprocessing.Process 對(duì)象具有如下方法和屬性：

2. 2個(gè)while循環(huán)一起執(zhí)行

# -*- coding:utf-8 -*-from multiprocessing import Processimport timedef run_proc():     """子進(jìn)程要執(zhí)行的代碼"""     while True:         print("----2----")         time.sleep(1)if __name__=='__main__':     p = Process(target=run_proc)     p.start()     while True:         print("----1----")         time.sleep(1)

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

說(shuō)明：創(chuàng)建子進(jìn)程時(shí)，只需要傳入一個(gè)執(zhí)行函數(shù)和函數(shù)的參數(shù)，創(chuàng)建一個(gè)Process實(shí)例，用start()方法啟動(dòng)

3. 進(jìn)程pid

# -*- coding:utf-8 -*-from multiprocessing import Processimport osimport timedef run_proc():     """子進(jìn)程要執(zhí)行的代碼"""     print('子進(jìn)程運(yùn)行中，pid=%d...' % os.getpid())  # os.getpid獲取當(dāng)前進(jìn)程的進(jìn)程號(hào)     print('子進(jìn)程將要結(jié)束...')if __name__ == '__main__':     print('父進(jìn)程pid: %d' % os.getpid())  # os.getpid獲取當(dāng)前進(jìn)程的進(jìn)程號(hào)     p = Process(target=run_proc)     p.start()

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

4. 給子進(jìn)程指定的函數(shù)傳遞參數(shù)

# -*- coding:utf-8 -*-from multiprocessing import Processimport osfrom time import sleepdef run_proc(name, age, **kwargs):     for i in range(10):         print('子進(jìn)程運(yùn)行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))         print(kwargs)         sleep(0.2)if __name__=='__main__':     p = Process(target=run_proc, args=('test',18), kwargs={"m":20})     p.start()     sleep(1)  # 1秒中之后，立即結(jié)束子進(jìn)程     p.terminate()     p.join()

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

5. 進(jìn)程間不同享全局變量

# -*- coding:utf-8 -*-from multiprocessing import Processimport osimport time  nums = [11, 22]def work1():     """子進(jìn)程要執(zhí)行的代碼"""     print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))     for i in range(3):         nums.append(i)         time.sleep(1)         print("in process1 pid=%d ,nums=%s" % (os.getpid(), nums))def work2():     """子進(jìn)程要執(zhí)行的代碼"""     print("in process2 pid=%d ,nums=%s" % (os.getpid(), nums))if __name__ == '__main__':     p1 = Process(target=work1)     p1.start()     p1.join()      p2 = Process(target=work2)     p2.start()

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

in process1 pid=11349 ,nums=[11, 22]in process1 pid=11349 ,nums=[11, 22, 0]in process1 pid=11349 , nums=[11, 22, 0, 1]in process1 pid=11349 ,nums=[11, 22, 0, 1, 2]in process2 pid=11350 ,nums=[11, 22]

三、進(jìn)程間同步-Queue

Process之間有時(shí)需要通信，操作系統(tǒng)提供了很多機(jī)制來(lái)實(shí)現(xiàn)進(jìn)程間的通信。

1. Queue類(lèi)語(yǔ)法說(shuō)明

2. Queue的使用

可以使用multiprocessing模塊的Queue實(shí)現(xiàn)多進(jìn)程之間的數(shù)據(jù)傳遞，Queue本身是一個(gè)消息列隊(duì)程序，首先用一個(gè)小實(shí)例來(lái)演示一下Queue的工作原理：

#coding=utf-8from multiprocessing import Queue q=Queue(3) #初始化一個(gè)Queue對(duì)象，最多可接收三條put消息q.put("消息1") q.put("消息2")print(q.full())  #Falseq.put("消息3")print(q.full()) #True#因?yàn)橄⒘嘘?duì)已滿(mǎn)下面的try都會(huì)拋出異常，第一個(gè)try會(huì)等待2秒后再拋出異常，第二個(gè)Try會(huì)立刻拋出異常try:     q.put("消息4",True,2)except:     print("消息列隊(duì)已滿(mǎn)，現(xiàn)有消息數(shù)量:%s"%q.qsize())try:     q.put_nowait("消息4")except:     print("消息列隊(duì)已滿(mǎn)，現(xiàn)有消息數(shù)量:%s"%q.qsize())#推薦的方式，先判斷消息列隊(duì)是否已滿(mǎn)，再寫(xiě)入if not q.full():     q.put_nowait("消息4")#讀取消息時(shí)，先判斷消息列隊(duì)是否為空，再讀取if not q.empty():     for i in range(q.qsize()):         print(q.get_nowait())

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

FalseTrue消息列隊(duì)已滿(mǎn)，現(xiàn)有消息數(shù)量:3消息列隊(duì)已滿(mǎn)，現(xiàn)有消息數(shù)量:3消息1消息2消息3

3. Queue實(shí)例

我們以Queue為例，在父進(jìn)程中創(chuàng)建兩個(gè)子進(jìn)程，一個(gè)往Queue里寫(xiě)數(shù)據(jù)，一個(gè)從Queue里讀數(shù)據(jù)：

from multiprocessing import Process, Queueimport os, time, random# 寫(xiě)數(shù)據(jù)進(jìn)程執(zhí)行的代碼:def write(q):     for value in ['A', 'B', 'C']:         print('Put %s to queue...' % value)         q.put(value)         time.sleep(random.random())# 讀數(shù)據(jù)進(jìn)程執(zhí)行的代碼:def read(q):     while True:         if not q.empty():             value = q.get(True)             print('Get %s from queue.' % value)             time.sleep(random.random())         else:             breakif __name__=='__main__':     # 父進(jìn)程創(chuàng)建Queue，并傳給各個(gè)子進(jìn)程：     q = Queue()     pw = Process(target=write, args=(q,))     pr = Process(target=read, args=(q,))     # 啟動(dòng)子進(jìn)程pw，寫(xiě)入:     pw.start()         # 等待pw結(jié)束:     pw.join()     # 啟動(dòng)子進(jìn)程pr，讀取:     pr.start()     pr.join()     # pr進(jìn)程里是死循環(huán)，無(wú)法等待其結(jié)束，只能強(qiáng)行終止:     print('')     print('所有數(shù)據(jù)都寫(xiě)入并且讀完')

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

四、進(jìn)程間同步-Lock

鎖是為了確保數(shù)據(jù)一致性。比如讀寫(xiě)鎖，每個(gè)進(jìn)程給一個(gè)變量增加 1，但是如果在一個(gè)進(jìn)程讀取但還沒(méi)有寫(xiě)入的時(shí)候，另外的進(jìn)程也同時(shí)讀取了，并寫(xiě)入該值，則最后寫(xiě)入的值是錯(cuò)誤的，這時(shí)候就需要加鎖來(lái)保持?jǐn)?shù)據(jù)一致性。

通過(guò)使用Lock來(lái)控制一段代碼在同一時(shí)間只能被一個(gè)進(jìn)程執(zhí)行。Lock對(duì)象的兩個(gè)方法，acquire()用來(lái)獲取鎖，release()用來(lái)釋放鎖。當(dāng)一個(gè)進(jìn)程調(diào)用acquire()時(shí)，如果鎖的狀態(tài)為unlocked，那么會(huì)立即修改為locked并返回，這時(shí)該進(jìn)程即獲得了鎖。如果鎖的狀態(tài)為locked，那么調(diào)用acquire()的進(jìn)程則阻塞。

1. Lock的語(yǔ)法說(shuō)明：

• lock = multiprocessing.Lock()： 創(chuàng)建一個(gè)鎖

• lock.acquire() ：獲取鎖

• lock.release() ：釋放鎖

• with lock：自動(dòng)獲取、釋放鎖 類(lèi)似于 with open() as f:

2. 程序不加鎖時(shí)：

import multiprocessingimport timedef add(num, value):     print('add{0}:num={1}'.format(value, num))     for i in range(0, 2):         num += value        print('add{0}:num={1}'.format(value, num))         time.sleep(1)if __name__ == '__main__':     lock = multiprocessing.Lock()     num = 0     p1 = multiprocessing.Process(target=add, args=(num, 1))     p2 = multiprocessing.Process(target=add, args=(num, 2))     p1.start()     p2.start()

運(yùn)行結(jié)果：運(yùn)得沒(méi)有順序，兩個(gè)進(jìn)程交替運(yùn)行

add1:num=0add1:num=1add2:num=0add2:num=2add1:num=2add2:num=4

3. 程序加鎖時(shí)：

import multiprocessingimport timedef add(num, value, lock):     try:         lock.acquire()         print('add{0}:num={1}'.format(value, num))         for i in range(0, 2):             num += value            print('add{0}:num={1}'.format(value, num))             time.sleep(1)     except Exception as err:         raise err    finally:         lock.release()if __name__ == '__main__':     lock = multiprocessing.Lock()     num = 0     p1 = multiprocessing.Process(target=add, args=(num, 1, lock))     p2 = multiprocessing.Process(target=add, args=(num, 2, lock))     p1.start()     p2.start()

運(yùn)行結(jié)果：只有當(dāng)其中一個(gè)進(jìn)程執(zhí)行完成后，其它的進(jìn)程才會(huì)去執(zhí)行，且誰(shuí)先搶到鎖誰(shuí)先執(zhí)行

add1:num=0add1:num=1add1:num=2add2:num=0add2:num=2add2:num=4

五、進(jìn)程池Pool

當(dāng)需要?jiǎng)?chuàng)建的子進(jìn)程數(shù)量不多時(shí)，可以直接利用multiprocessing中的Process動(dòng)態(tài)成生多個(gè)進(jìn)程，但如果是上百甚至上千個(gè)目標(biāo)，手動(dòng)的去創(chuàng)建進(jìn)程的工作量巨大，此時(shí)就可以用到multiprocessing模塊提供的Pool方法。

1. Pool類(lèi)語(yǔ)法說(shuō)明

語(yǔ)法格式：multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

參數(shù)說(shuō)明：

• processes：工作進(jìn)程數(shù)目，如果 processes 為 None，則使用 os.cpu_count() 返回的值。

• initializer：如果 initializer 不為 None，則每個(gè)工作進(jìn)程將會(huì)在啟動(dòng)時(shí)調(diào)用 initializer(*initargs)。

• maxtasksperchild：一個(gè)工作進(jìn)程在它退出或被一個(gè)新的工作進(jìn)程代替之前能完成的任務(wù)數(shù)量，為了釋放未使用的資源。

• context：用于指定啟動(dòng)的工作進(jìn)程的上下文。

兩種方式向進(jìn)程池提交任務(wù)：

• apply(func[, args[, kwds]])：阻塞方式。

• apply_async(func[, args[, kwds]])：非阻塞方式。使用非阻塞方式調(diào)用func（并行執(zhí)行，堵塞方式必須等待上一個(gè)進(jìn)程退出才能執(zhí)行下一個(gè)進(jìn)程），args為傳遞給func的參數(shù)列表，kwds為傳遞給func的關(guān)鍵字參數(shù)列表

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

2. Pool實(shí)例

初始化Pool時(shí)，可以指定一個(gè)最大進(jìn)程數(shù)，當(dāng)有新的請(qǐng)求提交到Pool中時(shí)，如果池還沒(méi)有滿(mǎn)，那么就會(huì)創(chuàng)建一個(gè)新的進(jìn)程用來(lái)執(zhí)行該請(qǐng)求；但如果池中的進(jìn)程數(shù)已經(jīng)達(dá)到指定的最大值，那么該請(qǐng)求就會(huì)等待，直到池中有進(jìn)程結(jié)束，才會(huì)用之前的進(jìn)程來(lái)執(zhí)行新的任務(wù)，請(qǐng)看下面的實(shí)例：

# -*- coding:utf-8 -*-from multiprocessing import Poolimport os, time, randomdef worker(msg):     t_start = time.time()     print("%s開(kāi)始執(zhí)行,進(jìn)程號(hào)為%d" % (msg,os.getpid()))     # random.random()隨機(jī)生成0~1之間的浮點(diǎn)數(shù)     time.sleep(random.random()*2)      t_stop = time.time()     print(msg,"執(zhí)行完畢，耗時(shí)%0.2f" % (t_stop-t_start))po = Pool(3)  # 定義一個(gè)進(jìn)程池，最大進(jìn)程數(shù)3for i in range(0,10):     # Pool().apply_async(要調(diào)用的目標(biāo),(傳遞給目標(biāo)的參數(shù)元祖,))     # 每次循環(huán)將會(huì)用空閑出來(lái)的子進(jìn)程去調(diào)用目標(biāo)     po.apply_async(worker,(i,))print("----start----")po.close()       # 關(guān)閉進(jìn)程池，關(guān)閉后po不再接收新的請(qǐng)求po.join()       # 等待po中所有子進(jìn)程執(zhí)行完成，必須放在close語(yǔ)句之后print("-----end-----")

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

----start---- 0開(kāi)始執(zhí)行,進(jìn)程號(hào)為21466 1開(kāi)始執(zhí)行,進(jìn)程號(hào)為21468 2開(kāi)始執(zhí)行,進(jìn)程號(hào)為21467 0 執(zhí)行完畢，耗時(shí)1.01 3開(kāi)始執(zhí)行,進(jìn)程號(hào)為21466 2 執(zhí)行完畢，耗時(shí)1.24 4開(kāi)始執(zhí)行,進(jìn)程號(hào)為21467 3 執(zhí)行完畢，耗時(shí)0.56 5開(kāi)始執(zhí)行,進(jìn)程號(hào)為21466 1 執(zhí)行完畢，耗時(shí)1.68 6開(kāi)始執(zhí)行,進(jìn)程號(hào)為21468 4 執(zhí)行完畢，耗時(shí)0.67 7開(kāi)始執(zhí)行,進(jìn)程號(hào)為21467 5 執(zhí)行完畢，耗時(shí)0.83 8開(kāi)始執(zhí)行,進(jìn)程號(hào)為21466 6 執(zhí)行完畢，耗時(shí)0.75 9開(kāi)始執(zhí)行,進(jìn)程號(hào)為21468 7 執(zhí)行完畢，耗時(shí)1.03 8 執(zhí)行完畢，耗時(shí)1.05 9 執(zhí)行完畢，耗時(shí)1.69 -----end-----

3. 進(jìn)程池中的Queue

如果要使用Pool創(chuàng)建進(jìn)程，就需要使用multiprocessing.Manager()中的Queue()

而不是multiprocessing.Queue()，否則會(huì)得到一條如下的錯(cuò)誤信息：RuntimeError: Queue objects should only be shared between processes through inheritance.

下面的實(shí)例演示了進(jìn)程池中的進(jìn)程如何通信：

# -*- coding:utf-8 -*-# 修改import中的Queue為Managerfrom multiprocessing import Manager,Poolimport os,time,randomdef reader(q):     print("reader啟動(dòng)(%s),父進(jìn)程為(%s)" % (os.getpid(), os.getppid()))     for i in range(q.qsize()):         print("reader從Queue獲取到消息：%s" % q.get(True))def writer(q):     print("writer啟動(dòng)(%s),父進(jìn)程為(%s)" % (os.getpid(), os.getppid()))     for i in "itcast":         q.put(i)if __name__=="__main__":     print("(%s) start" % os.getpid())     q = Manager().Queue()  # 使用Manager中的Queue     po = Pool()     po.apply_async(writer, (q,))      time.sleep(1)  # 先讓上面的任務(wù)向Queue存入數(shù)據(jù)，然后再讓下面的任務(wù)開(kāi)始從中取數(shù)據(jù)      po.apply_async(reader, (q,))     po.close()     po.join()     print("(%s) End" % os.getpid())

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

(11095) start writer啟動(dòng)(11097),父進(jìn)程為(11095)reader啟動(dòng)(11098),父進(jìn)程為(11095)reader從Queue獲取到消息：i reader從Queue獲取到消息：t reader從Queue獲取到消息：c reader從Queue獲取到消息：a reader從Queue獲取到消息：s reader從Queue獲取到消息：t(11095) End

六、進(jìn)程、線(xiàn)程對(duì)比

1. 功能

進(jìn)程：能夠完成多任務(wù)，比如 在一臺(tái)電腦上能夠同時(shí)運(yùn)行多個(gè)QQ
線(xiàn)程：能夠完成多任務(wù)，比如 一個(gè)QQ中的多個(gè)聊天窗口

定義的不同

• 進(jìn)程是系統(tǒng)進(jìn)行資源分配和調(diào)度的一個(gè)獨(dú)立單位.

• 線(xiàn)程是進(jìn)程的一個(gè)實(shí)體,是CPU調(diào)度和分派的基本單位,它是比進(jìn)程更小的能獨(dú)立運(yùn)行的基本單位.線(xiàn)程自己基本上不擁有系統(tǒng)資源,只擁有一點(diǎn)在運(yùn)行中必不可少的資源(如程序計(jì)數(shù)器,一組寄存器和棧),但是它可與同屬一個(gè)進(jìn)程的其他的線(xiàn)程共享進(jìn)程所擁有的全部資源.

2. 區(qū)別

• 一個(gè)程序至少有一個(gè)進(jìn)程,一個(gè)進(jìn)程至少有一個(gè)線(xiàn)程.
  -線(xiàn)程的劃分尺度小于進(jìn)程(資源比進(jìn)程少)，使得多線(xiàn)程程序的并發(fā)性高。
  -進(jìn)程在執(zhí)行過(guò)程中擁有獨(dú)立的內(nèi)存單元，而多個(gè)線(xiàn)程共享內(nèi)存，從而極大地提高了程序的運(yùn)行效率
  
• 線(xiàn)線(xiàn)程不能夠獨(dú)立執(zhí)行，必須依存在進(jìn)程中
• 可以將進(jìn)程理解為工廠中的一條流水線(xiàn)，而其中的線(xiàn)程就是這個(gè)流水線(xiàn)上的工人

3. 優(yōu)缺點(diǎn)

• 線(xiàn)程：線(xiàn)程執(zhí)行開(kāi)銷(xiāo)小，但不利于資源的管理和保護(hù)
• 進(jìn)程：進(jìn)程執(zhí)行開(kāi)銷(xiāo)大，但利于資源的管理和保護(hù)

推薦學(xué)習(xí)：python視頻教程