python多任務編程---進程

多任務編程

1. 意義:充分利用計算機CPU的多和資源,同時處理多個應用程序任務,一次提升程序的運行效率.
2. 實現方案:多進程,多線程

進程(process)

進程理論基礎

定義:程序在計算機中的一次運行.

• 程序是一個可執行的文件,是靜態的戰友磁盤.
• 進程是一個動態的過程描述,佔有計算機運行資源,有必定的生命週期

系統中如何產生一個進程

1. 用戶空間經過調用程序接口或者命令發起請求
2. 操做系統接受用戶請求,開始建立進程
3. 操做系統調配計算機資源,肯定進程狀態等
4. 操做系統將建立的進程提供給用戶使用

 

進程基本概念

• CPU時間片:若是一個進程佔有CPU內核則稱這個進程在cpu時間片上.
• PCB(進程控制塊):在內存中開闢的一塊空間,用於存放進程的基本信息,也用於系統查找識別進程
• 進程ID(PID):系統爲每一個進程分配的一個大於0的整數,做爲進程ID.每一個ID不重複

　　　　Linux查看進程ID:ps - aux

• 父子進程:系統中每個進程(除了系統初始化進程)都有惟一的父進程,能夠有0個或者多個子進程.父子進程關係便於進程管理

　　　　查看進程樹: pstree

• 進程狀態
• 三態

1. 就緒態:進程具有執行條件,等待分配CPU資源
2. 運行態:進程佔有CPU時間片正在運行
3. 等待態:進程暫時中止運行,讓出CPU

 五態(在三態基礎上增長新建和終止)

• 新建: 建立一個進程,獲取資源過程
• 終止:進程結束,釋放資源過程

狀態查看命令: ps - aux --->STAT列

S---等待態

R---執行態

Z---殭屍

+---前臺進程

l---有多線程

進程的運行特徵

1. 多進程能夠更充分使用計算機多核資源
2. 進程之間的運行互不影響,各自獨立
3. 每一個進程擁有獨立的空間,各自使用本身空間資源

面試要求:

　　什麼是進程,進程和程序有什麼區別?

　　進程有哪些狀態,狀態之間如何轉化?

基於fork的多進程編程

fork使用

 pdi = os.fork() 

功能:建立新的進程

返回值:整數,若是建立進程失敗返回一個負數,若是成功則在原有進程中返回新進程的PID,在新進程中返回0

注意:

• 子進程會複製父進程所有內存空間,從fork下一句開始執行
• 父子進程各自獨立運行,運行順序不必定
• 利用父子進程fork返回值的區別,配合if結構讓父子進程執行不一樣的內容幾乎是固定搭配.
• 父子進程有各自特有特徵好比PID,PCB命令集等.
• 父進程fork以前開闢的空間子進程一樣擁有,父子進程對各自空間的操做不會相互影響

fork進程代碼示例:

 1 import os
 2 from time import sleep
 3 
 4 # 建立子進程
 5 pid = os.fork()
 6 
 7 if pid < 0:
 8     print("Create process failed")
 9 elif pid == 0:
10     # 只有子進程執行
11     sleep(3)
12     print("The new process")
13 else:
14     # 只有父進程執行
15     sleep(4)
16     print("The old process")
17 
18 # 父子進程都執行
19 print("process test over")

View Code

 

fork進程代碼示例細節:

 1 import os
 2 from time import sleep
 3 
 4 print("=========================")
 5 a = 1
 6 def fun():
 7     print("fun .... ")
 8 
 9 pid = os.fork()
10 
11 if pid < 0:
12     print("Create process failed")
13 elif pid == 0:
14     print("Child process")
15     print("a = ",a)  # 從父進程空間拷貝了變量
16     fun()
17     a = 10000  # 只是修改了本身空間的a
18 else:
19     sleep(1)
20     print("Parent process")
21     print("a:",a)
22 
23 print("All a ->",a)

View Code

 

