python 全棧開發，Day38(在python程序中的進程操作,multiprocess.Process模塊)

昨日內容回顧

操作系統
紙帶打孔計算機
批處理 —— 磁帶
　　聯機
　　脫機
多道操作系統 —— 極大的提高了CPU的利用率
　　在計算機中 可以有超過一個進程
　　進程遇到IO的時候 切換給另外的進程使用CPU
　　數據隔離 進程與進程之間的數據是隔離的
　　時空復用 在同一時刻 多個程序宏觀上的並行
分時系統 —— 反而降低了CPU的效率 提高了用戶體驗
　　將時間分片 每一個進程都能夠使用CPU一個時間片的時間
　　時間片輪轉 一個進程在使用cpu的時候時間片到了，
　　　　就會切換到另一個進程
實時系統 —— 實時的響應任務
　　有一個進程時時刻刻能使用CPU
個人計算機 ： 多道操作系統 分時系統 實時系統
4核 —— 4個任務
4核8線程 —— 8個任務
進程
　　進程的三個狀態 就緒 阻塞 運行
　　進程的創建和結束
　　同步 異步 阻塞 非阻塞
　　兩個任務依次執行 —— 同步
　　兩個任務一起執行 —— 異步
　　同步 + 阻塞
　　　　inp = input()
　　　　print(inp)
異步 + 非阻塞
　　a.py input() 異步阻塞
　　b.py print() 異步非阻塞

 

在python代碼中，是從上向下執行的。它是屬於同步的
同步往往是 同步+阻塞

同步總是伴隨着阻塞
比如上面的例子，input都得等着用戶輸入

 

異步往往和非阻塞在一起

比如a.py和b.py，2個py在不同的文件中，執行互相不影響

當a.py和b.py同時執行時，就是異步非阻塞

單純的一個py文件，沒有異步非阻塞。異步非阻塞存在多個文件執行時才存在。
比如上面的例子

 

在python程序中的進程操作

　　之前我們已經了解了很多進程相關的理論知識，了解進程是什么應該不再困難了，剛剛我們已經了解了，運行中的程序就是一個進程。所有的進程都是通過它的父進程來創建的。因此，運行起來的python程序也是一個進程，那么我們也可以在程序中再創建進程。多個進程可以實現並發效果，也就是說，當我們的程序中存在多個進程的時候，在某些時候，就會讓程序的執行速度變快。以我們之前所學的知識，並不能實現創建進程這個功能，所以我們就需要借助python中強大的模塊。

multiprocess模塊

      仔細說來，multiprocess不是一個模塊而是python中一個操作、管理進程的包。 之所以叫multi是取自multiple的多功能的意思,在這個包中幾乎包含了和進程有關的所有子模塊。由於提供的子模塊非常多，為了方便大家歸類記憶，我將這部分大致分為四個部分：創建進程部分，進程同步部分，進程池部分，進程之間數據共享。

 

multiprocess.process模塊

process模塊介紹

process模塊是一個創建進程的模塊，借助這個模塊，就可以完成進程的創建。

Process([group [, target [, name [, args [, kwargs]]]]])，由該類實例化得到的對象，表示一個子進程中的任務（尚未啟動）

強調：
1. 需要使用關鍵字的方式來指定參數
2. args指定的為傳給target函數的位置參數，是一個元組形式，必須有逗號

參數介紹：
1 group參數未使用，值始終為None
2 target表示調用對象，即子進程要執行的任務
3 args表示調用對象的位置參數元組，args=(1,2,'egon',)
4 kwargs表示調用對象的字典,kwargs={'name':'egon','age':18}
5 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開啟的進程  

p.daemon：默認值為False，如果設為True，代表p為后台運行的守護進程，當p的父進程終止時，p也隨之終止，並且設定為True后，p不能創建自己的新進程，必須在p.start()之前設置
p.name:進程的名稱
p.pid：進程的pid
p.exitcode:進程在運行時為None、如果為–N，表示被信號N結束(了解即可)
p.authkey:進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是為涉及網絡連接的底層進程間通信提供安全性，這類連接只有在具有相同的身份驗證鍵時才能成功（了解即可）

