第十五篇：多任務系列之進程（二）

 

　　本篇主要介紹關於多進程的相關知識，同時需對進程與線程之間的關係和區別及應用進行了解，其實包含 multiprocessing模塊下的 Process類、進行進程間通訊的 Queue類以及進程池 Pool類的學習，最後實現一個實例 模擬文件夾copy器；

1、什麼是進程？

　　首先咱們在瞭解進程以前，咱們須要瞭解一下什麼是程序？

　　程序：

　　  其實就是一堆躺在操做系統下的二進制文件，是靜態的；例如 wechar.exe文件，在咱們沒點擊它時，就是一個靜態的可執行文件。

　　進程：

　　  進程就是跑起來的程序，即代碼 + 操做系統根據其需求爲其分配的資源稱之爲進程；例如：當一個 Wechat.exe文件咱們點擊運行時，操做系統會爲其調配其所需資源，則運行該程序就是一個進程；進程是操做系統分配資源的基本單位。

　　須要注意的是：同一個程序執行兩次，那就是兩個進程；例如 打開騰訊視頻--》一個能夠播放西遊記，一個能夠播放紅樓夢；

　　進程的狀態：

　　  在現實工做中，任務的數量要遠遠大於CPU的數量，因此要想實現真正意義上的並行幾乎是不可能的，因此大多數狀況下都是併發（僞並行），故當運行多個進程時，有的程序處於等待的狀態、有的程序處於運行的狀態，則進程就能夠分爲如下幾種狀態：

　　

　三種狀態：

　　就緒：全部準備工做均完成，等待着操做系統分配CPU進行執行；

　　運行：正在被CPU執行的程序

　   堵塞：即待某些條件知足，例如一個程序sleep，此時就處於等待態。

 

---

 

 

2、多進程的建立過程：、

　　咱們建立多進程主要經過模塊 multiprocessing 下的Process類來建立進程對象，經過進程對象來實現建立進程的效果，很少說直接看示例，例如：

#!/usr/bin/env python # -*- coding:utf-8 -*-

import multiprocessing,time def sing(name): for i in range(10000): print("%s is singing --->%s"%(name,i)) # time.sleep(0.5)

def dance(name): for i in range(10000): print("%s is dancing ---> %s"%(name,i)) # time.sleep(0.5)

def main(): # 一、建立進程對象，傳入工做函數，以及工做函數所需參數
    p1 = multiprocessing.Process(target=sing,args=("alex",)) p2 =multiprocessing.Process(target=dance,args=("liudehua",)) # 二、啓動進程
 p1.start() p2.start() print("This is main process...") # 三、主進程等子進程運行完才關閉
 p1.join() p2.join() if __name__=="__main__": main()

　　從上述例子中能夠看出：進程的建立過程與線程的建立過程幾乎相同，除了使用模塊不相同，其餘的幾乎相似。那麼既然有了線程爲何還要有進程呢？接下來咱們就瞭解一下進程和線程的區別。

 

---

 

3、線程和進程的區別及應用場景

　　首先咱們須要知道不管是線程仍是進程都可以實現多任務的效果，可是一個進程必然有一個線程，同時一個進程也能夠有多個線程，例如：一個WeChat能夠打開多個聊天窗口，與不一樣的人進行聊天。其關係以下圖：

　　

　從上面的圖咱們能夠很明確的看出進程與線程之間的關係和區別：

　　　一、一個程序至少有一個進程（固然也能夠有多個進程，例如：一個QQ能夠同時登入兩個用戶；），一個進程至少有一個線程（固然也可由多個線程，例如：一個QQ能夠同時與兩我的進行聊天。）

　　　二、線程沒法獨立存在，只能依附在進程內執行，即在QQ沒有運行的狀況下QQ聊天是沒法實現的。

　　　三、一個進程的運行須要調配的資源要多得多，可是進程要穩定許多，即一個進程的關閉不會影響另一個進程的運行，即關閉網易雲音樂隊每天靜聽的執行沒有任何影響；可是一個主線程的關閉必然會致使全部的子線程死亡，穩定性不如進程。

　　　四、線程能夠共享全局變量，而進程間沒法全局變量的共享；（要想實現進程間的通訊需使用隊列等，在後面介紹。）

　　　五、進程是操做系統進行資源管理的基本單位，而線程是對操做系統的分配下來的資源調用的基本單位。

 

　其實咱們也能夠將進程和線程比喻成：車間流水線的工做（進程） 和  流水線上工人的工做（線程）：

　　

　我的理解： 

