python網(wǎng)絡(luò)編程之進(jìn)程詳解

更新時間：2022年01月09日 17:01:27 作者：Top Secret

這篇文章主要為大家介紹了python網(wǎng)絡(luò)編程之進(jìn)程，具有一定的參考價值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助

1.進(jìn)程

它們的主要作用：多任務(wù)同時執(zhí)行

1.1進(jìn)程：

Windows打開的程序就是一個進(jìn)程例如打開qq 打開微信

如果打開2個qq代表打開了2個進(jìn)程

1.2在python中創(chuàng)建進(jìn)程

只能Linux 使用os.fork()用這個可以創(chuàng)建多進(jìn)程

Linux/Windows使用multiprocessing模塊和Pool進(jìn)程池（他倆是跨平臺模塊）

1.3 使用multiprocessing創(chuàng)建進(jìn)程

1.3.1 單個進(jìn)程時：

from multiprocessing import Process #從multiprocessing庫中導(dǎo)入Process模塊
#執(zhí)行進(jìn)程代碼
#一個子進(jìn)程
def test(interval):
    print("我是一個子進(jìn)程")
 
#執(zhí)行主程序
def main():
    print("主進(jìn)程啟動")
    p = Process(target=test,args=(1,)) #使用進(jìn)程模塊，目標(biāo)參數(shù)target為子進(jìn)程函數(shù)
    p.start() #啟動子進(jìn)程
    print("主進(jìn)程結(jié)束")
 
if __name__ == '__main__': 
    main()
"""
>>> runfile('D:/python_files/python_fiew/網(wǎng)絡(luò)編程_01.py', wdir='D:/python_files/python_fiew')
主進(jìn)程啟動
主進(jìn)程結(jié)束
我是一個子進(jìn)程
"""

1.3.2 多個子進(jìn)程時：

from multiprocessing import Process
import time,os
#子進(jìn)程1
def child_1(interval):
    print("子進(jìn)程(%s)開始執(zhí)行，父進(jìn)程為(%s)"%(os.getpid(),os.getpid())) #os.getpid()為獲取進(jìn)程號
    t_stat = time.time() #計時開始
    time.sleep(interval) #程序?qū)⒈粧炱餹秒
    t_end = time.time() #計時結(jié)束
    print("子進(jìn)程(%s)的執(zhí)行時間為'%0.2f'秒"%(os.getpid(),t_end-t_stat))
#子進(jìn)程2
def chlid_2(interval):
    print("子進(jìn)程(%s)開始執(zhí)行，父進(jìn)程為(%s)"%(os.getpid(),os.getpid()))
    t_stat = time.time()
    time.sleep(interval)
    t_end = time.time()
    print("子進(jìn)程(%s)的執(zhí)行時間為(%0.2f)"%(os.getpid(),t_end-t_stat))
 
if __name__ == '__main__':
    print("-----父進(jìn)程開始執(zhí)行-------")
    print("父進(jìn)程啟動時，父進(jìn)程PID為：(%s)"%(os.getpid()))  #打印父進(jìn)程啟動時父進(jìn)程的PID
    p1 = Process(target=child_1,args=(1,)) #實例化子進(jìn)程1
    p2 = Process(target=chlid_2,args=(1,)) #實例化子進(jìn)程2
    p1.start() #啟動子進(jìn)程1
    p2.start() #啟動子進(jìn)程2
    #此時父進(jìn)程仍然在執(zhí)行
    print("pi.is_alive=%s"%(p1.is_alive()))  #p1.is_alive() 判斷子進(jìn)程p1是否還在執(zhí)行，執(zhí)行則返回ture
    print("p2_is_alive=%s"%(p2.is_alive()))
    #輸出此時進(jìn)程執(zhí)行過程中的PID(進(jìn)程號)
    print("執(zhí)行過程中的p1的進(jìn)程號p1.pid=%s"%(p1.pid))
    print("執(zhí)行過程中的p2的進(jìn)程號p2.pid=%s" % (p2.pid))
    print("---等待子進(jìn)程結(jié)束----")
    p1.join() #等待子程序p1結(jié)束
    p2.join()
    print("父進(jìn)程結(jié)束，此時父進(jìn)程的進(jìn)程號為：(%s)"%(os.getpid()))
#總結(jié)：#os.getpid()為獲取進(jìn)程號   #p1.is_alive() 判斷子進(jìn)程p1是否還在執(zhí)行，執(zhí)行則返回ture