在Windows操作系統中由於沒有fork(linux操作系統中創建進程的機制)，在創建子進程的時候會自動 import 啟動它的這個文件，而在 import 的時候又執行了整個文件。因此如果將process()直接寫在文件中就會無限遞歸創建子進程報錯。所以必須把創建子進程的部分使用if __name__ ==‘__main__’ 判斷保護起來，import 的時候  ，就不會遞歸運行了。

 

每一個進程，都有一個進程id號
查看進程號

import os
import time
print(os.getppid())
time.sleep(1000)  # 方便查看進程

執行輸出：12492

打開windows的任務管理器，找到pid為12492進程，它是一個Pycharm進程

系統中的進程id號，是不會沖突的。
每一個進程，對應一個唯一的進程id號

 

假設這台電腦是4核，它可以開4個進程，每一個進程對應一個核。

但是上面的代碼，在同一時間，只能用一個核。如果想充分利用CPU，需要開4個線程才行。

多進程的作用，就是為了提高效率

 

使用process模塊創建進程

import os
import time
from multiprocessing import Process
def process1():
    print('process1:',os.getppid())
    time.sleep(1)

print(os.getppid())  # 獲取當前進程id
process1()  # 執行函數

執行輸出：

12492
process1: 12492

進程id全都是一樣的。程序一旦執行，進程id不會變。

 使用target調用對象，即子進程要執行的任務

import os
import time
from multiprocessing import Process
def process1():
    print('process1:',os.getppid())
    time.sleep(1)

print(os.getppid())  # 獲取當前進程id
p = Process(target=process1)  # 調用函數
p.start()  # 啟動進程

執行報錯：

在mac和linux執行不會報錯
而windows會報錯，why ?

 

在Windows操作系統中由於沒有fork(linux操作系統中創建進程的機制)，在創建子進程的時候會自動 import 啟動它的這
個文件，而在 import 的時候又執行了整個文件。因此如果將process()直接寫在文件中就會無限遞歸創建子進程報錯。
所以必須把創建子進程的部分使用if __name__ ==‘__main__’ 判斷保護起來，import 的時候 ，就不會遞歸運行了

linux 執行函數，會把代碼copy一份，到一個空間里面
那么下一次執行時，直接調用，就可以了。

解決方案：

import os
import time
from multiprocessing import Process
def process1():
    print('process1:',os.getppid())
    time.sleep(1)

print(os.getppid())  # 獲取當前進程id
if __name__ == '__main__':
    p = Process(target=process1)  # 調用函數
    p.start()  # 啟動進程

執行輸出：

12492
16876
process1: 16876

16876為父進程

 

import os
import time
from multiprocessing import Process
def process1():
    print('process1:',os.getppid())
    time.sleep(1)


if __name__ == '__main__':
    print(os.getppid())  # 獲取當前進程id
    p = Process(target=process1)  # 調用函數
    p.start()  # 啟動進程

執行輸出：

12492
process1: 15624

執行結果，表示在當前進程中，開啟了一個新的進程

 

使用args關鍵字給函數傳參

import os
import time
from multiprocessing import Process
def process1(n,name):
    print('process1:',os.getppid())
    print('n:',n,name)
    time.sleep(1)


if __name__ == '__main__':
    print(os.getppid())  # 獲取當前進程id
    p = Process(target=process1,args=[1,'alex'])  # 調用函數並傳參
    p.start()  # 啟動進程

執行輸出：

12492
process1: 16676
n: 1 alex

 

那么函數如果加return，可以獲取結果嗎？等學到后面，就可以獲取進程的返回值了。稍安勿躁！

 

進程與子進程

import os
print(os.getpid())  # 當前進程
print(os.getppid())  # 父進程

執行輸出：

14920
12492

12492就是Pycharm進程，看下圖

多執行幾次，當前進程會一直變，但是父進程卻不會變。除非你把Pycharm關了。

 