進程相關函數

 os.getpid() 

功能:獲取一個進程的PID值

返回值:返回當前進程的PID

 os.getppid() 

功能:獲取父進程的PID值

返回值:返回父進程PID

 os._exit(status) 

功能:結束一個進程

參數:進程的終止狀態

 sys.exit([status]) 

功能:退出進程

參數:整數　　表示退出狀態

　　字符串　　表示退出時打印內容

獲取進程pid號代碼示例:

 1 import os
 2 from time import sleep
 3 
 4 pid = os.fork()
 5 
 6 if pid < 0:
 7     print("Error")
 8 elif pid == 0:
 9     sleep(1)
10     print("Child PID:",os.getpid()) # 本身pid
11     print("Get parent PID:",os.getppid()) # 父pid
12 else:
13     print("Parent PID:", os.getpid())  # 本身pid
14     print("Get child PID:",pid)

View Code

 

進程退出代碼示例:

 1 import os,sys
 2 
 3 pid = os.fork()
 4 
 5 # 父子進程退出不會影響對方繼續執行
 6 if pid < 0:
 7     print("Error")
 8 elif pid == 0:
 9     # os._exit(0) # 子進程退出
10     print("Child process")
11 else:
12     sys.exit("退出父進程")
13     print("Parent process")

View Code

 

孤兒和殭屍

• 孤兒進程:父進程先於子進程退出,此時子進程成爲孤兒進程

　　特色:孤兒進程會被系統進程收養,此時系統進程就會成爲孤兒進程新的父進程,孤兒進程退出該進程會自動處理

• 殭屍進程:子進程先與父進程退出,父進程又沒有處理子進程的退出狀態,此時子進程就會稱爲殭屍進程

　　特色:殭屍進程雖然結束,可是會存留部分PCD在內存中,大量的殭屍進程會浪費系統的內存資源

• 如何避免殭屍進程產生
• 使用wait函數處理子進程退出

模擬殭屍進程產生以及處理示例:

 1 import os,sys
 2 import signal
 3 
 4 # 忽略子進程的退出行爲，子進程退出自動由系統處理
 5 signal.signal(signal.SIGCHLD,signal.SIG_IGN)
 6 
 7 pid = os.fork()
 8 if pid < 0:
 9     print("Error")
10 elif pid == 0:
11     print("Child PID:",os.getpid())
12     sys.exit(2)
13 else:
14     """
15     os.wait() 處理殭屍
16     """
17     # pid,status = os.wait()
18     # print("pid:",pid)
19     # print('status:',os.WEXITSTATUS(status))
20     while True: # 讓父進程不退出
21         pass

View Code

 

• 建立二級子進程處理殭屍

1. 父進程建立子進程,等待回收子進程
2. 子進程建立二級子進程而後退出
3. 二級子進程稱爲孤兒進程,和原來父進程一同執行事件

代碼示例:

 1 import os
 2 from time import sleep
 3 
 4 def f1():
 5     for i in range(3):
 6         sleep(2)
 7         print("寫代碼")
 8 
 9 def f2():
10     for i in range(2):
11         sleep(4)
12         print("測代碼")
13 
14 pid = os.fork()
15 if pid == 0:
16     p = os.fork()  # 建立二級子進程
17     if p == 0:
18         f1()
19     else:
20         os._exit(0)  # 一級子進程退出
21 else:
22     os.wait()  # 等待回收一級子進程
23     f2()

View Code

 

經過信號處理子進程退出

原理:子進程退出時會發送信號給父進程,若是父進程忽略子進程信號,則系統就會自動處理子進程退出

方法:使用signal模塊在父進程建立子進程前寫以下語句:

 import signal

signal.signal(signal.SIGCHLD,signal.SIG_IGN) 

特色:非阻塞,不會影響父進程運行,能夠處理全部子進程退出

multiprocessing模塊建立進程

進程建立方法

• 流程特色