由運(yùn)行結(jié)果分析：子進(jìn)程在執(zhí)行的過程中，進(jìn)程號不變。父進(jìn)程號在啟動時，和在結(jié)束所有子進(jìn)程時相同，而當(dāng)子進(jìn)程同時執(zhí)行時，父進(jìn)程號會動態(tài)變化。

掛起進(jìn)程在操作系統(tǒng)中可以定義為暫時被淘汰出內(nèi)存的進(jìn)程，機(jī)器的資源是有限的，在資源不足的情況下，操作系統(tǒng)對在內(nèi)存中的程序進(jìn)行合理的安排，其中有的進(jìn)程被暫時調(diào)離出內(nèi)存，當(dāng)條件允許的時候，會被操作系統(tǒng)再次調(diào)回內(nèi)存，重新進(jìn)入等待被執(zhí)行的狀態(tài)即就緒態(tài)，系統(tǒng)在超過一定的時間沒有任何動作。

1.3.3 自定義進(jìn)程類方法

使用Process子類創(chuàng)建進(jìn)程Process(target=test)實現(xiàn)多進(jìn)程，復(fù)雜的要定義一個類繼承Process，每次實例化這個類的時候就等同于實例化一個進(jìn)程對象

（解決上一個程序中兩個子進(jìn)程重復(fù)代碼的問題）

from multiprocessing import Processimport time,os#繼承Process類class SubProcess(Process):    def __init__(self,interval,name=""): #子類SubProcess的構(gòu)造方法        Process.__init__(self) #Process.__init__(self) #繼承父類的構(gòu)造方法        self.interval = interval        self.name = name         #重寫父類方法    def run(self):        print("子進(jìn)程(%s)開始執(zhí)行，父進(jìn)程為(%s)" % (os.getpid(), os.getpid()))  # os.getpid()為獲取進(jìn)程號        t_stat = time.time()  # 計時開始        time.sleep(self.interval)  # 程序?qū)⒈粧炱餹秒        t_end = time.time()  # 計時結(jié)束        print("子進(jìn)程(%s)的執(zhí)行時間為'%0.2f'秒" % (os.getpid(), t_end - t_stat))if __name__ == '__main__':    print("-----父進(jìn)程開始執(zhí)行-------")    print("父進(jìn)程啟動時，父進(jìn)程PID為：(%s)"%(os.getpid()))  #打印父進(jìn)程啟動時父進(jìn)程的PID    p1 = SubProcess(interval=1,name='ZARD1') #實例化子進(jìn)程1    p2 = SubProcess(interval=2,name=('ZARD2')) #實例化子進(jìn)程2    p1.start() #啟動子進(jìn)程1    p2.start() #啟動子進(jìn)程2    #此時父進(jìn)程仍然在執(zhí)行    print("pi.is_alive=%s"%(p1.is_alive()))  #p1.is_alive() 判斷子進(jìn)程p1是否還在執(zhí)行，執(zhí)行則返回ture    print("p2_is_alive=%s"%(p2.is_alive()))    #輸出此時進(jìn)程執(zhí)行過程中的PID(進(jìn)程號)    print("p1.name = %s"%(p1.name))    print("執(zhí)行過程中的p1的進(jìn)程號p1.pid=%s"%(p1.pid))    print("p2.name = %s" % (p2.name))    print("執(zhí)行過程中的p2的進(jìn)程號p2.pid=%s" % (p2.pid))    print("---等待子進(jìn)程結(jié)束----")    p1.join() #等待子程序p1結(jié)束    p2.join()    print("父進(jìn)程結(jié)束，此時父進(jìn)程的進(jìn)程號為：(%s)"%(os.getpid()))from multiprocessing import Process
import time,os
 
#繼承Process類
class SubProcess(Process):
    def __init__(self,interval,name=""): #子類SubProcess的構(gòu)造方法
        Process.__init__(self) #Process.__init__(self) #繼承父類的構(gòu)造方法
        self.interval = interval
        self.name = name
     
    #重寫父類方法
    def run(self):
        print("子進(jìn)程(%s)開始執(zhí)行，父進(jìn)程為(%s)" % (os.getpid(), os.getpid()))  # os.getpid()為獲取進(jìn)程號
        t_stat = time.time()  # 計時開始
        time.sleep(self.interval)  # 程序?qū)⒈粧炱餹秒
        t_end = time.time()  # 計時結(jié)束
        print("子進(jìn)程(%s)的執(zhí)行時間為'%0.2f'秒" % (os.getpid(), t_end - t_stat))
 