import os
import time
from multiprocessing import Process
print('1',os.getpid())  # 當前進程
time.sleep(1)
print('2',os.getppid())  # 父進程
time.sleep(1)
def func():
    print('3',os.getpid(),os.getppid())
    time.sleep(1)

if __name__ == '__main__':
    #p = Process(target=func)
    #p.start()
    # 上面2句等於下面這句
    Process(target=func).start()

執行輸出：

1 14512
2 12492
1 17048
2 14512
3 17048 14512

 

為啥會輸出5個結果呢？應該是3個才對呀？

在Windows操作系統中由於沒有fork(linux操作系統中創建進程的機制)，在創建子進程的時候會自動 import 啟動它的這個文件，而在 import 的時候又執行了整個文件。

所以輸出5個結果

 

整個py文件運行之后，就是主進程

總結：

主進程默認會等待子進程執行完畢之后才結束

 

看下面的代碼，print會立刻執行嗎？

import os
import time
from multiprocessing import Process

def func():
    print('func',os.getpid(),os.getppid())
    time.sleep(1)

if __name__ == '__main__':
    print(os.getpid(), os.getppid())
    Process(target=func).start()
    print('*'*20)

執行輸出：

8652 12492
********************
func 16928 8652

立馬就打印了****，這是為什么呢？

它是一個異步程序，主程序執行時，它不會等待1秒(子進程)
但是執行結束的動作時，它會等待子進程結束，除非異常退出。
否則子進程會變成一個僵屍進程。自己創建的，需要自己銷毀。

總結：

主進程默認會等待子進程執行完畢之后才結束
主進程和子進程之間的代碼是異步的
為什么主進程要等待子進程結束 回收一些子進程的資源
開啟一個進程是有時間開銷的 ：操作系統響應開啟進程指令，給這個進程分配必要的資源

 

import os
import time
from multiprocessing import Process

def func():
    print('func',os.getpid(),os.getppid())
    time.sleep(1)

if __name__ == '__main__':
    print(os.getpid(), os.getppid())
    Process(target=func).start()
    print('*'*20)
    print('*' * 30)
    time.sleep(0.5)
    print('*' * 40)

執行輸出：

10424 12492
********************
******************************
func 4564 10424
****************************************

 

同步控制

import os
import time
from multiprocessing import Process

def func():
    print('func',os.getpid(),os.getppid())
    time.sleep(1)

if __name__ == '__main__':
    print(os.getpid(), os.getppid())
    p = Process(target=func)
    p.start()
    p.join() #主線程等待p終止（強調：是主線程處於等的狀態，而p是處於運行的狀態）
    print('*'*20)  # 等待1秒后執行

執行輸出：

18328 12492
func 18376 18328
********************

上面的代碼沒啥意義，它和面向過程執行一個函數，沒啥區別。

 

舉一個實際的例子

讓子進程計算一個值，主進程必須等到子進程計算完之后，根據計算的值，來進行下一步計算

以文件為消息中間件，來完成主進程獲取子進程的值，從而計算最終結果

 

import os
import time
from multiprocessing import Process

def func(exp):
    print('func',os.getpid(),os.getppid())
    result = eval(exp)
    with open('file','w') as f:
        f.write(str(result))  # 寫入的內容必須是一個字符串

if __name__ == '__main__':
    print(os.getpid(), os.getppid())
    p = Process(target=func,args=['3*5'])
    p.start()
    ret = 5/6
    p.join() #主線程等待子進程計算完
    with open('file') as f:
        result = f.read()  # 讀取結果
    ret = ret + int(result)  # 最終計算結果
    print(ret)

執行輸出：

16784 12492
func 17304 16784
15.833333333333334

 

一般情況下，是主進程，開啟子進程
子進程開啟子進程的情況很少。

 

開啟多個子進程

import os
import time
from multiprocessing import Process

def func(n):
    print(n,os.getpid(),os.getppid())
    #print(n)

if __name__ == '__main__':
    Process(target=func,args=[1]).start()
    Process(target=func, args=[2]).start()

執行輸出：

1 14648 9832
2 17024 9832

 

使用for循環開啟10個進程

import os
import time
from multiprocessing import Process

