摘要：本節講學習的多進程。和之前的的不同點是丟向的函數有返回值，而的沒有返回值。所以接下來讓我們來看下這兩個進程是否會出現沖突。

本節講學習Python的多進程。

多進程 Multiprocessing 和多線程 threading 類似, 他們都是在 python 中用來并行運算的. 不過既然有了 threading, 為什么 Python 還要出一個 multiprocessing 呢? 原因很簡單, 就是用來彌補 threading 的一些劣勢, 比如在 threading 教程中提到的GIL.

使用 multiprocessing 也非常簡單, 如果對 threading 有一定了解的朋友, 你們的享受時間就到了. 因為 python 把 multiprocessing 和 threading 的使用方法做的幾乎差不多. 這樣我們就更容易上手. 也更容易發揮你電腦多核系統的威力了!

import multiprocessing as mp
import threading as td

def job(a,d):
    print("aaaaa")

t1 = td.Thread(target=job,args=(1,2))
p1 = mp.Process(target=job,args=(1,2))
t1.start()
p1.start()
t1.join()
p1.join()

從上面的使用對比代碼可以看出，線程和進程的使用方法相似。

在運用時需要添加上一個定義main函數的語句

if __name__=="__main__":

完整的應用代碼：

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_test.py
@time: 18/8/26 01:12
"""

import multiprocessing as mp

def job(a, d):
    print a, d

if __name__ == "__main__":
    p1 = mp.Process(target=job, args=(1, 2))
    p1.start()
    p1.join()

運行環境要在terminal環境下，可能其他的編輯工具會出現運行結束后沒有打印結果，在terminal中的運行后打印的結果為：

?  baseLearn python ./process/process_test.py
1 2
?  baseLearn

Queue的功能是將每個核或線程的運算結果放在隊里中， 等到每個線程或核運行完畢后再從隊列中取出結果， 繼續加載運算。原因很簡單, 多線程調用的函數不能有返回值, 所以使用Queue存儲多個線程運算的結果

process_queue.py

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_queue.py
@time: 18/8/26 01:12
"""

import multiprocessing as mp

# 定義一個被多線程調用的函數，q 就像一個隊列，用來保存每次函數運行的結果
def job(q):
    res = 0
    for i in range(1000):
        res += i + i**2 + i**3
    q.put(res)   #queue

if __name__ == "__main__":
    q = mp.Queue()
    p1 = mp.Process(target=job, args=(q,))
    p2 = mp.Process(target=job, args=(q,))

    # 分別啟動、連接兩個線程
    p1.start()
    p2.start()
    p1.join()
    p2.join()

    # 上面是分兩批處理的，所以這里分兩批輸出，將結果分別保存
    res1 = q.get()
    res2 = q.get()

    print res1,res2

打印輸出結果：

? python ./process/process_queue.py
249833583000 249833583000

進程池就是我們將所要運行的東西，放到池子里，Python會自行解決多進程的問題。

首先import multiprocessing 和定義job()

import multiprocessing as mp

def job(x):
    return x*x

然后我們定義一個Pool

pool = mp.Pool()

有了池子之后，就可以讓池子對應某一個函數，我們向池子里丟數據，池子就會返回函數返回的值。 Pool和之前的Process的不同點是丟向Pool的函數有返回值，而Process的沒有返回值。

接下來用map()獲取結果，在map()中需要放入函數和需要迭代運算的值，然后它會自動分配給CPU核，返回結果

res = pool.map(job, range(10))

讓我們來運行一下

def multicore():
    pool = mp.Pool()
    res = pool.map(job, range(10))
    print(res)
    
if __name__ == "__main__":
    multicore()