if __name__ == '__main__':
    print("-----父進(jìn)程開始執(zhí)行-------")
    print("父進(jìn)程啟動時，父進(jìn)程PID為：(%s)"%(os.getpid()))  #打印父進(jìn)程啟動時父進(jìn)程的PID
    p1 = SubProcess(interval=1,name='ZARD1') #實例化子進(jìn)程1
    p2 = SubProcess(interval=2,name=('ZARD2')) #實例化子進(jìn)程2
    p1.start() #啟動子進(jìn)程1
    p2.start() #啟動子進(jìn)程2
    #此時父進(jìn)程仍然在執(zhí)行
    print("pi.is_alive=%s"%(p1.is_alive()))  #p1.is_alive() 判斷子進(jìn)程p1是否還在執(zhí)行，執(zhí)行則返回ture
    print("p2_is_alive=%s"%(p2.is_alive()))
    #輸出此時進(jìn)程執(zhí)行過程中的PID(進(jìn)程號)
    print("p1.name = %s"%(p1.name))
    print("執(zhí)行過程中的p1的進(jìn)程號p1.pid=%s"%(p1.pid))
    print("p2.name = %s" % (p2.name))
    print("執(zhí)行過程中的p2的進(jìn)程號p2.pid=%s" % (p2.pid))
    print("---等待子進(jìn)程結(jié)束----")
    p1.join() #等待子程序p1結(jié)束
    p2.join()
    print("父進(jìn)程結(jié)束，此時父進(jìn)程的進(jìn)程號為：(%s)"%(os.getpid()))

代碼分析：

1.4 Pool進(jìn)程池

（解決當(dāng)要創(chuàng)造成百上千個進(jìn)程的情況）

例子：有三個水槽，要接10桶水，我們最多只能同時接3盆，第10盆隨便找一個盆，其他兩個閑置。

from multiprocessing import Pool #導(dǎo)入進(jìn)程池
import os,time
 
def task(name):
    print("子進(jìn)程(%s)執(zhí)行task %s..."%(os.getpid(),name))
    time.sleep(2) #休眠2秒
 
if __name__ == '__main__':
    print("---父進(jìn)程(%s)啟動---"%(os.getpid()))
    p = Pool(3)  #定義一個進(jìn)程池，一次最多容納三個進(jìn)程，即一次最多可同時執(zhí)行三個子進(jìn)程
    for i in range(10):
        p.apply_async(task,args=(i,)) #使用非阻塞的方式調(diào)用task
    print("---等待所有子進(jìn)程結(jié)束---")
    p.close() #關(guān)閉進(jìn)程池，關(guān)閉后進(jìn)程池不再接收新的請求
    p.join() #等待子進(jìn)程結(jié)束
    print("全部子進(jìn)程結(jié)束")

小結(jié)：定義一個進(jìn)程池，并規(guī)定一個池子中可以同時執(zhí)行多少個進(jìn)程，可以實現(xiàn)多個進(jìn)程分批次的執(zhí)行。

注意：

（1）區(qū)分使用Process模塊與使用進(jìn)程池模塊Pool的區(qū)別

（2）對于實例對象 p，p.start()表示開始執(zhí)行進(jìn)程；p.join()表示結(jié)束進(jìn)程

（3）注意加強(qiáng)阻塞與非阻塞知識點的學(xué)習(xí)

2. 驗證進(jìn)程是否能共享信息

引例：

from multiprocessing import Process
 
#子進(jìn)程1
def plus():
    print("---子進(jìn)程1開始執(zhí)行---")
    global g_num #聲明全局變量
    g_num += 50
    print("在子進(jìn)程1下：g_num = %d"%(g_num))
    print("---子進(jìn)程1結(jié)束運(yùn)行---")
 
#子進(jìn)程2
def minus():
    print("---子進(jìn)程2開始執(zhí)行---")
    global g_num #聲明全局變量
    g_num -= 50
    print("在子進(jìn)程2下：g_num = %d"%(g_num))
    print("---子進(jìn)程2結(jié)束運(yùn)行---")
 