1. 將須要子進程執行的事件封裝爲函數
2. 經過模塊的Process類建立進程對象,關聯函數
3. 能夠經過進程對象設置進程信息及屬性
4. 經過進程對象調用start啓動進程
5. 經過進程對象調用join回收進程

• 基本接口使用

Process()
功能 ： 建立進程對象
參數 ： target 綁定要執行的目標函數 
    args 元組，用於給target函數位置傳參
    kwargs 字典，給target函數鍵值傳參

p.start()
功能 ： 啓動進程

注意:啓動進程此時target綁定函數開始執行,該函數做爲子進程執行內容,此時進程真正被建立

p.join([timeout])
功能：阻塞等待回收進程
參數：超時時間

注意:

• 使用multiprocessing建立進程一樣是子進程複製父進程空間代碼段，父子進程運行互不影響。

• 子進程只運行target綁定的函數部分，其他內容均是父進程執行內容。

• multiprocessing中父進程每每只用來建立子進程回收子進程，具體事件由子進程完成。

• multiprocessing建立的子進程中沒法使用標準輸入

進程對象

p.name　　       進程名稱

p.pid　　　　   對應子進程的PID號

p.is_alive()　　查看子進程是否在生命週期

p.daemon　　  設置父子進程的退出關係

• 　　若是設置爲True則子進程會隨父進程的退出而結束
• 　　要求必須在start()前設置
• 　　若是daemon設置成True一般就不會使用join()

進程對象代碼示例:

 1 from multiprocessing import Process
 2 import time
 3 
 4 def tm():
 5     for i in range(3):
 6         print(time.ctime())
 7         time.sleep(2)
 8 
 9 p = Process(target = tm,name = 'Tarena')
10 
11 # 父進程退出，其全部子進程也退出
12 p.daemon = True
13 
14 p.start()  # 進程真正產生
15 
16 print("Name:",p.name)  # 進程名
17 print("PID：",p.pid) # pid號
18 print("is alive:",p.is_alive()) # 是否在生命週期

View Code

 

自定義進程類

• 建立步驟

1. 繼承Process類
2. 重寫__init__方法添加本身的屬性,使用super()加載父類屬性
3. 重寫run()方法

• 使用方法

1. 實例化對象
2. 調用start自動執行run方法
3. 調用join()回收進程

自定義進程類代碼示例:

 1 from multiprocessing import Process
 2 from time import *
 3 
 4 # 自定義進程類
 5 class MyProcess(Process):
 6     def __init__(self,value):
 7         self.value = value
 8         super().__init__() # 加載父類init
 9 
10     def f1(self):
11         print("步驟1")
12     def f2(self):
13         print("步驟2")
14 
15     # 做爲流程啓動函數
16     def run(self):
17         for i in range(self.value):
18             self.f1()
19             self.f2()
20 
21 if __name__ == '__main__':
22     p = MyProcess(2)
23     p.start()
24     p.join()

View Code

 

進程池實現

• 必要性

1. 進程的建立和銷燬過程消耗的資源較多
2. 當任務量衆多,每一個任務在很短期內完成時,須要頻繁的建立和銷燬進程.此時對計算機壓力較大
3. 進程池技術很好的解決了以上問題

• 原理

　　建立必定數量的進程來處理事件,事件處理完進程不退出而是繼續處理其餘時間,直到全部事件全都處理完畢統一銷燬.增長進程的重複利用,下降資源消耗.

• 進程池實現

1.建立進程池對象,放入適當的進程

from multiprocessing import Pool
Pool(processes)
功能： 建立進程池對象
參數： 指定進程數量，默認根據系統自動斷定

2.將事件加入進程池隊列執行

pool.apply_async(func,args,kwds)
功能: 使用進程池執行 func事件
參數： func 事件函數
      args 元組  給func按位置傳參
      kwds 字典  給func按照鍵值傳參
返回值： 返回函數事件對象

3.關閉進程池

pool.close()
功能： 關閉進程池

