摘要：本節(jié)講學(xué)習(xí)的多進(jìn)程。和之前的的不同點(diǎn)是丟向的函數(shù)有返回值，而的沒(méi)有返回值。所以接下來(lái)讓我們來(lái)看下這兩個(gè)進(jìn)程是否會(huì)出現(xiàn)沖突。

本節(jié)講學(xué)習(xí)Python的多進(jìn)程。

多進(jìn)程 Multiprocessing 和多線程 threading 類似, 他們都是在 python 中用來(lái)并行運(yùn)算的. 不過(guò)既然有了 threading, 為什么 Python 還要出一個(gè) multiprocessing 呢? 原因很簡(jiǎn)單, 就是用來(lái)彌補(bǔ) threading 的一些劣勢(shì), 比如在 threading 教程中提到的GIL.

使用 multiprocessing 也非常簡(jiǎn)單, 如果對(duì) threading 有一定了解的朋友, 你們的享受時(shí)間就到了. 因?yàn)?python 把 multiprocessing 和 threading 的使用方法做的幾乎差不多. 這樣我們就更容易上手. 也更容易發(fā)揮你電腦多核系統(tǒng)的威力了!

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()

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

在運(yùn)用時(shí)需要添加上一個(gè)定義main函數(shù)的語(yǔ)句

if __name__=="__main__":

完整的應(yīng)用代碼：

# -*- 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()

運(yùn)行環(huán)境要在terminal環(huán)境下，可能其他的編輯工具會(huì)出現(xiàn)運(yùn)行結(jié)束后沒(méi)有打印結(jié)果，在terminal中的運(yùn)行后打印的結(jié)果為：

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

Queue的功能是將每個(gè)核或線程的運(yùn)算結(jié)果放在隊(duì)里中， 等到每個(gè)線程或核運(yùn)行完畢后再?gòu)年?duì)列中取出結(jié)果， 繼續(xù)加載運(yùn)算。原因很簡(jiǎn)單, 多線程調(diào)用的函數(shù)不能有返回值, 所以使用Queue存儲(chǔ)多個(gè)線程運(yùn)算的結(jié)果

process_queue.py

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

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

import multiprocessing as mp

# 定義一個(gè)被多線程調(diào)用的函數(shù)，q 就像一個(gè)隊(duì)列，用來(lái)保存每次函數(shù)運(yùn)行的結(jié)果
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,))

    # 分別啟動(dòng)、連接兩個(gè)線程
    p1.start()
    p2.start()
    p1.join()
    p2.join()

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

    print res1,res2

打印輸出結(jié)果：

? python ./process/process_queue.py
249833583000 249833583000

進(jìn)程池就是我們將所要運(yùn)行的東西，放到池子里，Python會(huì)自行解決多進(jìn)程的問(wèn)題。

首先import multiprocessing 和定義job()

import multiprocessing as mp

def job(x):
    return x*x

然后我們定義一個(gè)Pool

pool = mp.Pool()

有了池子之后，就可以讓池子對(duì)應(yīng)某一個(gè)函數(shù)，我們向池子里丟數(shù)據(jù)，池子就會(huì)返回函數(shù)返回的值。 Pool和之前的Process的不同點(diǎn)是丟向Pool的函數(shù)有返回值，而Process的沒(méi)有返回值。

接下來(lái)用map()獲取結(jié)果，在map()中需要放入函數(shù)和需要迭代運(yùn)算的值，然后它會(huì)自動(dòng)分配給CPU核，返回結(jié)果

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

讓我們來(lái)運(yùn)行一下

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  # 注意這里的函數(shù)有return返回值

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

執(zhí)行結(jié)果：

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

我們?cè)趺粗?b>Pool是否真的調(diào)用了多個(gè)核呢？我們可以把迭代次數(shù)增大些，然后打開(kāi)CPU負(fù)載看下CPU運(yùn)行情況