g_num = 100 #賦值全局變量
if __name__ == '__main__':
    print("---主進(jìn)程啟動---")
    print("在主進(jìn)程運(yùn)行中，g_num = %d"%(g_num))
    child1 = Process(target=plus) #實例化子進(jìn)程1
    child2 = Process(target=minus)  # 實例化子進(jìn)程2
    child1.start() #啟動子進(jìn)程1
    child2.start()
 
    child1.join() #等待子進(jìn)程結(jié)束
    child2.join()
    print("---主進(jìn)程結(jié)束---")

由以上例子可見，對于進(jìn)程1，2而言，全局變量g_num并不互相影響。即有如下關(guān)系：

那么就有如下問題了，思考：如何才能實現(xiàn)進(jìn)程之間的通訊？

答：通過 multiprocessing Queue（隊列）， Pipes（管道），接下來主要演示 Queue（隊列）模塊。

2.1 Queue（隊列）模塊：

2.1.1 隊列簡介：

1.新來的排隊的在隊尾

2.最前面的完成離隊后，后面一個跟上

多進(jìn)程隊列的使用Queue，本身他就是一個消息隊列程序：

實踐：當(dāng)Queue(3)時：

from multiprocessing import Queue #導(dǎo)入隊列模塊
if __name__ == '__main__':
    q = Queue(3) #初始化一個Queue對象，最多只能接受3條put信息
    #消息寫入隊列
    q.put("消息1") #將此消息1寫入隊列
    q.put("消息2")
    print(q.full()) #q.full() 是驗證隊列是否已滿  沒滿則返回False
    q.put("消息3")  #將此消息3寫入隊列
    print(q.full()) #q.full() 是驗證隊列是否已滿  滿了則返回Ture
 
    #利用try看看隊列已滿是否還可再塞入信息
    try:
        q.put("嘗試塞入第4條信息",True,1) #可能出問題的代碼
    except:
        print("嘗試?yán)胵.put()向隊列中繼續(xù)添加信息：")
        print("隊列已滿，現(xiàn)有消息數(shù)量為：%d，無法繼續(xù)添加信息。"%(q.qsize()))  #q.qsize() 返回隊列中的已有信息數(shù)量
 
    try:
        q.put_nowait("嘗試塞入第4條信息")  # 可能出問題的代碼
    except:
        print("嘗試?yán)胵.put_nowait()向隊列中繼續(xù)添加信息：")
        print("隊列已滿，現(xiàn)有消息數(shù)量為：%d，無法繼續(xù)添加信息。" % (q.qsize()))  # q.qsize() 返回隊列中的已有信息數(shù)量
 
    #獲取（打?。╆犃兄械男畔?
    if not q.empty():
        print("---從隊列中讀取信息---")
        for i in range(q.qsize()):  #循環(huán)打印隊列信息
            print(q.get_nowait())  #q.get_nowait() 讀取隊列信息

解決：

當(dāng)Queue(4)時，打印的結(jié)果如下，可見此時可繼續(xù)向隊列中塞入第4條信息。

小結(jié)：

multiprocessing.Process 可以創(chuàng)建多進(jìn)程，使用multiprocessing Queue可以實現(xiàn)隊列操作。

2.2 實現(xiàn)進(jìn)程間的通信

from multiprocessing import Process,Queue
import time
 
#子進(jìn)程1：向隊列中寫入數(shù)據(jù)
def write_date(q):
    if not q.full(): #若隊列沒滿，則寫入數(shù)據(jù)
        for i in range(5):
            date = "數(shù)據(jù)" + str(i)
            q.put(date) #向隊列中寫入數(shù)據(jù)
            print("已寫入：%s" % date)
#子進(jìn)程2：向隊列中讀取數(shù)據(jù)
def read_date(q):
    time.sleep(1)  # 休眠1s
    while not q.empty():
        print("讀取:%s" % q.get(True, 2))  # 等待2s，如果還沒讀取到消息，拋出異常
 
#主程序
if __name__ == '__main__':
    print("---主程序啟動---")
    q = Queue()  # 父進(jìn)程創(chuàng)建Queue，并傳給各個子進(jìn)程
    write_child = Process(target=write_date,args=(q,)) # 實例化進(jìn)程對象，其中args=(q,)表示把隊列傳給子進(jìn)程1
    read_child = Process(target=read_date,args=(q,))
    write_child.start()  # 啟動子進(jìn)程1，寫入
    read_child.start()  # 啟動子進(jìn)程2，讀取
    write_child.join()  # 等待子進(jìn)程1結(jié)束
    read_child.join()  # 等待子進(jìn)程2結(jié)束
    print('-----父進(jìn)程結(jié)束-----')