完成代碼：

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_queue.py
@time: 18/8/26 01:12
"""

import multiprocessing as mp

def job(x):
    return x*x  # 注意這里的函數有return返回值

def multicore():
    pool = mp.Pool()
    res = pool.map(job, range(10))
    print(res)
    
if __name__ == "__main__":
    multicore()

執行結果：

?  baseLearn python ./process/process_pool.py
[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

我們怎么知道Pool是否真的調用了多個核呢？我們可以把迭代次數增大些，然后打開CPU負載看下CPU運行情況

打開CPU負載(Mac)：活動監視器 > CPU > CPU負載(單擊一下即可)

Pool默認大小是CPU的核數，我們也可以通過在Pool中傳入processes參數即可自定義需要的核數量

def multicore():
    pool = mp.Pool(processes=3) # 定義CPU核數量為3
    res = pool.map(job, range(10))
    print(res)

Pool除了map()外，還有可以返回結果的方式，那就是apply_async().

apply_async()中只能傳遞一個值，它只會放入一個核進行運算，但是傳入值時要注意是可迭代的，所以在傳入值后需要加逗號, 同時需要用get()方法獲取返回值

def multicore():
    pool = mp.Pool() 
    res = pool.map(job, range(10))
    print(res)
    res = pool.apply_async(job, (2,))
    # 用get獲得結果
    print(res.get())

運行結果；

[0, 1, 4, 9, 16, 25, 36, 49, 64, 81]  # map()
4 # apply_async()

Pool默認調用是CPU的核數，傳入processes參數可自定義CPU核數

map() 放入迭代參數，返回多個結果

apply_async()只能放入一組參數，并返回一個結果，如果想得到map()的效果需要通過迭代

這節我們學習如何定義共享內存。只有用共享內存才能讓CPU之間有交流。

我們可以通過使用Value數據存儲在一個共享的內存表中。

import multiprocessing as mp

value1 = mp.Value("i", 0) 
value2 = mp.Value("d", 3.14)

其中d和i參數用來設置數據類型的，d表示一個雙精浮點類型 double，i表示一個帶符號的整型。

在Python的 mutiprocessing 中，有還有一個Array類，可以和共享內存交互，來實現在進程之間共享數據。

array = mp.Array("i", [1, 2, 3, 4])

這里的Array和numpy中的不同，它只能是一維的，不能是多維的。同樣和Value 一樣，需要定義數據形式，否則會報錯。 我們會在后一節舉例說明這兩種的使用方法.

錯誤形式

array = mp.Array("i", [[1, 2], [3, 4]]) # 2維list

"""
TypeError: an integer is required
"""

讓我們看看沒有加進程鎖時會產生什么樣的結果。

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_no_lock.py
@time: 18/8/26 09:22
"""

import multiprocessing as mp
import time

def job(v, num):
    for _ in range(5):
        time.sleep(0.5) # 暫停0.5秒，讓輸出效果更明顯
        v.value += num  # v.value獲取共享變量值
        print(v.value)

def multicore():
    v = mp.Value("i", 0)  # 定義共享變量
    p1 = mp.Process(target=job, args=(v, 1))
    p2 = mp.Process(target=job, args=(v, 4)) # 設定不同的number看如何搶奪內存
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == "__main__":
    multicore()

在上面的代碼中，我們定義了一個共享變量v，兩個進程都可以對它進行操作。 在job()中我們想讓v每隔0.1秒輸出一次累加num的結果，但是在兩個進程p1和p2 中設定了不同的累加值。所以接下來讓我們來看下這兩個進程是否會出現沖突。

結果打?。?/p>

?  baseLearn python ./process/process_no_lock.py
1
5
9
9
13
13
17
17
18
18
?  baseLearn

我們可以看到，進程1和進程2在相互搶著使用共享內存v。

為了解決上述不同進程搶共享資源的問題，我們可以用加進程鎖來解決。

首先需要定義一個進程鎖

 l = mp.Lock() # 定義一個進程鎖

然后將進程鎖的信息傳入各個進程中

p1 = mp.Process(target=job, args=(v,1,l)) # 需要將Lock傳入
p2 = mp.Process(target=job, args=(v,3,l)) 

在job()中設置進程鎖的使用，保證運行時一個進程的對鎖內內容的獨占

def job(v, num, l):
    l.acquire() # 鎖住
    for _ in range(5):
        time.sleep(0.1) 
        v.value += num # v.value獲取共享內存
        print(v.value)
    l.release() # 釋放

全部代碼：

# -*- coding:utf-8 -*-

"""
@author: Corwien
@file: process_lock.py
@time: 18/8/26 09:22
"""

import multiprocessing as mp
import time

def job(v, num, l):
    l.acquire() # 鎖住
    for _ in range(5):
        time.sleep(0.5) # 暫停0.5秒，讓輸出效果更明顯
        v.value += num  # v.value獲取共享變量值
        print(v.value)
    l.release() # 釋放

def multicore():
    l = mp.Lock() # 定義一個進程鎖
    v = mp.Value("i", 0)  # 定義共享變量
    p1 = mp.Process(target=job, args=(v, 1, l)) # 需要將lock傳入
    p2 = mp.Process(target=job, args=(v, 4, l)) # 設定不同的number看如何搶奪內存
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == "__main__":
    multicore()

運行一下，讓我們看看是否還會出現搶占資源的情況：

結果打?。?/p>

?  baseLearn python ./process/process_lock.py
1
2
3
4
5
9
13
17
21
25

顯然，進程鎖保證了進程p1的完整運行，然后才進行了進程p2的運行