python進階（8）多進程

進程

 

前置知識點

• 併發：指的是任務數多餘cpu核數，經過操做系統的各類任務調度算法，實現用多個任務「一塊兒」執行（實際上總有一些任務不在執行，由於切換任務的速度至關快，看上去一塊兒執行而已）
• 並行：指的是任務數小於等於cpu核數，即任務真的是一塊兒執行的
   

進程的建立

multiprocessing模塊就是跨平臺版本的多進程模塊，提供了一個Process類來表明一個進程對象，這個對象能夠理解爲是一個獨立的進程，能夠執行另外的事情

from multiprocessing import Process
import os


# 子進程要執行的代碼
def run_proc(name):
    print('啓動子線程{}{}'.format(name, os.getpid()))


if __name__ == '__main__':
    print('父進程{}'.format(os.getpid()))
    p = Process(target=run_proc, args=('test',))
    print('子進程將要啓動')
    p.start()
    p.join()
    print('子線程結束')

 

進程pid

from multiprocessing import Process
import os
import time


def run_proc():
    """子進程要執行的代碼"""
    print('子進程運行中，pid=%d...' % os.getpid())  # os.getpid獲取當前進程的進程號
    print('子進程將要結束...')


if __name__ == '__main__':
    print('父進程pid: %d' % os.getpid())  # os.getpid獲取當前進程的進程號
    p = Process(target=run_proc)
    p.start()

>>> 父進程pid: 3580
>>> 子進程運行中，pid=3581...
>>> 子進程將要結束...

 

Process語法結構

Process([group [, target [, name [, args [, kwargs]]]]])

• target：若是傳遞了函數的引用，能夠任務這個子進程就執行這裏的代碼
• args：給target指定的函數傳遞的參數，以元組的方式傳遞
• kwargs：給target指定的函數傳遞命名參數
• name：給進程設定一個名字，能夠不設定
• group：指定進程組，大多數狀況下用不到
  Process建立的實例對象的經常使用方法：
• start()：啓動子進程實例（建立子進程）
• is_alive()：判斷進程子進程是否還在活着
• join([timeout])：是否等待子進程執行結束，或等待多少秒
• terminate()：無論任務是否完成，當即終止子進程
  Process建立的實例對象的經常使用屬性：
• name：當前進程的別名，默認爲Process-N，N爲從1開始遞增的整數
• pid：當前進程的pid（進程號）
   

給子進程指定的函數傳遞參數

from multiprocessing import Process
import os
from time import sleep


def run_proc(name, age, **kwargs):
    for i in range(10):
        print('子進程運行中，name= %s,age=%d ,pid=%d...' % (name, age, os.getpid()))
        print(kwargs)
        sleep(0.2)

if __name__=='__main__':
    p = Process(target=run_proc, args=('test',18), kwargs={"m":20})
    p.start()
    sleep(1)  # 1秒中以後，當即結束子進程
    p.terminate()
    p.join()

>>> 子進程運行中，name= test,age=18 ,pid=3593...
>>> {'m': 20}
>>> 子進程運行中，name= test,age=18 ,pid=3593...
>>> {'m': 20}
>>> 子進程運行中，name= test,age=18 ,pid=3593...
>>> {'m': 20}
>>> 子進程運行中，name= test,age=18 ,pid=3593...
>>> {'m': 20}
>>> 子進程運行中，name= test,age=18 ,pid=3593...
>>> {'m': 20}

 

進程和線程的區別

• 進程是資源調度的基本單位，而線程是程序執行的基本單位
• 不一樣進程的地址空間是獨立的，而同一進程中的線程之間共享
• 進程之間通訊必須使用操做系統提供的進程間通訊機制，同一進程中的各線程能夠直接通訊
• 多進程之間能夠併發執行，多線程之間也能夠併發執行
• 線程切換的開銷要比進程切換的開銷小
   

進程間通訊

from multiprocessing import Process, Queue
import os, time, random


# 寫數據進程執行的代碼:
def write(q):
    print('Process to write: %s' % os.getpid())
    for value in ['A', 'B', 'C']:
        print('Put %s to queue...' % value)
        q.put(value)
        time.sleep(random.random())


# 讀數據進程執行的代碼:
def read(q):
    print('Process to read: %s' % os.getpid())
    while True:
        value = q.get(True)
        print('Get %s from queue.' % value)


if __name__ == '__main__':
    # 父進程建立Queue，並傳給各個子進程：
    q = Queue()
    pw = Process(target=write, args=(q,))
    pr = Process(target=read, args=(q,))
    # 啓動子進程pw，寫入:
    pw.start()
    # 啓動子進程pr，讀取:
    pr.start()
    # 等待pw結束:
    pw.join()
    # pr進程裏是死循環，沒法等待其結束，只能強行終止:
    pr.terminate()

初始化Queue()對象時（例如：q=Queue()），若括號中沒有指定最大可接收的消息數量，或數量爲負值，那麼就表明可接受的消息數量沒有上限（直到內存的盡頭）；

