Python並發編程之進程
一、理論概念
1、定義
進程（Process 也可以稱為重量級進程）是程序的一次執行。在每個進程中都有自己的地址空間、內存、數據棧以及記錄運行的輔助數據，它是系統進行資源分配和調度的一個獨立單位。
2、並行和並發
並行：並行是指多個任務同一時間執行；
並發：是指在資源有限的情況下，兩個任務相互交替著使用資源；
3、同步和異常
同步是指多個任務在執行時有一個先後的順序，必須是一個任務執行完成另外一個任務才能執行；
異步是指多個任務在執行時沒有先後順序，多個任務可以同時執行；
4、同步/異步/阻塞/非阻塞/
同步阻塞：這個阻塞的形成效率是最低的；比如你在下載一個東西是，你一直盯著下載進度條，到達100%時下載完成；
同步體現在：你等待下載完成通知；
阻塞體現在：等待下載的過程中，不能做別的事情
同步非阻塞：你在下載東西時，你把任務提交後就去幹別的事情了，只是每過一段時間就看一下是不是下載完成；
同步體現在：等待下載完成通知；
非阻塞提現在：等待下載完成通知過程中，去幹別的事情了，只是時不時會瞄一眼進度條；
異步阻塞：你在下載東西時換了一個現在使用的客戶端比如迅雷，下載完成後會有一個提示音，但是這時候你仍然一直在等待那個完成後的提示音；
異步體現在：下載完成時有提示音；
阻塞體現在：等待下載完成提示音時，不做任何事情；
異步非阻塞：你然然使用的是迅雷下載軟件，這時候你把下載任務提交後就去幹別的事情去了，當你聽到‘叮’以後就知道下載完成；
異步體現在：下載完成叮一聲完成通知
非阻塞體現在：等待下載完成“叮”一聲通知過程中，去幹別的任務了，只需要接收“叮”聲通知即可；
二、進程的創建與結束
multiprocessing模塊：multiprocess不是一個模塊而是python中一個操作、管理進程的包。 之所以叫multi是取自multiple的多功能的意思,在這個包中幾乎包含了和進程有關的所有子模塊。由於提供的子模塊非常多，為了方便大家歸類記憶，我將這部分大致分為四個部分：創建進程部分，進程同步部分，進程池部分，進程之間數據共享。
Process模塊的各種方法介紹
Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化得到的對象，表示一個子進程中的任務（尚未啟動）
強調：
1. 需要使用關鍵字的方式來指定參數
2. args指定的為傳給target函數的位置參數，是一個元組形式，必須有逗號
參數介紹：
group參數未使用，值始終為None
target表示調用對象，即子進程要執行的任務
args表示調用對象的位置參數元組，args=(1,2,‘egon‘,)
kwargs表示調用對象的字典,kwargs={‘name‘:‘egon‘,‘age‘:18}
name為子進程的名稱
p.start()：啟動進程，並調用該子進程中的p.run()
p.run():進程啟動時運行的方法，正是它去調用target指定的函數，我們自定義類的類中一定要實現該方法
p.terminate():強制終止進程p，不會進行任何清理操作，如果p創建了子進程，該子進程就成了僵屍進程，使用該方法需要特別小心這種情況。如果p還保存了一個鎖那麽也將不會被釋放，進而導致死鎖
p.is_alive():如果p仍然運行，返回True
p.join([timeout]):主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）。timeout是可選的超時時間，需要強調的是，p.join只能join住start開啟的進程，而不能join住run開啟的進程
在windows中使用process註意事項：
在Windows操作系統中由於沒有fork(linux操作系統中創建進程的機制)，在創建子進程的時候會自動 import 啟動它的這個文件，而在 import 的時候又執行了整個文件。因此如果將process()直接寫在文件中就會無限遞歸創建子進程報錯。所以必須把創建子進程的部分使用if __name__ ==‘__main__’ 判斷保護起來，import 的時候 ，就不會遞歸運行了。
process模塊創建進程：
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#Author: caoyf
import time
from multiprocessing import Process
def func(name):
print(‘hello %s‘%name)
print(‘我是子進程‘)
if __name__ == ‘__main__‘:
p = Process(target=func,args=(‘caoyf‘,)) #在實例化時候，args的參數必須是一個元祖形式(註冊一個子進程)
p.start() #啟動一個子進程
time.sleep(3)
print(‘執行主進程內容了‘)
創建第一個進程
多個進程同時運行，子進程的執行順序不是根據啟動的順序來決定的；
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#Author: caoyf
import time
from multiprocessing import Process
def func(name):
print(‘hello %s‘%name)
time.sleep(2)
if __name__ == ‘__main__‘:
p_lst = []
for i in range(10):
p = Process(target=func, args=(‘caoyf‘,))
p.start()
p_lst.append(p)
for p in p_lst: p.join() # 是感知一個子進程的結束,將異步的程序改為同步
print(‘父進程在運行‘)
多個進程同時運行
另一種開啟進程的方法，繼承process的形式
#!/usr/bin/env python
# -*- coding:utf-8 -*-
#Author: caoyf
import time
import os
from multiprocessing import Process
class Func(Process):
def __init__(self,name):
super().__init__()
self.name = name
def run(self):
print(os.getpid())
print(‘%s正在和小明聊天‘%self.name)
if __name__ == ‘__main__‘:
p1 = Func(‘caoyf‘)
p2 = Func(‘Zhao‘)
p1.start()
p2.start()
p1.join()
p2.join()
繼承的方式開啟進程
守護進程：會隨著主進程的結束而結束，進程之間是相互獨立的，主進程的代碼運行結束，守護進程也會隨即結束
#!/usr/bin/env python # -*- coding:utf-8 -*- #Author: caoyf import time import os from multiprocessing import Process def foo(): print(‘start123‘) time.sleep(2) print(‘end123‘) def func(): print(‘start456‘) time.sleep(5) print(‘end456‘) if __name__ == ‘__main__‘: p1 = Process(target=foo) p2 = Process(target=func) p1.daemon = True p1.start() p2.start() time.sleep(0.1) print(‘main------------‘)#打印該行則主進程代碼結束,則守護進程p1應該被終止.#可能會有p1任務執行的打印信息123,因為主進程打印main--- # -時,p1也執行了,但是隨即被終止.守護進程
三、進程同步（multiprocessing.Lock\Spemaphore\Event）
鎖（Lock)：
資源是有限的，多個進程如果對同一個對象進行操作，則有可能造成資源的爭用，甚至導致死鎖，在並發進程中就可以利用鎖進行操作來避免訪問的沖突；
加鎖可以保證多個進程修改同一塊數據時，同一時間只能有一個任務可以進行修改，即串行的修改，但是速度就變慢了，但犧牲了速度卻保證了數據安全。
雖然可以用文件共享數據實現進程間通信，但問題是：
1.效率低（共享數據基於文件，而文件是硬盤上的數據）
2.需要自己加鎖處理
我們可以模擬一個火車搶票的過程，當過個客戶同時對一個程序發起訪問時，假設此時有5張票，有10個人搶
from multiprocessing import Process,Lock import time,json,random def search(): dic=json.load(open(‘db‘)) print(‘\033[43m剩余票數%s\033[0m‘ %dic[‘count‘]) def get(): dic=json.load(open(‘db‘)) time.sleep(random.random()) #模擬讀數據的網絡延遲 if dic[‘count‘] >0: dic[‘count‘]-=1 time.sleep(random.random()) #模擬寫數據的網絡延遲 json.dump(dic,open(‘db‘,‘w‘)) print(‘\033[32m購票成功\033[0m‘) else: print(‘\033[31m購票失敗\033[0m‘) def task(lock): search() lock.acquire() get() lock.release() if __name__ == ‘__main__‘: lock = Lock() for i in range(100): #模擬並發100個客戶端搶票 p=Process(target=task,args=(lock,)) p.start()搶火車票
信號量：
信號量Semaphore是同時允許一定數量的線程更改數據 。
#!/usr/bin/env python # -*- coding:utf-8 -*- #Author: caoyf import time import random from multiprocessing import Semaphore from multiprocessing import Process def f(i,a): a.acquire() print(‘%s走進了房間‘%i) time.sleep(random.randint(1,5)) print(‘%s走出了房間‘%i) a.release() if __name__ == ‘__main__‘: a = Semaphore(5) for i in range(10): p = Process(target=f,args=(i,a)) p.start()信號量
事件：
用於主線程控制其他線程的執行，事件主要提供了三個方法 set、wait、clear。
事件處理的機制：全局定義了一個“Flag”，如果“Flag”值為 False，那麽當程序執行 event.wait 方法時就會阻塞，如果“Flag”值為True，那麽event.wait 方法時便不再阻塞。
clear：將“Flag”設置為False
set：將“Flag”設置為True
#!/usr/bin/env python # -*- coding:utf-8 -*- #Author: caoyf from multiprocessing import Event,Process import random import time def cars(a,i): if not a.is_set(): print(‘car%s在等待‘%i) a.wait() print(‘\033[31mcar%s通過\033[0m‘ % i) def f(a): while True: if a.is_set(): a.clear() print(‘\033[31m紅燈亮了\033[0m‘) else: a.set() print(‘\033[32m綠燈亮了\033[0m‘) time.sleep(2) if __name__ == ‘__main__‘: a = Event() p = Process(target=f,args=(a,)) p.start() for i in range(20): car = Process(target=cars,args=(a,i)) car.start() time.sleep(random.random())事件/紅綠燈實例
四、進程間通信---隊列和管道
隊列Queue：適用於多線程編程的先進先出數據結構，可以用來安全的傳遞多線程信息。
通過隊列實現了 主進程與子進程的通信 子進程與子進程之間的通信
q=Queue(10) #實例化一個對象，允許隊列對多10個元素
q.put() #放入隊列
q.get() #從隊列中取出
假設現在有一個隊伍，隊伍裏最多只能站5個人，但是有15個人想要進去
#!/usr/bin/env python # -*- coding:utf-8 -*- #Author: caoyf from multiprocessing import Process from multiprocessing import Queue def getin(q): #進入隊伍的子進程 for i in range(15): q.put(i) # print(q) def getout(q): #離開隊伍的子進程 for i in range(6): print(q.get()) if __name__==‘__main__‘: q=Queue(5) #隊伍內最多可以容納的人數 p=Process(target=getin,args=(q,)) #進入隊伍的進程 p.start() p2=Process(target=getout,args=(q,)) #離開隊伍的進程 p2.start()隊列實例
管道（Pipes）
#!/usr/bin/env python # -*- coding:utf-8 -*- #Author: caoyf from multiprocessing import Process,Pipe,Manager,Lock import time import random # 管道 進程之間創建的一條管道，默認是全雙工模式，兩頭都可以進和出， # 註意 必須在產生Process對象之前產生管道 # 如果在Pipe括號裏面填寫False後就變成了單雙工， # 左邊的只能收，右邊的只能發，recv（接收），send（發送） #如果沒有消息可以接收，recv會一直阻塞，如果連接的另外一段關閉後， #recv會拋出EOFError錯誤 # close 關閉連接 #下面的實例是在Pipe的括號裏填寫和不填寫False的區別 # from multiprocessing import Process,Pipe # def func(pro): # pro.send(‘hello‘) # pro.close() # # if __name__==‘__main__‘: # con,pro = Pipe(False) # p = Process(target=func,args=(pro,)) # p.start() # print(con.recv()) # p.join() # 模擬recv阻塞情況 # def func(con,pro): # con.close() # while 1: # try: # print(pro.recv()) # except EOFError: # pro.close() # break # # # if __name__==‘__main__‘: # con,pro = Pipe() # p = Process(target=func,args=(con,pro,)) # p.start() # pro.close() # con.send(‘aaaaa‘) # con.close() # p.join() # 利用管道實現生產者和消費者 # def sc(con,pro,name,food): # con.close() # for i in range(5): # time.sleep(random.random()) # f = ‘%s生產了%s%s‘%(name,food,i) # print(f) # pro.send(f) # def xf(con,pro,name): # pro.close() # while 1: # try: # baozi = con.recv() # print(‘%s消費了%s‘%(name,baozi)) # except EOFError: # break # if __name__==‘__main__‘: # con,pro = Pipe() # p1 = Process(target=sc,args=(con,pro,‘caoyf‘,‘包子‘)) # c1 = Process(target=xf,args=(con,pro,‘zhoaf‘)) # p1.start() # c1.start() # con.close() # pro.close() # p1.join()管道
數據共享：
隊列和管道只是實現了數據的傳遞，還沒有實現數據的共享，如實現數據共享，就要用到Managers（註：進程間通信應該盡量避免使用共享數據的方式）
from multiprocessing import Process,Manager
import os
def f(dict1,list1):
dict1[os.getpid()] = os.getpid() # 往字典裏放當前PID
list1.append(os.getpid()) # 往列表裏放當前PID
print(list1)
if __name__ == "__main__":
with Manager() as manager:
d = manager.dict() #生成一個字典，可在多個進程間共享和傳遞
l = manager.list(range(5)) #生成一個列表，可在多個進程間共享和傳遞
p_list = []
for i in range(10):
p = Process(target=f,args=(d,l))
p.start()
p_list.append(p) # 存進程列表
for res in p_list:
res.join()
print(‘\n%s‘ %d) #若要保證數據安全，需要加鎖lock=Lock()
進程池
對於需要使用幾個甚至十幾個進程時,我們使用Process還是比較方便的,但是如果要成百上千個進程,用Process顯然太笨了,multiprocessing提供了Pool類,即現在要講的進程池,能夠將眾多進程放在一起,設置一個運行進程上限,每次只運行設置的進程數,等有進程結束,再添加新的進程
- Pool(processes =num):設置運行進程數,當一個進程運行完,會添加新的進程進去
- apply_async：異步，串行
- apply：同步，並行
- close():關閉pool,不能再添加新的任務
import os import time import random from multiprocessing import Pool from multiprocessing import Process def func(i): i += 1 if __name__ == ‘__main__‘: p = Pool(5) # 創建了5個進程 start = time.time() p.map(func,range(1000)) p.close() # 是不允許再向進程池中添加任務 p.join() #阻塞等待 執行進程池中的所有任務直到執行結束 print(time.time() - start) start = time.time() l = [] for i in range(1000): p = Process(target=func,args=(i,)) # 創建了一百個進程 p.start() l.append(p) [i.join() for i in l] print(time.time() - start)
回調函數：
import os
import time
from multiprocessing import Pool
# 參數 概念 回調函數
def func(i): # 多進程中的io多，分出去一部分
print(‘子進程%s:%s‘%(i,os.getpid()))
return i*‘*‘
def call(arg): # 回調函數是在主進程中完成的，不能傳參數，只能接受多進程中函數的返回值
print(‘回調 ：‘,os.getpid())
print(arg)
if __name__ == ‘__main__‘:
print(‘主進程‘,os.getpid())
p = Pool(5)
for i in range(10):
p.apply_async(func,args=(i,),callback=call) #callback 回調函數 ：主進程執行 參數是子進程執行的函數的返回值
p.close()
p.join()
Python並發編程之進程