def func(n):
    print(n,os.getpid(),os.getppid())
    #rint(n)

if __name__ == '__main__':
    for i in range(10):
        Process(target=func,args=[i]).start()

執行輸出：

1 14048 16976
2 16876 16976
0 10944 16976
3 18276 16976
6 9484 16976
4 17092 16976
5 8652 16976
8 15624 16976
7 17832 16976
9 7012 16976

這是為什么是隨機的？而不是順序的？

多個進程同時運行（注意，子進程的執行順序不是根據啟動順序決定的）
是操作系統來決定的。它不一定是按照你的順序來開啟進程的。
它有自己的算法，比如開啟一個進程，時間片輪轉了。那么就不是順序的。

 

再來一個高大上的例子
假如有10個表達式

import os
import time
from multiprocessing import Process

def func(n):
    print(n,os.getpid(),os.getppid())
    #rint(n)

if __name__ == '__main__':
    for i in range(10):
        p = Process(target=func,args=[i])
        p.start()
    print('求和')

但是這樣，是沒有意義的。和面向過程沒啥區別了

 

高級的辦法，使用append

import os
import time
from multiprocessing import Process

def process(n):
    print(os.getpid(),os.getppid())
    time.sleep(1)
    print(n)

if __name__ == '__main__':
    p_lst = []  # 定義一個列表
    for i in range(10):
        p = Process(target=process,args=[i,])
        p.start()
        p_lst.append(p)  # 將所有進程寫入列表中
    for p in p_lst:p.join()  # 檢測p是否結束,如果沒有結束就阻塞直到結束,如果已經結束了就不阻塞
    print('求和')　

它是將所有進程放入一個列表里面，那么當列表里面的每一個進程都執行完成之后，
執行最后一個求和結果
這里用到了 異步+同步。每一個子進程執行，屬於異步。最終計算結果時，屬於同步，因為它要等待所有子進程結束。

執行輸出：

17180 16032
10080 16032
17992 16032
18112 16032
18064 16032
17872 16032
13440 16032
8704 16032
15536 16032
14032 16032
0
2
3
4
6
1
7
5
8
9
求和

多執行幾次，求和是最后一個輸出的。

 

 開啟進程的第二種方式

上面講的子進程開啟方式是：

p = Process(target=process,args=[1])
        p.start()

第二種方式，是通過繼承來實現的
必須要重寫run方法，名字必須是run
看源碼

from multiprocessing import Process

女主播的例子：

import os
from multiprocessing import Process
class Myprocess(Process):
    def run(self):
        print(os.getpid())
        print('和女主播聊天')

if __name__ == '__main__':
    print(os.getpid())
    p = Myprocess()
    p.start()  # 在執行start的時候,會自動幫我們主動執行run方法

執行輸出：

11604
13764
和女主播聊天

 

接收一個參數

那誰和女主播聊天呢？這個時候，需要參數。怎么傳呢？使用__init__方法

import os
from multiprocessing import Process
class Myprocess(Process):
    def __init__(self,name):
        super().__init__()
        self.name = name
    def run(self):
        print(os.getpid())
        print('%s和女主播聊天'%self.name)

if __name__ == '__main__':
    print(os.getpid())
    p = Myprocess('alex')
    p.start()  # 在執行start的時候,會自動幫我們主動執行run方法

執行輸出：

10464
17860
alex和女主播聊天

 

接收多個參數

import os
from multiprocessing import Process
class Myprocess(Process):
    def __init__(self,*args):
        super().__init__()
        self.args = args
    def run(self):
        print(os.getpid())
        for i in self.args:
            print('%s和女主播聊天'%i)

if __name__ == '__main__':
    print(os.getpid())
    p = Myprocess('alex','taibai')
    p.start()  # 在執行start的時候,會自動幫我們主動執行run方法

執行輸出：

17996
17532
alex和女主播聊天
taibai和女主播聊天

self.args 是一個元組類型，因為傳了多個參數

 

打印進程名

