Python 多進程原理及實現(xiàn)
1 進程的基本概念
什么是進程？
進程就是一個程序在一個數(shù)據(jù)集上的一次動態(tài)執(zhí)行過程。進程一般由程序、數(shù)據(jù)集、進程控制塊三部分組成。我們編寫的程序用來描述進程要完成哪些功能以及如何完成；數(shù)據(jù)集則是程序在執(zhí)行過程中所需要使用的資源；進程控制塊用來記錄進程的外部特征，描述進程的執(zhí)行變化過程，系統(tǒng)可以利用它來控制和管理進程，它是系統(tǒng)感知進程存在的唯一標(biāo)志。
進程的生命周期：創(chuàng)建（New）、就緒（Runnable）、運行（Running）、阻塞（Block）、銷毀（Destroy）
進程的狀態(tài)（分類）：（Actived）活動進程、可見進程（Visiable）、后臺進程（Background）、服務(wù)進程（Service）、空進程
2 父進程和子進程
Linux 操作系統(tǒng)提供了一個 fork() 函數(shù)用來創(chuàng)建子進程，這個函數(shù)很特殊，調(diào)用一次，返回兩次，因為操作系統(tǒng)是將當(dāng)前的進程（父進程）復(fù)制了一份（子進程），然后分別在父進程和子進程內(nèi)返回。子進程永遠返回0，而父進程返回子進程的 PID。我們可以通過判斷返回值是不是 0 來判斷當(dāng)前是在父進程還是子進程中執(zhí)行。
在 Python 中同樣提供了 fork() 函數(shù)，此函數(shù)位于 os 模塊下。
# -*- coding: utf-8 -*-
import os
import time
print("在創(chuàng)建子進程前: pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
pid = os.fork()
if pid == 0:
print("子進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
time.sleep(5)
else:
print("父進程信息: pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
# pid表示回收的子進程的pid
#pid, result = os.wait() # 回收子進程資源 阻塞
time.sleep(5)
#print("父進程：回收的子進程pid=%d" % pid)
#print("父進程：子進程退出時 result=%d" % result)
# 下面的內(nèi)容會被打印兩次，一次是在父進程中，一次是在子進程中。
# 父進程中拿到的返回值是創(chuàng)建的子進程的pid，大于0
print("fork創(chuàng)建完后: pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
2.1 父子進程如何區(qū)分?
子進程是父進程通過fork()產(chǎn)生出來的，pid = os.fork()
通過返回值pid是否為0，判斷是否為子進程，如果是0，則表示是子進程
由于 fork() 是 Linux 上的概念，所以如果要跨平臺，最好還是使用 subprocess 模塊來創(chuàng)建子進程。
2.2 子進程如何回收？
python中采用os.wait()方法用來回收子進程占用的資源
pid, result = os.wait() # 回收子進程資源 阻塞，等待子進程執(zhí)行完成回收
如果有子進程沒有被回收的，但是父進程已經(jīng)死掉了，這個子進程就是僵尸進程。
3 Python進程模塊
python的進程multiprocessing模塊有多種創(chuàng)建進程的方式，每種創(chuàng)建方式和進程資源的回收都不太相同，下面分別針對Process,Pool及系統(tǒng)自帶的fork三種進程分析。
3.1 fork()
import os pid = os.fork() # 創(chuàng)建一個子進程 os.wait() # 等待子進程結(jié)束釋放資源 pid為0的代表子進程。
缺點：
1.兼容性差，只能在類linux系統(tǒng)下使用，windows系統(tǒng)不可使用；
2.擴展性差，當(dāng)需要多條進程的時候，進程管理變得很復(fù)雜；
3.會產(chǎn)生“孤兒”進程和“僵尸”進程，需要手動回收資源。
優(yōu)點：
是系統(tǒng)自帶的接近低層的創(chuàng)建方式，運行效率高。
3.2 Process進程
multiprocessing模塊提供Process類實現(xiàn)新建進程
# -*- coding: utf-8 -*-
import os
from multiprocessing import Process
import time
def fun(name):
print("2 子進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
print("hello " + name)
def test():
print('ssss')
if __name__ == "__main__":
print("1 主進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
ps = Process(target=fun, args=('jingsanpang', ))
print("111 ##### ps pid: " + str(ps.pid) + ", ident:" + str(ps.ident))
print("3 進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
print(ps.is_alive()) # 啟動之前 is_alive為False(系統(tǒng)未創(chuàng)建)
ps.start()
print(ps.is_alive()) # 啟動之后，is_alive為True(系統(tǒng)已創(chuàng)建)
print("222 #### ps pid: " + str(ps.pid) + ", ident:" + str(ps.ident))
print("4 進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
ps.join() # 等待子進程完成任務(wù) 類似于os.wait()
print(ps.is_alive())
print("5 進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
ps.terminate() #終斷進程
print("6 進程信息： pid=%s, ppid=%s" % (os.getpid(), os.getppid()))
特點：
1.注意：Process對象可以創(chuàng)建進程，但Process對象不是進程，其刪除與否與系統(tǒng)資源是否被回收沒有直接的關(guān)系。
2.主進程執(zhí)行完后會默認(rèn)等待子進程結(jié)束后回收資源，不需要手動回收資源；join()函數(shù)用來控制子進程結(jié)束的順序,其內(nèi)部也有一個清除僵尸進程的函數(shù)，可以回收資源；
3.Process進程創(chuàng)建時，子進程會將主進程的Process對象完全復(fù)制一份，這樣在主進程和子進程各有一個 Process對象，但是p.start()啟動的是子進程，主進程中的Process對象作為一個靜態(tài)對象存在，不執(zhí)行。
4.當(dāng)子進程執(zhí)行完畢后，會產(chǎn)生一個僵尸進程，其會被join函數(shù)回收，或者再有一條進程開啟，start函數(shù)也會回收僵尸進程，所以不一定需要寫join函數(shù)。
5.windows系統(tǒng)在子進程結(jié)束后會立即自動清除子進程的Process對象，而linux系統(tǒng)子進程的Process對象如果沒有join函數(shù)和start函數(shù)的話會在主進程結(jié)束后統(tǒng)一清除。
另外還可以通過繼承Process對象來重寫run方法創(chuàng)建進程
3.3 進程池POOL (多個進程)
import multiprocessing
import time
def work(msg):
mult_proces_name = multiprocessing.current_process().name
print('process: ' + mult_proces_name + '-' + msg)
if __name__ == "__main__":
pool = multiprocessing.Pool(processes=5) # 創(chuàng)建5個進程
for i in range(20):
msg = "process %d" %(i)
pool.apply_async(work, (msg, ))
pool.close() # 關(guān)閉進程池，表示不能在往進程池中添加進程
pool.join() # 等待進程池中的所有進程執(zhí)行完畢，必須在close()之后調(diào)用
print("Sub-process all done.")
上述代碼中的pool.apply_async()是apply()函數(shù)的變體，apply_async()是apply()的并行版本，apply()是apply_async()的阻塞版本，使用apply()主進程會被阻塞直到函數(shù)執(zhí)行結(jié)束，所以說是阻塞版本。apply()既是Pool的方法，也是Python內(nèi)置的函數(shù)，兩者等價?？梢钥吹捷敵鼋Y(jié)果并不是按照代碼for循環(huán)中的順序輸出的。
多個子進程并返回值
apply_async()本身就可以返回被進程調(diào)用的函數(shù)的返回值。上一個創(chuàng)建多個子進程的代碼中，如果在函數(shù)func中返回一個值，那么pool.apply_async(func, (msg, ))的結(jié)果就是返回pool中所有進程的值的對象（注意是對象，不是值本身）。
import multiprocessing
import time
def func(msg):
return multiprocessing.current_process().name + '-' + msg
if __name__ == "__main__":
pool = multiprocessing.Pool(processes=4) # 創(chuàng)建4個進程
results = []
for i in range(20):
msg = "process %d" %(i)
results.append(pool.apply_async(func, (msg, )))
pool.close() # 關(guān)閉進程池，表示不能再往進程池中添加進程，需要在join之前調(diào)用
pool.join() # 等待進程池中的所有進程執(zhí)行完畢
print ("Sub-process(es) done.")
for res in results:
print (res.get())
與之前的輸出不同，這次的輸出是有序的。
如果電腦是八核，建立8個進程，在Ubuntu下輸入top命令再按下大鍵盤的1，可以看到每個CPU的使用率是比較平均的
4 進程間通信方式
管道pipe：管道是一種半雙工的通信方式，數(shù)據(jù)只能單向流動，而且只能在具有親緣關(guān)系的進程間使用。進程的親緣關(guān)系通常是指父子進程關(guān)系。
命名管道FIFO：有名管道也是半雙工的通信方式，但是它允許無親緣關(guān)系進程間的通信。
消息隊列MessageQueue：消息隊列是由消息的鏈表，存放在內(nèi)核中并由消息隊列標(biāo)識符標(biāo)識。消息隊列克服了信號傳遞信息少、管道只能承載無格式字節(jié)流以及緩沖區(qū)大小受限等缺點。
共享存儲SharedMemory：共享內(nèi)存就是映射一段能被其他進程所訪問的內(nèi)存，這段共享內(nèi)存由一個進程創(chuàng)建，但多個進程都可以訪問。共享內(nèi)存是最快的 IPC 方式，它是針對其他進程間通信方式運行效率低而專門設(shè)計的。它往往與其他通信機制，如信號兩，配合使用，來實現(xiàn)進程間的同步和通信。
以上幾種進程間通信方式中，消息隊列是使用的比較頻繁的方式。
（1）管道pipe
import multiprocessing
def foo(conn):
conn.send('hello father') #向管道pipe發(fā)消息
print(conn.recv())
if __name__ == '__main__':
conn1,conn2=multiprocessing.Pipe(True) #開辟兩個口，都是能進能出，括號中如果False即單向通信
p=multiprocessing.Process(target=foo,args=(conn1,)) #子進程使用sock口，調(diào)用foo函數(shù)
p.start()
print(conn2.recv()) #主進程使用conn口接收，從管道（Pipe）中讀取消息
conn2.send('hi son') #主進程使用conn口發(fā)送
（2）消息隊列Queue
Queue是多進程的安全隊列，可以使用Queue實現(xiàn)多進程之間的數(shù)據(jù)傳遞。
Queue的一些常用方法：
- Queue.qsize()：返回當(dāng)前隊列包含的消息數(shù)量；
- Queue.empty()：如果隊列為空，返回True，反之False ；
- Queue.full()：如果隊列滿了，返回True,反之False；
- Queue.get():獲取隊列中的一條消息，然后將其從列隊中移除，可傳參超時時長。
- Queue.get_nowait()：相當(dāng)Queue.get(False),取不到值時觸發(fā)異常：Empty；
- Queue.put():將一個值添加進數(shù)列，可傳參超時時長。
- Queue.put_nowait():相當(dāng)于Queue.get(False),當(dāng)隊列滿了時報錯：Full。
案例：
from multiprocessing import Process, Queue
import time
def write(q):
for i in ['A', 'B', 'C', 'D', 'E']:
print('Put %s to queue' % i)
q.put(i)
time.sleep(0.5)
def read(q):
while True:
v = q.get(True)
print('get %s from queue' % v)
if __name__ == '__main__':
q = Queue()
pw = Process(target=write, args=(q,))
pr = Process(target=read, args=(q,))
print('write process = ', pw)
print('read process = ', pr)
pw.start()
pr.start()
pw.join()
pr.join()
pr.terminate()
pw.terminate()
Queue和pipe只是實現(xiàn)了數(shù)據(jù)交互，并沒實現(xiàn)數(shù)據(jù)共享，即一個進程去更改另一個進程的數(shù)據(jù)。
注：進程間通信應(yīng)該盡量避免使用共享數(shù)據(jù)的方式
5 多進程實現(xiàn)生產(chǎn)者消費者
以下通過多進程實現(xiàn)生產(chǎn)者，消費者模式
import multiprocessing
from multiprocessing import Process
from time import sleep
import time
class MultiProcessProducer(multiprocessing.Process):
def __init__(self, num, queue):
"""Constructor"""
multiprocessing.Process.__init__(self)
self.num = num
self.queue = queue
def run(self):
t1 = time.time()
print('producer start ' + str(self.num))
for i in range(1000):
self.queue.put((i, self.num))
# print 'producer put', i, self.num
t2 = time.time()
print('producer exit ' + str(self.num))
use_time = str(t2 - t1)
print('producer ' + str(self.num) + ',
use_time: '+ use_time)
class MultiProcessConsumer(multiprocessing.Process):
def __init__(self, num, queue):
"""Constructor"""
multiprocessing.Process.__init__(self)
self.num = num
self.queue = queue
def run(self):
t1 = time.time()
print('consumer start ' + str(self.num))
while True:
d = self.queue.get()
if d != None:
# print 'consumer get', d, self.num
continue
else:
break
t2 = time.time()
print('consumer exit ' + str(self.num))
print('consumer ' + str(self.num) + ', use time:' + str(t2 - t1))
def main():
# create queue
queue = multiprocessing.Queue()
# create processes
producer = []
for i in range(5):
producer.append(MultiProcessProducer(i, queue))
consumer = []
for i in range(5):
consumer.append(MultiProcessConsumer(i, queue))
# start processes
for i in range(len(producer)):
producer[i].start()
for i in range(len(consumer)):
consumer[i].start()
# wait for processs to exit
for i in range(len(producer)):
producer[i].join()
for i in range(len(consumer)):
queue.put(None)
for i in range(len(consumer)):
consumer[i].join()
print('all done finish')
if __name__ == "__main__":
main()
6 總結(jié)
python中的多進程創(chuàng)建有以下兩種方式：
（1）fork子進程
（2）采用 multiprocessing 這個庫創(chuàng)建子進程
需要注意的是隊列中queue.Queue是線程安全的，但并不是進程安全，所以多進程一般使用線程、進程安全的multiprocessing.Queue()
另外, 進程池使用 multiprocessing.Pool實現(xiàn)，pool = multiprocessing.Pool(processes = 3)，產(chǎn)生一個進程池，pool.apply_async實現(xiàn)非租塞模式，pool.apply實現(xiàn)阻塞模式。
apply_async和 apply函數(shù)，前者是非阻塞的，后者是阻塞。可以看出運行時間相差的倍數(shù)正是進程池數(shù)量。
同時可以通過result.append(pool.apply_async(func, (msg, )))獲取非租塞式調(diào)用結(jié)果信息的。
以上就是Python 多進程原理及實現(xiàn)的詳細內(nèi)容，更多關(guān)于python 多進程的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
這篇文章主要介紹了Python實現(xiàn)的企業(yè)粉絲抽獎功能,涉及Python數(shù)值運算與隨機數(shù)生成相關(guān)操作技巧,需要的朋友可以參考下2019-07-07
這篇文章主要介紹了通過實例解析Python文件操作實現(xiàn)步驟,文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2020-09-09
本文是結(jié)合前面的三篇關(guān)于python制作爬蟲的基礎(chǔ)文章，給大家分享的一份爬取糗事百科的小段子的源碼，有需要的小伙伴可以參考下2016-09-09
這篇文章主要介紹了如何利用python的matplotlib繪制雙Y軸圖像,文中有非常詳細的代碼示例,對正在學(xué)習(xí)python的小伙伴們有很好地幫助,需要的朋友可以參考下2022-04-04
Python數(shù)據(jù)分析之獲取雙色球歷史信息的方法示例
這篇文章主要介紹了Python數(shù)據(jù)分析之獲取雙色球歷史信息的方法,涉及Python網(wǎng)頁抓取、正則匹配、文件讀寫及數(shù)值運算等相關(guān)操作技巧,需要的朋友可以參考下2018-02-02
這篇文章主要介紹了python 爬取馬蜂窩景點翻頁文字評論的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2020-01-01
這篇文章主要介紹了Python如何輸出警告信息，文中講解非常細致，代碼幫助大家更好的理解和學(xué)習(xí)，感興趣的朋友可以了解下2020-07-07
這篇文章主要為大家詳細介紹了python分析作業(yè)提交情況，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-11-11
聊聊python中令人迷惑的duplicated和drop_duplicates()用法
這篇文章主要介紹了聊聊python中令人迷惑的duplicated和drop_duplicates()用法，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05
Pycharm創(chuàng)建項目時如何自動添加頭部信息
這篇文章主要介紹了Pycharm創(chuàng)建項目時 自動添加頭部信息,需要的朋友可以參考下2019-11-11