行程

Python 中的多線程其實不是真正的多線程，如果想要充分地使用多核心 CPU 的資源，在 Python 中大部分情況都需要使用多進程。 Python 提供了非常好用的多進程套件 multiprocessing，只需要定義一個函數，Python 會完成其他所有事情。借助這個套件，可以輕鬆完成從單一進程到並發執行的轉換。 multiprocessing 支援子進程、通訊和共享資料、執行不同形式的同步，提供了 Process、Queue、Pipe、Lock 等元件。

1、類別Process

建立流程的類別：Process([group [, target [, name [, args [, kwargs]]]]])

target 表示呼叫物件

args 表示呼叫物件的位置參數元組

#kwargs表示呼叫物件的字典

name為別名

group實質上不使用

下面看一個創建函數並將其作為多個進程的例子：

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
import multiprocessing
import time
def worker(interval, name):
    print(name + '【start】')
    time.sleep(interval)
    print(name + '【end】')
if __name__ == "__main__":
    p1 = multiprocessing.Process(target=worker, args=(2, '两点水1'))
    p2 = multiprocessing.Process(target=worker, args=(3, '两点水2'))
    p3 = multiprocessing.Process(target=worker, args=(4, '两点水3'))
    p1.start()
    p2.start()
    p3.start()
    print("The number of CPU is:" + str(multiprocessing.cpu_count()))
    for p in multiprocessing.active_children():
        print("child   p.name:" + p.name + "\tp.id" + str(p.pid))
    print("END!!!!!!!!!!!!!!!!!")

輸出的結果：

多進程輸出結果

2、把進程創建成類

當然我們也可以把進程創建成一個類，如下面的例子，當進程p 呼叫start() 時，自動調用run() 方法。

# -*- coding: UTF-8 -*-
import multiprocessing
import time
class ClockProcess(multiprocessing.Process):
    def __init__(self, interval):
        multiprocessing.Process.__init__(self)
        self.interval = interval
    def run(self):
        n = 5
        while n > 0:
            print("当前时间: {0}".format(time.ctime()))
            time.sleep(self.interval)
            n -= 1
if __name__ == '__main__':
    p = ClockProcess(3)
    p.start()

輸出結果如下：

建立進程類別

#3、daemon 屬性

想知道daemon 屬性有什麼用，看下面兩個例子吧，一個加了daemon 屬性，一個沒有加，對比輸出的結果：

沒有加deamon 屬性的例子：

# -*- coding: UTF-8 -*-
import multiprocessing
import time
def worker(interval):
    print('工作开始时间：{0}'.format(time.ctime()))
    time.sleep(interval)
    print('工作结果时间：{0}'.format(time.ctime()))
if __name__ == '__main__':
    p = multiprocessing.Process(target=worker, args=(3,))
    p.start()
    print('【EMD】')

輸出結果：

【EMD】
工作开始时间：Mon Oct  9 17:47:06 2017
工作结果时间：Mon Oct  9 17:47:09 2017

在上面範例中，進程p 新增daemon 屬性：

# -*- coding: UTF-8 -*-
import multiprocessing
import time
def worker(interval):
    print('工作开始时间：{0}'.format(time.ctime()))
    time.sleep(interval)
    print('工作结果时间：{0}'.format(time.ctime()))
if __name__ == '__main__':
    p = multiprocessing.Process(target=worker, args=(3,))
    p.daemon = True
    p.start()
    print('【EMD】')

輸出結果：

【EMD】

根據輸出結果可見，如果在子進程中新增了daemon 屬性，那麼當主進程結束的時候，子進程也會跟著結束。所以沒有列印子進程的資訊。

4、join 方法

結合上面的範例繼續，如果我們想要讓子執行緒執行完該怎麼做呢？

那麼我們可以用到 join 方法，join 方法的主要作用是：阻塞目前進程，直到呼叫 join 方法的那個進程執行完，然後再繼續執行目前進程。

