使用multiprocessing中的常見問題

在python的解釋器中，CPython是應用範圍最廣的一種，其具備豐富的擴展包，方便了開發者的使用。固然CPython也不是完美的，因爲全局解釋鎖（GIL）的存在，python的多線程能夠近似看做單線程。爲此，開發者推出了multiprocessing，這裏介紹一下使用中的常見問題。

環境

>>> import sys >>> print(sys.version) 3.6.0 |Anaconda 4.3.1 (64-bit)| (default, Dec 23 2016, 12:22:00) \n[GCC 4.4.7 20120313 (Red Hat 4.4.7-1)]

共享變量

任務可否切分紅多個子任務是判斷一個任務可否使用多進程或多線程的重要標準。在任務切分時，不可避免的須要數據通信，而共享變量是數據通信的重要方式。在multiprocess中，共享變量有兩種方式：Shared memory和Server process。

share memory

multiprocess經過Array和Value來共享內存

from multiprocessing import Array, Value num = 10 elements = Array("i", [2 * i + 1 for i in range(num)]) val = Value('d', 0.0)

而後就能夠將數據同步到Process中。這裏舉一個例子，即將elements翻倍，val增長1，首先定義函數

def func(elements, val): for i, ele in enumerate(elements): elements[i] = ele * 2 val.value += 1

再定義Process

from multiprocessing import Process p = Process(target=func, args=(elements, val, )) p.start() # 運行Process p.join() # 等待Process運行結束

最終運行結果

=====Process運行前=======
[elements]:1 3 5 7 9 11 13 15 17 19 [Value]:0.0 =====Process運行後======= [elements]:2 6 10 14 18 22 26 30 34 38 [Value]:1.0

在某些特定的場景下要共享string類型，方式以下：

from ctypes import c_char_p str_val = Value(c_char_p, b"Hello World")

關於Share Memory支持的更多類型，能夠查看module-multiprocessing.sharedctypes。

Server process

此種方式經過建立一個Server process來管理python object，而後其餘process經過代理來訪問這些python object。相較於share memory，它支持任意類型的共享，包括：list、dict、Namespace等。這裏以dict和list舉一個例子：

from multiprocessing import Process, Manager def func(d, l): d[1] = '1' d['2'] = 2 d[0.25] = None l.reverse() if __name__ == '__main__': with Manager() as manager: d = manager.dict() l = manager.list(range(10)) print("=====Process運行前=======") print(d) print(l) p = Process(target=func, args=(d, l)) p.start() p.join() print("=====Process運行後=======") print(d) print(l)

運行結果以下