打開(kāi)CPU負(fù)載(Mac)：活動(dòng)監(jiān)視器 > CPU > CPU負(fù)載(單擊一下即可)

Pool默認(rèn)大小是CPU的核數(shù)，我們也可以通過(guò)在Pool中傳入processes參數(shù)即可自定義需要的核數(shù)量

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

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

apply_async()中只能傳遞一個(gè)值，它只會(huì)放入一個(gè)核進(jìn)行運(yùn)算，但是傳入值時(shí)要注意是可迭代的，所以在傳入值后需要加逗號(hào), 同時(shí)需要用get()方法獲取返回值

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

運(yùn)行結(jié)果；

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

Pool默認(rèn)調(diào)用是CPU的核數(shù)，傳入processes參數(shù)可自定義CPU核數(shù)

map() 放入迭代參數(shù)，返回多個(gè)結(jié)果

apply_async()只能放入一組參數(shù)，并返回一個(gè)結(jié)果，如果想得到map()的效果需要通過(guò)迭代

這節(jié)我們學(xué)習(xí)如何定義共享內(nèi)存。只有用共享內(nèi)存才能讓CPU之間有交流。

我們可以通過(guò)使用Value數(shù)據(jù)存儲(chǔ)在一個(gè)共享的內(nèi)存表中。

import multiprocessing as mp

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

其中d和i參數(shù)用來(lái)設(shè)置數(shù)據(jù)類型的，d表示一個(gè)雙精浮點(diǎn)類型 double，i表示一個(gè)帶符號(hào)的整型。

在Python的 mutiprocessing 中，有還有一個(gè)Array類，可以和共享內(nèi)存交互，來(lái)實(shí)現(xiàn)在進(jìn)程之間共享數(shù)據(jù)。

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

這里的Array和numpy中的不同，它只能是一維的，不能是多維的。同樣和Value 一樣，需要定義數(shù)據(jù)形式，否則會(huì)報(bào)錯(cuò)。 我們會(huì)在后一節(jié)舉例說(shuō)明這兩種的使用方法.

錯(cuò)誤形式

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

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

讓我們看看沒(méi)有加進(jìn)程鎖時(shí)會(huì)產(chǎn)生什么樣的結(jié)果。

# -*- 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)) # 設(shè)定不同的number看如何搶奪內(nèi)存
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == "__main__":
    multicore()

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

結(jié)果打?。?/p>

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

我們可以看到，進(jìn)程1和進(jìn)程2在相互搶著使用共享內(nèi)存v。

為了解決上述不同進(jìn)程搶共享資源的問(wèn)題，我們可以用加進(jìn)程鎖來(lái)解決。

首先需要定義一個(gè)進(jìn)程鎖

 l = mp.Lock() # 定義一個(gè)進(jìn)程鎖

然后將進(jìn)程鎖的信息傳入各個(gè)進(jìn)程中

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

在job()中設(shè)置進(jìn)程鎖的使用，保證運(yùn)行時(shí)一個(gè)進(jìn)程的對(duì)鎖內(nèi)內(nèi)容的獨(dú)占

def job(v, num, l):
    l.acquire() # 鎖住
    for _ in range(5):
        time.sleep(0.1) 
        v.value += num # v.value獲取共享內(nèi)存
        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() # 定義一個(gè)進(jìn)程鎖
    v = mp.Value("i", 0)  # 定義共享變量
    p1 = mp.Process(target=job, args=(v, 1, l)) # 需要將lock傳入
    p2 = mp.Process(target=job, args=(v, 4, l)) # 設(shè)定不同的number看如何搶奪內(nèi)存
    p1.start()
    p2.start()
    p1.join()
    p2.join()

if __name__ == "__main__":
    multicore()

運(yùn)行一下，讓我們看看是否還會(huì)出現(xiàn)搶占資源的情況：

結(jié)果打?。?/p>

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

顯然，進(jìn)程鎖保證了進(jìn)程p1的完整運(yùn)行，然后才進(jìn)行了進(jìn)程p2的運(yùn)行