因此看下加了join 方法的例子：

import multiprocessing
import time
def worker(interval):
    print('工作开始时间：{0}'.format(time.ctime()))
    time.sleep(interval)
    print('工作结果时间：{0}'.format(time.ctime()))
if __name__ == '__main__':
    p = multiprocessing.Process(target=worker, args=(3,))
    p.daemon = True
    p.start()
    p.join()
    print('【EMD】')

輸出的結果：

工作开始时间：Tue Oct 10 11:30:08 2017
工作结果时间：Tue Oct 10 11:30:11 2017
【EMD】

5、Pool

如果需要很多的子進程，我們就需要一個一個的去創建嗎？

當然不用，我們可以使用進程池的方法批次建立子進程。

範例如下：

# -*- coding: UTF-8 -*-
from multiprocessing import Pool
import os, time, random
def long_time_task(name):
    print('进程的名称：{0} ；进程的PID: {1} '.format(name, os.getpid()))
    start = time.time()
    time.sleep(random.random() * 3)
    end = time.time()
    print('进程 {0} 运行了 {1} 秒'.format(name, (end - start)))
if __name__ == '__main__':
    print('主进程的 PID：{0}'.format(os.getpid()))
    p = Pool(4)
    for i in range(6):
        p.apply_async(long_time_task, args=(i,))
    p.close()
    # 等待所有子进程结束后在关闭主进程
    p.join()
    print('【End】')

輸出的結果如下：

主进程的 PID：7256
进程的名称：0 ；进程的PID: 1492
进程的名称：1 ；进程的PID: 12232
进程的名称：2 ；进程的PID: 4332
进程的名称：3 ；进程的PID: 11604
进程 2 运行了 0.6500370502471924 秒
进程的名称：4 ；进程的PID: 4332
进程 1 运行了 1.0830621719360352 秒
进程的名称：5 ；进程的PID: 12232
进程 5 运行了 0.029001712799072266 秒
进程 4 运行了 0.9720554351806641 秒
进程 0 运行了 2.3181326389312744 秒
进程 3 运行了 2.5331451892852783 秒
【End】

這裡有一點要注意： Pool 物件呼叫join() 方法會等待所有子程序執行完畢，呼叫join() 之前必須先呼叫close() ，呼叫close() 之後就不能繼續加入新的Process 了。

請注意輸出的結果，子進程0，1，2，3是立刻執行的，而子進程4 要等待前面某個子進程完成後才執行，這是因為Pool 的預設大小在我的電腦上是4，因此，最多同時執行4 個行程。這是 Pool 有意設計的限制，並不是作業系統的限制。如果改成：

p = Pool(5)

就可以同時跑 5 個進程。

6、進程間通訊

Process 之間一定是需要通訊的，作業系統提供了許多機制來實現進程間的通訊。 Python 的 multiprocessing 模組包裝了底層的機制，提供了Queue、Pipes 等多種方式來交換資料。

以 Queue 為例，在父進程中建立兩個子進程，一個往 Queue 寫數據，一個從 Queue 讀取資料：

#!/usr/bin/env python3
# -*- coding: UTF-8 -*-
from multiprocessing import Process, Queue
import os, time, random
def write(q):
    # 写数据进程
    print('写进程的PID:{0}'.format(os.getpid()))
    for value in ['两点水', '三点水', '四点水']:
        print('写进 Queue 的值为：{0}'.format(value))
        q.put(value)
        time.sleep(random.random())
def read(q):
    # 读取数据进程
    print('读进程的PID:{0}'.format(os.getpid()))
    while True:
        value = q.get(True)
        print('从 Queue 读取的值为：{0}'.format(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()

輸出的結果為：

读进程的PID:13208
写进程的PID:10864
写进 Queue 的值为：两点水
从 Queue 读取的值为：两点水
写进 Queue 的值为：三点水
从 Queue 读取的值为：三点水
写进 Queue 的值为：四点水
从 Queue 读取的值为：四点水