　　4.回收進程池中進程

pool.join()
功能： 回收進程池中進程

進程池代碼示例:

 1 from multiprocessing import Pool
 2 from time import sleep,ctime
 3 
 4 # 進程池事件
 5 def worker(msg):
 6     sleep(2)
 7     print(ctime(),'--',msg)
 8 
 9 # 建立進程池
10 pool = Pool(4)
11 
12 # 向進程池隊列中添加事件
13 for i in range(10):
14     msg = "Tedu %d"%i
15     pool.apply_async(func=worker,args=(msg,))
16 
17 # 關閉進程池
18 pool.close()
19 
20 # 回收進程池
21 pool.join()

View Code

 

管道通訊

• 通訊原理

在內存中開闢管道空間,生成管道操做對象,多個進程使用同一管道對象,進行讀寫便可實現通訊

• 實現方法

from  multiprocessing import Pipe

fd1,fd2 = Pipe(duplex = True) 功能: 建立管道 參數：默認表示雙向管道 若是爲False 表示單向管道 返回值：表示管道兩端的讀寫對象 若是是雙向管道都可讀寫 若是是單向管道fd1只讀 fd2只寫 fd.recv() 功能 ： 從管道獲取內容 返回值：獲取到的數據 fd.send(data) 能夠是任何內容,如:整數,浮點數,字符串等 功能： 向管道寫入內容 參數： 要寫入的數據

管道操做

注意:

1. 　　multiprocessing中提供的通訊只用於親緣關係進程間通訊
2. 　　管道在父進程中建立,子進程從父進程中獲取管道對象

管道操做代碼示例:

 1 from multiprocessing import Process,Pipe
 2 
 3 # 建立管道對象
 4 # 參數False 表示fd1 只能 recv , fd2 只能 send
 5 fd1,fd2 = Pipe()
 6 
 7 # APP1可使用app2提供的信息登陸
 8 def app1():
 9     print("啓動app1，請登陸")
10     print("請求app2受權")
11     # 寫管道
12     fd1.send("app1 能夠用你的帳號登陸嗎？")
13     data = fd1.recv()
14     if data:
15         print("登陸成功：",data)
16 
17 def app2():
18     request = fd2.recv() # 阻塞等待讀取管道
19     print(request)
20     fd2.send(('Joy','123')) # 發送python數據類型
21 
22 p1 = Process(target=app1)
23 p2 = Process(target=app2)
24 p1.start()
25 p2.start()
26 p1.join()
27 p2.join()
28 
29 代碼示例

View Code

 

消息隊列

• 通訊原理

　　在內存中創建隊列模型,進程經過隊列將消息存入,或者從隊列取出完成進程間通訊.

• 實現方法

from multiprocessing import Queue

q = Queue(maxsize=0)
功能: 建立隊列對象
參數：最多存放消息個數
返回值：隊列對象

q.put(data,[block,timeout])
功能：向隊列存入消息
參數：data  要存入的內容
block  設置是否阻塞 False爲非阻塞
timeout  超時檢測

q.get([block,timeout])
功能：從隊列取出消息
參數：block  設置是否阻塞 False爲非阻塞
timeout  超時檢測
返回值： 返回獲取到的內容

q.full()   判斷隊列是否爲滿
q.empty()  判斷隊列是否爲空
q.qsize()  獲取隊列中消息個數
q.close()  關閉隊列

消息隊列代碼示例:

 1 from multiprocessing import Queue,Process
 2 from time import sleep
 3 from random import randint
 4 
 5 # 建立隊列
 6 q = Queue(5) # 最大存儲5個消息
 7 
 8 def request():
 9     for i in range(10):
10         x = randint(1,100)
11         y = randint(1,100)
12         q.put((x,y))  # 寫消息隊列
13         print("=============")
14 
15 def handle():
16     while not q.empty():
17         data = q.get() # 讀消息隊列
18         print("x + y = ",data[0]+data[1])
19         sleep(2)
20 
21 p1 = Process(target=request)
22 p2 = Process(target=handle)
23 p1.start()
24 p2.start()
25 p1.join()
26 p2.join()
27 
28 代碼示例