import os
from multiprocessing import Process
class Myprocess(Process):
    def __init__(self,*args):
        super().__init__()
        self.args = args
    def run(self):
        print(os.getpid(), self.name, self.pid)  # 打印進程id,進程名,也是進程id
        for i in self.args:
            print('%s和女主播聊天'%i)

if __name__ == '__main__':
    print(os.getpid())
    p = Myprocess('alex','taibai')
    p.start()  # 在執行start的時候,會自動幫我們主動執行run方法

執行輸出：

15776
2628 Myprocess-1 2628
alex和女主播聊天
taibai和女主播聊天

 

注意：開啟多個進程時，必須要寫

if __name__ == '__main__':

這2種啟動方式，喜歡哪種方式，就可以使用哪種方式
至少會一個就可以了。

 

進程中的數據隔離

如何證明是隔離的呢？

from multiprocessing import Process
n = 100  # 全局變量
def func():
    global n
    n += 1  # 修改全局變量
    print('son : ',n)


if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    p.join()  # 等待子進程結束
    print(n)  # 打印全局變量

執行輸出：

son : 101
100

子進程的變量不會影響主進程的變量

 

守護進程

守護進程會隨着主進程的結束而結束。

主進程創建守護進程
　　其一：守護進程會在主進程代碼執行結束后就終止
　　其二：守護進程內無法再開啟子進程,否則拋出異常：AssertionError: daemonic processes are not allowed to have children

注意：進程之間是互相獨立的，主進程代碼運行結束，守護進程隨即終止

import time
from multiprocessing import Process

def func():
    print('son start')
    time.sleep(1)
    print('son end')

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(p.name) # 打印進程名
    print('在主進程中')  # 主進程會等待子進程結束而結束

執行輸出：

Process-1
在主進程中
son start
son end

 

注意： 

一定要在p.start()前設置,設置p為守護進程,禁止p創建子進程,並且父進程代碼執行結束,p即終止運行

 

現在設置一個守護進程

import time
from multiprocessing import Process

def func():
    print('son start')
    time.sleep(1)
    print('son end')

if __name__ == '__main__':
    p = Process(target=func)
    #在一個進程開啟之前可以設置它為一個守護進程
    p.daemon = True
    p.start()
    print('main------')

執行輸出： 在主進程中

為什么只輸出了一句話，子進程的輸出呢？

打印該行則主進程代碼結束,則守護進程Process-1應該被終止.
可能會有Process-1任務執行的打印信息son start,
因為主進程打印main----時,Process-1也執行了,但是隨即被終止.

 

總結：

守護進程的意義：
　　子進程會隨着主進程代碼的執行結束而結束
　　注意：守護進程不會關系主進程什么時候結束，我只關心主進程中的代碼什么時候結束
守護進程的作用：
　　守護主進程，程序報活
　　主進程開啟的時候 建立一個守護進程
　　守護進程只負責每隔1分鍾 就給檢測程序發一條消息

 

如果沒有守護，使用代碼實現：

import time
while True:
    time.sleep(60)
    sk.send('我還活着')

但是這樣，就無法做其他事情了

 

看下面的代碼，它的執行效果是什么?

import time
from multiprocessing import Process

def func():
    print('son start')
    while True:
        time.sleep(1)
        print('son end')

def fun2():
    print('start :in fun2')
    time.sleep(5)
    print('end : in fun2')

if __name__ == '__main__':
    p = Process(target=func)
    #在一個進程開啟之前可以設置它為一個守護進程
    p.daemon = True
    p.start() # 異步執行
    print(p.name)
    p2 = Process(target=fun2)
    p2.start() # 異步執行
    print(p2.name)
    time.sleep(2)
    print('在主進程中')

執行輸出：

Process-1
Process-2
son start
start :in fun2
son end
在主進程中
end : in fun2

 

分析：
　　主進程的代碼 大概在2s多的時候就結束了
　　p2子進程實在5s多的時候結束
　　主進程結束
　　p是在什么時候結束的？
　　p是在主進程的代碼執行完畢之后就結束了

 

總結：