　　　能夠將進程好比成流水線的運行，即在該流水線的運行須要各類資源（即進程的運行須要操做系統調配資源），以及許多工人，這些工人在不停的工做（線程），他們對該流水線上的資源是共享的（線程間共享全局變量），同時若想提升流水線的運行速度，則能夠爲流水線增長一些工人（多線程），而同時若想提升車間的生產能力，則能夠經過增長多條流水線（多進程），可是每增長一條流水線，所需的成本也高一些（即進程需調配的資源增長），同時每一個流水線上的資源是不共享的（即進程間不共享全局變量。）。

 

---

 

四、獲取進程的PID和PPID

　　每一個進程均會有進程號以及父進程號，在Linux中能夠經過 ps - aux或者top命令查看當前進程，kill命令+pid殺死進程，而在Windows中能夠經過在cmd終端中 tasklist命令查看進程，經過taskkill +pid 殺死進程；那麼咱們如何在獲取進程號呢？

 

#!/usr/bin/env python # -*- coding:utf-8 -*-

#!/usr/bin/env python # -*- coding:utf-8 -*-

import multiprocessing,time,os def sing(name): """唱歌"""
    for i in range(3): print("%s is singing --->%s"%(name,i)) time.sleep(0.5) print("---in sing process-- pid:%s, ppid:%s ."%(os.getpid(),os.getppid()))  # 查看運行sing函數的子進程的pid和ppid

def dance(name): for i in range(3): print("%s is dancing ---> %s"%(name,i)) time.sleep(0.5) print("---in dance process-- pid:%s, ppid:%s ." % (os.getpid(), os.getppid()))   # 查看運行dance函數的子進程的pid和 ppid

def main(): # 一、建立進程對象，傳入工做函數，以及工做函數所需參數
    p1 = multiprocessing.Process(target=sing,args=("alex",)) p2 =multiprocessing.Process(target=dance,args=("liudehua",)) # 二、啓動進程
 p1.start() p2.start() print("This is main process...") print("---in main process-- pid:%s, ppid:%s ." % (os.getpid(), os.getppid())) time.sleep(0.5) p1.terminate() #殺死子進程

    # 三、主進程等子進程運行完才關閉
 p1.join() p2.join() if __name__=="__main__": main()

 

  注：從運行結果得知：子進程的父進程就是主進程，而主進程的父進程其實爲爲當前集成開發IDE的運行進程PID。

　　 關於進程的運行順序：

　　　　主進程最早運行，而子進程的運行順序徹底由操做系統所決定。

 

 　　孤兒進程：

　　　　當主進程被殺死的時候，子進程是不會死亡的，這種進程被稱爲孤兒進程；而除了一個pid爲1的進程外，其餘的進程均有父進程，這時這些孤兒進程會被收養到孤兒院，根據孤兒進程的ppid咱們能夠得知該孤兒院爲 進程ID爲2041的upstart進程。

　　 殭屍進程：

　　　　當子進程被殺死時，父進程不會爲該子進程收屍即不會對其所佔資源進行回收，則該子進程被稱爲殭屍進程，這類殭屍進程是對系統的運行時不利的，故咱們一般經過join()方法來避免殭屍進程的出現。

 

---

 

五、Queue實現進程間的通訊

　　首先，咱們須要知道進程間不共享全局變量，例如：網易雲音樂沒法播放酷狗中的音樂，接下來咱們從一個例子中驗證：　

#!/usr/bin/env python # -*- coding:utf-8 -*-
import multiprocessing,time # 建立全局變量
name =["alex","little-five","amanda"] def work1(): # work1修改全局變量
    name.append("hello") print(name) def work2(): # work2查看全局變量
    print("in the work2:--> ",name) def main(): p1 = multiprocessing.Process(target=work1) p2 = multiprocessing.Process(target=work2) p1.start() time.sleep(1)  # 休眠1s，保證進程p1先運行
 p2.start() print("in main process : ",name) p1.join() p2.join() if __name__=="__main__": main()

　　從輸出結果能夠得出 ： 進程間不共享全局變量，即即便主進程與子進程間也不共享全局變量。這是因爲：

　　進程每建立子線程時，老版本內核會至關於將主進程的資源和代碼copy一份，但每一個子線程運行的代碼位置不一樣，可是新版本則不會拷貝代碼而是代碼共享，而當某個進程要修改代碼時，則是相應的代碼進行拷貝即寫時拷貝。

 　  注：在線程中咱們有互斥鎖，而進程用於不存在全局變量共享問題，則就不須要互斥鎖，而是利用Queue來實現進程間的通訊。

 

　　那麼咱們怎麼實現進程間的通訊呢？

#!/usr/bin/env python # -*- coding:utf-8 -*-

import multiprocessing def recv_msg(q): """數據的接收""" names =["alex","wupeiqi","linghaifeng"] for name in names: q.put(name) #將每一個名字傳入隊列