• Queue.qsize()：返回當前隊列包含的消息數量；
• Queue.empty()：若是隊列爲空，返回True，反之False ；
• Queue.full()：若是隊列滿了，返回True,反之False；
• Queue.get([block[, timeout]])：獲取隊列中的一條消息，而後將其從列隊中移除，block默認值爲True；
  1）若是block使用默認值，且沒有設置timeout（單位秒），消息列隊若是爲空，此時程序將被阻塞（停在讀取狀態），直到從消息列隊讀到消息爲止，若是設置了timeout，則會等待timeout秒，若還沒讀取到任何消息，則拋出"Queue.Empty"異常；
   
  2）若是block值爲False，消息列隊若是爲空，則會馬上拋出"Queue.Empty"異常；
• Queue.get_nowait()：至關Queue.get(False)；
• Queue.put(item,[block[, timeout]])：將item消息寫入隊列，block默認值爲True；
   
  1）若是block使用默認值，且沒有設置timeout（單位秒），消息列隊若是已經沒有空間可寫入，此時程序將被阻塞（停在寫入狀態），直到從消息列隊騰出空間爲止，若是設置了timeout，則會等待timeout秒，若還沒空間，則拋出"Queue.Full"異常；
   
  2）若是block值爲False，消息列隊若是沒有空間可寫入，則會馬上拋出"Queue.Full"異常；
• Queue.put_nowait(item)：至關Queue.put(item, False)；
   

進程池

"""
若是要啓動大量的子進程，能夠用進程池的方式批量建立子進程：
"""
from multiprocessing import Pool
import os, time, random


def long_time_task(name):
    print('運行任務 %s (%s)...' % (name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('任務 %s 運行 %0.2f 秒' % (name, (end - start)))


if __name__ == '__main__':
    print('父進程 %s.' % os.getpid())
    p = Pool(4)  # 建立進程池中最多存4個子進程
    for i in range(5):
          # Pool().apply_async(要調用的目標,(傳遞給目標的參數元祖,))
            # 每次循環將會用空閒出來的子進程去調用目標
        p.apply_async(long_time_task, args=(i,))
    print('等待全部子進程完成...')
    p.close()
    p.join()
    print('全部子進程完成.')

>>> 等待全部子進程完成...
>>> 運行任務 0 (3722)...
>>> 運行任務 1 (3723)...
>>> 運行任務 2 (3724)...
>>> 運行任務 3 (3725)...
>>> 任務 3 運行 0.67 秒
>>> 運行任務 4 (3725)...
>>> 任務 2 運行 1.29 秒
>>> 任務 0 運行 2.00 秒
>>> 任務 1 運行 2.77 秒
>>> 任務 4 運行 2.31 秒
>>> 全部子進程完成.

 

multiprocessing.Pool經常使用函數解析：

• apply_async(func[, args[, kwds]]) ：使用非阻塞方式調用func（並行執行，堵塞方式必須等待上一個進程退出才能執行下一個進程），args爲傳遞給func的參數列表，kwds爲傳遞給func的關鍵字參數列表；
• close()：關閉Pool，使其再也不接受新的任務；
• terminate()：無論任務是否完成，當即終止；
• join()：主進程阻塞，等待子進程的退出， 必須在close或terminate以後使用；
   

代碼解讀：

對Pool對象調用join()方法會等待全部子進程執行完畢，調用join()以前必須先調用close()，調用close()以後就不能繼續添加新的Process了。
請注意輸出的結果，task 0，1，2，3是馬上執行的，而task 4要等待前面某個task完成後才執行，這是由於Pool的默認大小在個人電腦上是4，所以，最多同時執行4個進程。這是Pool有意設計的限制，並非操做系統的限制。若是改爲：

p = Pool(5)

就能夠同時跑5個進程。
因爲Pool的默認大小是CPU的核數，若是你不幸擁有8核CPU，你要提交至少9個子進程才能看到上面的等待效果。
 

進程池中的Queue

# 修改import中的Queue爲Manager
from multiprocessing import Manager,Pool
import os,time,random

def reader(q):
    print("reader啓動(%s),父進程爲(%s)" % (os.getpid(), os.getppid()))
    for i in range(q.qsize()):
        print("reader從Queue獲取到消息：%s" % q.get(True))

def writer(q):
    print("writer啓動(%s),父進程爲(%s)" % (os.getpid(), os.getppid()))
    for i in "itcast":
        q.put(i)

if __name__=="__main__":
    print("(%s) start" % os.getpid())
    q = Manager().Queue()  # 使用Manager中的Queue
    po = Pool()
    po.apply_async(writer, (q,))

    time.sleep(1)  # 先讓上面的任務向Queue存入數據，而後再讓下面的任務開始從中取數據

    po.apply_async(reader, (q,))
    po.close()
    po.join()
    print("(%s) End" % os.getpid())

>>> (4157) start
>>> writer啓動(4159),父進程爲(4157)
>>> reader啓動(4160),父進程爲(4157)
>>> reader從Queue獲取到消息：i  
>>> reader從Queue獲取到消息：t
>>> reader從Queue獲取到消息：c
>>> reader從Queue獲取到消息：a
>>> reader從Queue獲取到消息：s
>>> reader從Queue獲取到消息：t
>>> (4157) End