=====Process運行前======= {} [0, 1, 2, 3, 4, 5, 6, 7, 8, 9] =====Process運行後======= {1: '1', '2': 2, 0.25: None} [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

進程間通信

衆所周知，併發編程中應該儘可能避免共享變量，多進程更是如此。在這種狀況下，多進程間的通信就要用到Queue和Pipe。

Queue

Queue是一種線程、進程安全的先進先出的隊列。使用中，首先定義Queue

from multiprocessing import Queue queue = Queue()

而後將須要處理的數據放入Queue中

elements = [i for i in range(100)] for i in elements: queue.put(i)

而後建立子進程process

from multiprocessing import Process process = Process(target=func, args=(queue, ))

其中func是子進程處理數據的邏輯。

from queue import Empty def func(queue): buff = [] while True: try: ele = queue.get(block=True, timeout=1) buff.append(str(ele)) except Empty: print(" ".join(buff)) print("Queue has been empty.....") break

使用queue.get時,若Queue中沒有數據，則會拋出queue.Empty錯誤。值得注意的是，在使用queue.get()時必定要設置block=True和timeout，不然它會一直等待，直到queue中放入數據（剛開始用的時候，我一直奇怪爲何程序一直處在等待狀態）。運行結果

=====單進程====== 0 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 Queue has been empty.....

Pipe

Pipe是一種管道，一端輸入，一端輸出。在multiprocess中，能夠經過Pipe()函數來定義，返回send和recv的connection。使用中，先定義

from multiprocessing import Pipe parent_conn, child_conn = Pipe()

而後一端放入數據，另外一端就能夠接受數據了

from multiprocessing import Process def f(conn): conn.send([42, None, 'hello']) conn.close() p = Process(target=f, args=(child_conn,)) p.start() print(parent_conn.recv()) p.join()

輸出結果

[42, None, 'hello']

另外，值得注意的是，若兩個或更多進程同時從管道一端讀或寫數據，會致使管道中的數據corrupt。爲了直觀的理解這種狀況，這裏舉一個例子，即在主進程將數據放入管道，在子進程從管道中讀出數據，並打印結果。區別之處在於，子進程的數量。首先將數據放入管道：

def func(conn): a = conn.recv() print(a) parent, child = Pipe() child.send("Hello world...")

而後開啓子進程

print("======單進程========") p = Process(target=func, args=(parent, )) p.start() p.join() print("======多進程========") num_process = 2 ps = [Process(target=func, args=(parent, )) for _ in range(num_process)] for p in ps: p.start() for p in ps: p.join()

輸出結果

多進程並未按照預想的輸出兩個Hello World，而是處於死鎖的狀態。

例子

關於Queue和Pipe的用法講了這麼多，下面作一個小練習，內容是：利用多線程從文件中讀取數據，處理後將數據保存到另一個文件中。具體方法以下： 
1. 開闢一個子進程，從文件中讀取數據，並將數據存入Queue中 
2. 開闢多個子進程，從Queue中讀取數據，處理數據，並將數據放入管道一端（注意加鎖） 
3. 開闢一個子進程，從管道另外一端獲取數據，並將數據寫入文件中

0.導包

from multiprocessing import Process, Array, Queue, Value, Pipe, Lock from queue import Empty import sys

1.讀取數據

def read_file(fin, work_queue): for line in fin: i = int(line.strip("\n")) work_queue.put_nowait(i)

其中work_queue用於連通「讀數據的進程」和「處理數據的進程」。

2.處理數據

def worker_func(work_queue, conn, lock, index): while True: try: ele = work_queue.get(block=True, timeout=0.5) + 1 with lock: conn.send(ele) except Empty: print("Process-{} finish...".format(index)) conn.send(-1) break

從隊列中讀取數據，直到隊列中的數據全被取走。當Queue中不存在數據時，向queue放入終止符（-1），告訴後面的進程，前面的人任務已經完成。

3.寫數據

def write_file(conn, fout, num_workers): record = 0 while True: val = conn.recv() if val == -1: record += 1 else: print(val, file=fout) fout.flush() if record == num_workers: break

當寫進程收到特定數量終止符（-1）時，寫進程就終止了。

4.執行

path_file_read = "./raw_data.txt" path_file_write = "./data.txt" with open(path_file_read) as fin, \ open(path_file_write, "w") as fout: queue = Queue() parent, child = Pipe() lock = Lock() read_Process = Process(target=read_file, args=(fin, queue, )) worker_Process = [Process(target=worker_func, args=(queue, parent, lock, index, )) for index in range(3)] write_Process = Process( target=write_file, args=(child, fout, len(worker_Process), )) read_Process.start() for p in worker_Process: p.start() write_Process.start() print("read....") read_Process.join() print("worker....") for p in worker_Process: p.join() print("write....") write_Process.join()

輸入/輸出

打印錯行

在使用多進程中，你會發現打印的結果發生錯行。這是由於python的print函數是線程不安全的，從而致使錯行。解決方法也很簡單，給print加一把鎖就行了，方式以下

from multiprocessing import Process, Lock def f(l, i): l.acquire() try: print('hello world', i) finally: l.release() if __name__ == '__main__': lock = Lock() for num in range(10): Process(target=f, args=(lock, num)).start()

沒法打印日誌信息

剛開始用多進程時，常常會出現日誌信息沒法打印的狀況。其實問題很簡單。在多進程中，打印內容會存在緩存中，直到達到必定數量纔會打印。解決這個問題，只須要加上

import sys sys.stdout.flush() sys.stderr.flush()

例如上面的例子，應該寫成

import sys def f(l, i): l.acquire() try: print('hello world', i) sys.stdout.flush() # 加入flush finally: l.release()

總結

以上就是我在使用multiprocessing遇到的問題。