def analysis_msg(q): """數據的分析""" new_names =list() while not q.empty(): new_name=q.get()  # 獲取隊列中的名字
 new_names.append(new_name) print(list(map(lambda x:x.title(),new_names)))  #經過map 函數對列表進行處理
 q=multiprocessing.Queue()  # 建立隊列對象

def main(): p1 = multiprocessing.Process(target=recv_msg,args=(q,))  #咱們將隊列做爲工做函數的參數傳入進程
    p2 = multiprocessing.Process(target=analysis_msg,args=(q,)) p1.start() p2.start() p1.join() p2.join() if __name__=="__main__": main()

　　注：其實隊列就像一根水管，即先入先出，後入後出，數據從水管的一邊傳入從另外一邊輸出，經過隊列來實現進程間的數據傳遞，實現進程間的通訊。

 

---

 

六、 進程池

　　在學習進程池以前，咱們須要瞭解爲何咱們須要學習進程池：

　　　咱們知道經過一個進程咱們能夠實現一個複雜的任務，可是當咱們有許多個任務即成千上萬的任務或者這些任務要重複執行時，這時候咱們就須要用到進程池。下面咱們看一下進程池的用法：

#!/usr/bin/env python # -*- coding:utf-8 -*-

import multiprocessing,time,os def cal(num):
　　"""計算""" n =0 for i in range(num): n+=1 time.sleep(0.2) print("01-the result is -->%s ,PID: %s ."%(n,os.getpid())) def worker(lis): """計算map函數運行時間""" start_time = time.time() new_list=list(map(lambda x:x.title(),lis )) time.sleep(1) end_time =time.time() print("02--close_time -->%s,PID:%s ,"%(end_time-start_time,os.getpid())) def main(): po = multiprocessing.Pool(3) # 定義進程池，規定進程池中最大的進程數爲3 for i in range(100): # 讓上面的兩個任務重複執行100次 po.apply_async(cal,args=(100,)) # 每次循環結束均會有子進程去執行任務 po.apply_async(worker,args=(["alex","little-five","hello"],)) print("---start----") po.close() # 關閉進程池 po.join() # 等待子進程所有執行完，必須放在close以後 print("---end-----") if __name__ == "__main__": main()

　　那麼進程池的實現機制是怎麼樣的呢？

　　始化Pool時，能夠指定一個最大進程數，當有新的請求提交到Pool中時，若是池尚未滿，那麼就會建立一個新的進程用來執行該請求；但若是池中的進程數已經達到指定的最大值，那麼該請求就會等待，直到池中有進程結束，纔會用以前的進程來執行新的任務。若任務數量十分多的時候，這些任務會被操做系統加載到一個內存中，被進程池中的進程循環執行。

 

---

 

七、實例：，模擬文件夾copy器 

#!/usr/bin/env python # -*- coding:utf-8 -*-
import os,multiprocessing,time def copy_file(queue,source_folder,new_folder,file_name): """將源文件內文件數據讀取出來 寫入新文件夾內的文件中""" source_file_path = source_folder + "/" + file_name f1 = open(source_file_path,"rb") file_data = f1.read() f1.close() new_file_path = new_folder + "/" + file_name f2 = open(new_file_path,"wb") f2.write(file_data) f2.close() # 每copy一個文件，將文件名傳入隊列，告訴主進程已完成一個文件的拷貝
 queue.put(file_name) def main(): # 一、獲取源文件夾文件名
    while True: source_folder_name = input("請輸入源文件夾名稱：--》") if os.path.exists(source_folder_name): break

    # 二、建立目標文件夾
    new_folder_name = source_folder_name + "-copy"
    try: os.mkdir(new_folder_name) except Exception as e: pass

    # 三、獲取源文件的文件
    file_list = os.listdir(source_folder_name) # 建立隊列來實現主線程和子線程間的通訊
    q = multiprocessing.Manager().Queue() # 四、經過進程池來實現原文件數據的讀取和寫入新文件(文件數量較多，故採用進程池)
    po = multiprocessing.Pool(5) for file_name in file_list: #沒一個文件的成品copy用進程池中的一個進程來實現
        po.apply_async(copy_file,args=(q,source_folder_name,new_folder_name,file_name)) po.close() po.join() # 實現進度條的效果
    source_len = len(file_list)  # 獲取源文件列表的長度
    while True: time.sleep(0.2) over_copy_filename = q.get()  # 獲取已完成copy文件的文件名
        # new_file_list = os.listdir(new_folder)
        if over_copy_filename in file_list:  # 若是該文件已完成copy，則移除從源文件列表中移除該文件名
 file_list.remove(over_copy_filename) len_tax = len(file_list)/source_len print(" \r 進度爲：--》%0.2f %%" % (100 * (1 - len_tax)), end="") if len_tax == 0: print("\n copying file is over ..") break



if __name__ =="__main__": main()