主進程會等待子進程的結束而結束
守護進程的意義：
　　子進程會隨着主進程代碼的執行結束而結束
　　注意：守護進程不會關系主進程什么時候結束，我只關心主進程中的代碼什么時候結束

守護進程的作用：
　　守護主進程，程序報活
　　主進程開啟的時候 建立一個守護進程
　　守護進程只負責每隔1分鍾 就給檢測程序發一條消息

 

進程中的其他方法：

顯示進程名和id

from multiprocessing import Process

def func():
    print('wahaha')

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(p.pid)  # 進程id
    print(p.name)  # 進程名

執行輸出：

18256
Process-1
wahaha

 

檢測當前進程是否活着

import time
from multiprocessing import Process

def func():
    print('wahaha')
    time.sleep(3)
    print('wahaha end')

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(p.is_alive())  # 是否活着,返回bool值
    time.sleep(3)

執行輸出：

True
wahaha
wahaha end

 

在主進程中結束一個子進程

import time
from multiprocessing import Process

def func():
    print('wahaha')
    time.sleep(5)
    print('wahaha end')

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(p.is_alive())  # 是否活着,返回bool值
    time.sleep(3)
    p.terminate()  # 在主進程中結束一個子進程

執行輸出：

True
wahaha

 

import time
from multiprocessing import Process

def func():
    print('wahaha')
    time.sleep(5)
    print('wahaha end')

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(p.is_alive())  # 是否活着,返回bool值
    time.sleep(1)
    p.terminate()  # 在主進程中結束一個子進程
    print(p.is_alive())

執行輸出：

True
wahaha
True

為啥最后會輸出True呢？因為terminate是告知操作系統，要關閉這個進程。但是操作系統不會立即執行，它還有別的任務執行。那么執行is_alive時，結果仍然為True

 

import time
from multiprocessing import Process

def func():
    print('wahaha')
    time.sleep(5)
    print('wahaha end')

if __name__ == '__main__':
    p = Process(target=func)
    p.start()
    print(p.is_alive())  # 是否活着,返回bool值
    time.sleep(1)
    p.terminate()  # 在主進程中結束一個子進程
    time.sleep(0.5)  #等待0.5秒
    print(p.is_alive())

執行輸出：

True
wahaha
False

 

由於停頓了0.5秒，操作系統有反應的時間，所以能結束掉子進程。

 

今日作業：

socket聊天並發實例，使用原生socket的TCP協議，實現一個聊天的並發實例

server.py

import socket
from multiprocessing import Process

server=socket.socket()
server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1)  # 允許端口復用
server.bind(('127.0.0.1',9000))
server.listen(5)  #開始監聽TCP傳入連接,表示允許最大5個連接數

def talk(conn,client_addr):  # 說話
    while True:
        try:
            msg=conn.recv(1024)  # 接收客戶端信息
            if not msg:break
            conn.send(msg.upper())  # 發送信息,將內容轉換為大寫
        except Exception:
            break

if __name__ == '__main__': #windows下start進程一定要寫到這下面
    while True:
        conn,client_addr=server.accept()  # 等待接受客戶端連接
        p=Process(target=talk,args=(conn,client_addr))  # 調用函數
        p.start()  # 啟動進程

client.py

import socket

client=socket.socket()
client.connect(('127.0.0.1',9000))

while True:
    msg=input('>>: ').strip()
    if not msg:continue

    client.send(msg.encode('utf-8'))
    msg=client.recv(1024)
    print(msg.decode('utf-8'))

先執行server.py，再執行client.py

client輸出:

 

 

明日默寫：

import os
from multiprocessing import Process

def func(exp):
    print(os.getpid(),os.getppid())
    result = eval(exp)
    with open('file','w') as f:
        f.write(str(result))

if __name__ == '__main__':
    print(os.getpid(),os.getppid())  # process id,parent process id
    # 3*5+5/6
    p = Process(target=func,args=['3*5'])    # func
    p.start()
    ret = 5/6
    p.join()  # join方法能夠檢測到p進程是否已經執行完了，阻塞知道p執行結束
    with open('file') as f:
        result = f.read()
    ret = ret + int(result)
    print(ret)