View Code

 

 共享內存

• 通訊原理:在內存中開闢一塊空間,進程能夠寫入內容和讀取內容完成通訊,可是每次寫入內容會覆蓋以前內容
• 實現方法

from multiprocessing import Value,Array

obj = Value(ctype,data)
功能 ： 開闢共享內存
參數 ： ctype  表示共享內存空間類型 'i'  'f'  'c'
       data   共享內存空間初始數據
返回值：共享內存對象

obj.value  對該屬性的修改查看即對共享內存讀寫


obj = Array(ctype,data)
功能： 開闢共享內存空間
參數： ctype  表示共享內存數據類型
      data   整數則表示開闢空間的大小，其餘數據類型表示開闢空間存放的初始化數據
返回值：共享內存對象

Array共享內存讀寫： 經過遍歷obj能夠獲得每一個值，直接能夠經過索引序號修改任意值。

* 可使用obj.value直接打印共享內存中的字節串

共享內存代碼示例:

注意:共享內存中只能有一個值

 1 from multiprocessing import Process,Value
 2 import time
 3 import random
 4 
 5 # 建立共享內存
 6 money = Value('i',5000)
 7 
 8 def man():
 9     for i in range(30):
10         time.sleep(0.2)
11         # 修改共享內存
12         money.value += random.randint(1,1000)
13 
14 def girl():
15     for i in range(30):
16         time.sleep(0.15)
17         money.value -= random.randint(100,800)
18 
19 p1 = Process(target=man)
20 p2 = Process(target=girl)
21 p1.start()
22 p2.start()
23 p1.join()
24 p2.join()
25 print("一個月餘額：",money.value) #讀取共享內存

View Code

 

 共享內存存放列表,字節串:

 1 from multiprocessing import Process,Array
 2 
 3 # 建立共享內存，初始數據 [1,2,3,4]
 4 # shm = Array('i',[1,2,3,4])
 5 # shm = Array('i',4) # 開闢4個整形的列表空間
 6 shm = Array('c',b'hello')
 7 
 8 def fun():
 9     # 共享內存對象能夠迭代
10     for i in shm:
11         print(i)
12     shm[0] = b'H' # 修改共享內存
13 
14 p = Process(target=fun)
15 p.start()
16 p.join()
17 for i in shm:
18     print(i)
19 print(shm.value) # 總體打印字節串

View Code

 

信號量(信號燈集)

• 通訊原理

　　給定一個數量對多個進程可見.多個進程均可以操做改數量增減,並根據數量值決定本身的行爲

• 實現方法

from multiprocessing import Semaphore

sem = Semaphore(num)
功能 ： 建立信號量對象
參數 ： 信號量的初始值
返回值 ： 信號量對象

sem.acquire()  將信號量減1 當信號量爲0時阻塞
sem.release()  將信號量加1
sem.get_value() 獲取信號量數量

信號量代碼示例:

注意: 信號量能夠當作是一種資源，執行任務須要消耗信號量資源，
這樣能夠控制進程執行行爲

 1 from multiprocessing import Process,Semaphore
 2 from time import sleep
 3 import os
 4 
 5 # 建立信號量資源
 6 sem = Semaphore(3)
 7 
 8 # 任務函數 (系統中最多可以同時運行三個該任務)
 9 def handle():
10     sem.acquire() # 消耗一個信號量
11     print("%s執行任務"%os.getpid())
12     sleep(2)
13     print("%s 拯救了宇宙"%os.getpid())
14     sem.release() # 增長一個信號量
15 
16 jobs = []
17 for i in range(10):
18     p = Process(target = handle)
19     jobs.append(p)
20     p.start()
21 
22 for i in jobs:
23     i.join()

View Code