[Python] multiprocessing 基本教學

程式語言：Python
Package：multiprocessing
官方文件

功能：並行處理
因 GIL (CPython) 緣故，multithread 需用 multiprocess 取代，可參考以下文章
Python 最難的問題
Python 的 GIL 是什么鬼，多线程性能究竟如何

注意事項
  • window 程式需在 if __name__ == '__main__': 之內運行
    不然會有 RuntimeError，甚至莫名的錯誤
    因 window 沒有 fork，所以執行時是採用 spawn，使用 runpy 實現
    所以每個 child process 會在一開始 import 原先的 module 再執行對應的 function
     參考 multiprocessing - Windows
    指定產生 process 的方式- Contexts and start methods
    1. from multiprocessing import Pool
    2. import sys
    3. import os
    4.  
    5.  
    6. def f(x):
    7.     return x
    8.  
    9.  
    10. if __name__ == '__main__':
    11.     # 依 CPU 數量建立 child process
    12.     pool = Pool()
    13.     pool.map(f, range(10))
    14.  
    15. # 因不在 '__main__' 中，所以 import 時每個 child process 都會執行
    16. try:
    17.     print('pid:{:4d} stdout:{:8d}, __name__:{:10s}, importedBy:{}'.format(os.getpid(), id(sys.stdout), __name__, sys._getframe(1).f_globals.get('__name__')))
    18. except ValueError:
    19.     print('pid:{:4d} stdout:{:8d}, __name__:{:10s}'.format(os.getpid(), id(sys.stdout), __name__))
    20.     
    21.     
    22. # 輸出結果
    23. # pid:5908 stdout: 4703408, __name__:__mp_main__, importedBy:runpy
    24. # pid:2296 stdout: 6866096, __name__:__mp_main__, importedBy:runpy
    25. # pid:2152 stdout:21546160, __name__:__main__
  • print 視需求加上 flush=True 強制刷新
    不然最後程式結束強制 flush 只是殘存在任一 sys.stdout 的東西
    1. from multiprocessing import Pool
    2. import time
    3. import sys
    4.  
    5.  
    6. def f(x):
    7.     time.sleep(0.1)
    8.     # 無換行，所以會先暫存在 sys.stdout，等待 flush
    9.     print(x, end=" ")
    10.     return x
    11.  
    12.  
    13. def g(x):
    14.     if x >=0 :
    15.         print(":g", id(sys.stdout), x)
    16.     return x
    17.  
    18.     
    19. if __name__ == '__main__':
    20.     # 可加入 maxtasksperchild 限定，因 process 結束時也會 flush sys.stdout
    21.     pool = Pool()
    22.     print(pool.map(f, range(10)))
    23.     
    24.     # 加上 close() & join() 可清空所有 sys.stdout
    25.     # pool.close()
    26.     # pool.join()    
    27.     
    28.     # 此行可強制輸出 sys.stdout 內的東西
    29.     # 因換行代表強制 flush
    30.     # print(pool.map(g, range(6)))
    31.     
    32. # 每個 process 可能有不同的 sys.stdout
    33. # 因不在 '__main__' 中，所以 import 時每個 child process 都會執行
    34. print("stdout:", id(sys.stdout))
    35.  
    36. # 輸出結果
    37. # stdout: 7324848
    38. # stdout: 4768944
    39. # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
    40. # stdout: 4900016
    41. # 0 1 2 3 6 7

建立 Process

Process
  1. import time
  2. from multiprocessing import Process, Lock, freeze_support, set_start_method
  3.  
  4. def f(lock, i):
  5.     print('{:2d} Stop'.format(i))
  6.     # 暫停 1s，這邊是為了看出其他 child process Stop 的效果，才暫停的
  7.     time.sleep(1)    
  8.     # block 直到上一個 child process 解鎖
  9.     lock.acquire()
  10.     try:
  11.         print('{:2d} Start'.format(i))
  12.     finally:
  13.         # 必做解鎖，不然會影響到下一個 child process
  14.         lock.release()
  15.  
  16.         
  17. # 因 window spawn 的緣故
  18. # 必須在 __name__ == '__main__' 之內執行
  19. if __name__ == '__main__':
  20.     # 在 window 中，像 PyInstaller 為了產生執行檔
  21.     # multiprocessing 會被禁止，導致無法執行，出現 RuntimeError
  22.     # 需在 start 前加入此 function，只在 window 有效，其餘無作用
  23.     # 且在 __name__ == '__main__' 之後
  24.     freeze_support()
  25.     # 指定產生 process 的方式，讓不同平台的表現一致
  26.     set_start_method('spawn')
  27.     lock = Lock()
  28.  
  29.     for num in range(10):
  30.         # 建立 child process
  31.         p = Process(target=f, args=(lock, num))        
  32.         # 開始執行 child process
  33.         p.start()
  34.         # blocking parent process 直到 child process 返回
  35.         # 等待 0.1s，可為無參數，則表示永久等待
  36.         p.join(timeout=0.1)
  37.         
  38. # 輸出結果，每次結果都不一樣
  39. # 可以看到並不是連續的，這是因為 child process 建立有快有慢
  40. # 且搶到 key 的，也不一定是按 stop 的順序，需注意
  41. # 0 Stop  
  42. # 1 Stop  
  43. # 3 Stop  
  44. # 2 Stop  
  45. # 5 Stop  
  46. # 4 Stop  
  47. # 0 Start 
  48. # 1 Start 
  49. # 7 Stop  
  50. # 2 Start 
  51. # 3 Start 
  52. # 9 Stop  
  53. # 8 Stop  
  54. # 6 Stop  
  55. # 5 Start 
  56. # 4 Start 
  57. # 7 Start 
  58. # 9 Start 
  59. # 8 Start 
  60. # 6 Start

Pool
多個 child process 建議用法 
  1. import os
  2. import time
  3. import traceback
  4. from multiprocessing import Pool
  5.  
  6.  
  7. def handle_error(e):
  8.     '''處理 child process 的錯誤，不然 code 寫錯時，不會回報任何錯誤'''
  9.     traceback.print_exception(type(e), e, e.__traceback__)
  10.     
  11.     
  12. def long_time_task(name):
  13.     print('任務 {} ({}) 開始'.format(name, os.getpid()))
  14.     start = time.time()
  15.     time.sleep(3)
  16.     end = time.time()
  17.     print('任務 {} 執行 {:0.2f} seconds.'.format(name, (end - start)))
  18.  
  19.  
  20. # 因 window spawn 的緣故
  21. # 必須在 __name__ == '__main__' 之內執行
  22. if __name__=='__main__':
  23.     print('Parent process {}.'.format(os.getpid()))
  24.     # 指定建立 child process 的數量，若無指定，預設為 cpu 數量
  25.     with Pool(processes=3) as p:
  26.         for i in range(5):
  27.             # 建立 child process 並執行，error_callback 必須指定，不然很難 debug
  28.             p.apply_async(long_time_task, args=(i,), error_callback=handle_error)
  29.         print('等待所有 child processes 完成')
  30.         # 關掉 pool 不再加入任何 child process
  31.         p.close()
  32.         # 調用 join() 之前必須先調用close()
  33.         p.join()
  34.     print('所有 child processes 完成')
  35.  
  36. # 執行結果    
  37. # Parent process 6072.      
  38. # 等待所有 child processes 完成   
  39. # 任務 0 (5192) 開始            
  40. # 任務 1 (6040) 開始            
  41. # 任務 2 (4576) 開始            
  42. # 任務 0 執行 3.01 seconds.     
  43. # 任務 3 (5192) 開始            
  44. # 任務 1 執行 3.01 seconds.     
  45. # 任務 4 (6040) 開始            
  46. # 任務 2 執行 3.01 seconds.     
  47. # 任務 3 執行 3.00 seconds.     
  48. # 任務 4 執行 3.00 seconds.     
  49. # 所有 child processes 完成

回傳資料

Pool
  1. from multiprocessing import Pool, TimeoutError
  2. import time
  3.  
  4.  
  5. def f(x):
  6.     return x*x
  7.  
  8.  
  9. # 因 window spawn 的緣故
  10. # 必須在 __name__ == '__main__' 之內執行    
  11. if __name__ == '__main__':
  12.     # start 4 worker processes
  13.     with Pool(processes=4) as pool:
  14.  
  15.         # 建立 child process 並執行
  16.         resA = pool.apply_async(f, (20,))
  17.  
  18.         # get timeout
  19.         resB = pool.apply_async(time.sleep, (10,))
  20.         try:
  21.             print(resB.get(timeout=1))
  22.         except TimeoutError:
  23.             print("得 resB 值超時")
  24.             
  25.         print("pool 此時仍可使用")
  26.         
  27.     # 已跳出 with，故 pool 已被 close
  28.     print("pool 已不可使用")
  29.     # 得到回傳值
  30.     print('resA', resA.get(timeout=1)) # resA 400

資料交換

資料不會被更改，只是單純的再複製一份
Queues
  1. from multiprocessing import Process, Queue
  2.  
  3.  
  4. def f(q):
  5.     q.put([42, None, 'hello'])
  6.     q.put(2)
  7.  
  8.  
  9. # 因 window spawn 的緣故
  10. # 必須在 __name__ == '__main__' 之內執行
  11. if __name__ == '__main__':
  12.     # FIFO
  13.     q = Queue()
  14.     p = Process(target=f, args=(q,))
  15.     p.start()
  16.     print(q.get())    # [42, None, 'hello']
  17.     print(q.get())    # 2
  18.     p.join()

Pipes 
  1. from multiprocessing import Process, Pipe
  2.  
  3.  
  4. def f(conn):
  5.     print('In f')
  6.     # 從 a 接收資料
  7.     print(conn.recv())   # 2
  8.     # 送出資料給 a
  9.     conn.send([42, None, 'hello'])
  10.     # 關閉 child process 的 b 端口，但不影響 parenet process 的 b 端口
  11.     conn.close()
  12.     print('b_conn close', conn.closed)
  13.     print('Out f')
  14.  
  15.  
  16. # 因 window spawn 的緣故
  17. # 必須在 __name__ == '__main__' 之內執行    
  18. if __name__ == '__main__':
  19.     # 會回傳兩個端點，如同水管的兩端，可互相溝通
  20.     a_conn, b_conn = Pipe()
  21.     # 送出資料給 b
  22.     a_conn.send(2)
  23.     # 建立 child process
  24.     p = Process(target=f, args=(b_conn,))
  25.     # 執行 child process
  26.     p.start()
  27.     # 從 b 接收資料
  28.     print(a_conn.recv())   # [42, None, 'hello']
  29.     # 等待直到 child process 完成
  30.     p.join()
  31.     # 在 child process 關閉，並不影響 parent process 的端口
  32.     print('b_conn close', b_conn.closed)
  33.     
  34.  
  35. # 輸出結果
  36. # In f
  37. # 2
  38. # b_conn close True
  39. # Out f
  40. # [42, None, 'hello']
  41. # b_conn close False

資料分享

資料會被更改
Shared memory
  1. from multiprocessing import Process, Value, Array
  2.  
  3.  
  4. def f(n, a):
  5.     n.value = 3.1415927
  6.     for i, x in enumerate(a):
  7.         a[i] = -x
  8.  
  9.  
  10. # 因 window spawn 的緣故
  11. # 必須在 __name__ == '__main__' 之內執行        
  12. if __name__ == '__main__':
  13.     num = Value('d', 0.0) # 0.0
  14.     arr = Array('i', range(10)) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  15.     print(num.value)
  16.     print(arr[:])
  17.     
  18.     # 建立 child process
  19.     p = Process(target=f, args=(num, arr))
  20.     # 執行 child process
  21.     p.start()
  22.     # 等待直到 child process 完成
  23.     p.join()
  24.  
  25.     print(num.value) # 3.1415927
  26.     print(arr[:]) # [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

Server process
速度較慢，但可通過網路分享
  1. from multiprocessing import Process, Manager
  2.  
  3.  
  4. def f(d, l):
  5.     d[1] = '1'
  6.     d['2'] = 2
  7.     d[0.25] = None
  8.     l.reverse()
  9.  
  10.  
  11. # 因 window spawn 的緣故
  12. # 必須在 __name__ == '__main__' 之內執行
  13. if __name__ == '__main__':
  14.     with Manager() as manager:
  15.         d = manager.dict()
  16.         l = manager.list(range(10))
  17.         print(d) # {}
  18.         print(l) # [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
  19.  
  20.         # 建立 child process
  21.         p = Process(target=f, args=(d, l))
  22.         # 執行 child process
  23.         p.start()
  24.         # 等待直到 child process 完成
  25.         p.join()
  26.  
  27.         print(d) # {0.25: None, 1: '1', '2': 2}
  28.         print(l) # [9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

class multiprocessing.Process

lass multiprocessing.Process(group=None, target=None, name=None, args=(), kwargs={}, *, daemon=None)
  • 參數
    • group
      • 必須為 None
    • target
      • 目標函數
    • name
      • process 的名字
    • args
      • 傳入的參數
    • kwargs
      • 傳入的 dict 參數
    • daemon
      • True
        • parent process 結束時，同時終止所有 child process
  • 屬性
    • name
      • child process 名字，只單純用來給人看
      • 預設為 ‘Process-N1:N2:...:Nk‘ ，k 代表第幾個 child process
    • daemon
      • 當結束時，是否隨 parent process 結束
      • 需在 start() 前設定
    • pid
      • process ID
    • exitcode
      • 終止的 code
      • child 未結束時仍是 None
    • authkey
      • parent process 的 身份驗證 key
    • sentinel
      • 數字表示的句柄。可以使用在系統等待函數裡
  • 方法
    • run()
      • 運行的主程式
    • start()
      • 啟動，並呼叫 run()
    • join([timeout])
      • 持續等待直到 child process 完成
      • process 不該呼叫自已的 join，會產生 deadlock
      • timeout (秒)
    • is_alive()
      • process 是否存活
    • terminate()
      • 終止 process
      • 不準在以下情況終止
        • 正在操作資料
        • 未解鎖就終止
  1. import multiprocessing, time, signal
  2.  
  3.  
  4. # 因 window spawn 的緣故
  5. # 必須在 __name__ == '__main__' 之內執行
  6. if __name__ == '__main__':
  7.     # 建立
  8.     p = multiprocessing.Process(target=time.sleep, args=(1000,))
  9.     print(p, p.is_alive()) # Process(Process-1, initial)> False
  10.  
  11.     # 啟動
  12.     p.start()
  13.     print(p, p.is_alive()) # Process(Process-1, started)> True
  14.  
  15.     # 終止
  16.     p.terminate()
  17.     # 等一下，以確定終止完成
  18.     time.sleep(0.1)
  19.     print(p, p.is_alive()) # Process(Process-1, stopped[SIGTERM])> False
  20.     print(p.exitcode == -signal.SIGTERM) # True
  21.  
  22. # 輸出結果
  23. # <Process(Process-1, initial)> False
  24. # <Process(Process-1, started)> True
  25. # <Process(Process-1, stopped[SIGTERM])> False
  26. # True

class multiprocessing.Queue([maxsize])

  • 多 process 溝通的建議首選
  • 參數
    • maxsize
      • queue 的大小
      • 設為 5 表示可放入 5 個 objects
  • 方法
    • 因為 multiprocessing 的關係，跟數量有關係的都只是預估值，並不可靠
    • qsize()
      • 目前的 objects 數量
    • empty()
      • 是否為空
    • full()
      • 是否已滿
    • put(obj[, block=True[, timeout=None]])
      • obj
        • 放入的物件
      • block
        • 持續等待直到物件被取走
      • Timeout
        • 等待時間(秒)
    • put_nowait(obj)
      • 等同 put(obj, False)
    • get([block=True[, timeout=None]])
      • block
        • 持續等待直到物件可領取
      • Timeout
        • 等待時間(秒)
    • get_nowait()
      • 等同 get(False)
    • close()
      • 關閉 queue
    • join_thread()
      • 確認所有的 data 都已被清除，再退出 process
      • 必須使用在 close() 後面
    • cancel_join_thread()
      • 不清除 data 直接退出 process
      • 必須使用在 close() 後面 
  1. from multiprocessing import Queue
  2.  
  3.  
  4. # 因 window spawn 的緣故
  5. # 必須在 __name__ == '__main__' 之內執行
  6. if __name__ == '__main__':
  7.     # 大小設定為 5 個 objects
  8.     q = Queue(5)
  9.     print(q.empty()) # True
  10.     print(q.qsize()) # 0
  11.     
  12.     q.put([1,2,3], False)
  13.     q.put(1, False)
  14.     q.put('123', False)
  15.     q.put({1,2,3}, False)
  16.     q.put({'list':[1,2,3], 'dict':{'1':1, '2':2}}, False)
  17.     
  18.     print(q.qsize()) # 5
  19.     print(q.full()) # True
  20.     
  21.     # 關閉，並等到清除資料後才退出 process
  22.     q.close()
  23.     q.join_thread()

multiprocessing.Pipe([duplex=True])

  • 回傳 Connection objects (conn1, conn2)
  • 參數
    • duplex
      • 是否為雙向
      • False
        • conn1 只能接收
        • conn2 只能傳送

class multiprocessing.Connection

  • 方法
    • recv()
      • 持續等待直到接收到 object
    • fileno()
      • 回傳 file descriptor or handle used by the connection.
    • close()
      • 關閉連接
    • poll([timeout])
      • 是否有任何資料可讀取
      • timeout (秒)
    • send_bytes(buffer[, offset[, size]])
      • 傳送 bytes
      • buffer
        • bytes-like object
      • offset
        • 從第幾個 byte 開始傳送
      • size
        • 限制傳送的 size
    • recv_bytes([maxlength])
      • 持續等待直到接收到 bytes
      • maxlength
        • 最大長度，需大於接受的長度
    • recv_bytes_into(buffer[, offset])
      • 持續等待直到接收到 bytes 並放入 buffer
      • buffer
        • 放入的 buffer
      • offset
        • 從第幾個 byte 開始填入
  1. from multiprocessing import Pipe
  2.  
  3.  
  4. a, b = Pipe()
  5. a.send([1, 'hello', None])
  6. print(b.poll()) # True
  7. print(b.recv()) #[1, 'hello', None]
  8. b.send_bytes(b'thank you')
  9. print(a.recv_bytes()) # b'thank you'
  10.  
  11. import array
  12. # 建立 array，格式為 byte
  13. arr1 = array.array('i', range(5))
  14. arr2 = array.array('i', [0] * 10)
  15. a.send_bytes(arr1)
  16. # 從第三個開始填入
  17. count = b.recv_bytes_into(arr2, 3*arr2.itemsize)
  18. # 確認收到的 bytes 一致
  19. assert count == len(arr1) * arr1.itemsize
  20. print(arr2) # array('b', [0, 0, 0, 0, 1, 2, 3, 4, 0, 0])

class multiprocessing.pool.Pool

class multiprocessing.pool.Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])
  • 參數
    • porcesses
      • 執行 process 的上限
      • 預設為 cpu 個數
    • initializer
      • 初始函數
      • 每個 process 啟動前呼叫 initializer(*initargs) 初始化
    • initargs
      • 初始函數的參數
    • maxtasksperchild
      • 完成多少數量的任務後需重啟 process
      • 避免運行時間很長的工作進程消耗太多的系統資源
      1. from multiprocessing import Pool
      2. import os
      3.  
      4.  
      5. def f():
      6.     print("PID: %d" % os.getpid())
      7.     
      8.  
      9. # 因 window spawn 的緣故
      10. # 必須在 __name__ == '__main__' 之內執行
      11. if __name__ == '__main__':
      12.     pool = Pool(1, maxtasksperchild=2)
      13.     
      14.     for _ in range(5):
      15.         pool.apply_async(f)
      16.     
      17.     # 關閉 pool
      18.     pool.close()
      19.     # 需關閉後才能 join()
      20.     # 不加此行，會看不到結果，因為會隨 parent process 關閉
      21.     pool.join()
      22.  
      23. # 輸出結果
      24. # PID: 5028
      25. # PID: 5028
      26. # PID: 4284
      27. # PID: 4284
      28. # PID: 5712
    • context
      • 可用來定義 context
  • 方法
    • 回傳的物件
      • blocking
        • 即是目標函數所回傳的物件
          內部程式碼為 return multiprocessing.pool.AsyncResult.get()
          因為 get() 故為 blocking
      • nonblocking
        • 即是 multiprocessing.pool.AsyncResult
          內部程式碼為 return multiprocessing.pool.AsyncResult
    • apply(func[, args[, kwds]])
      • 建立 child process 並執行
      • blocking，需等到完成才會進行下一步
    • apply_async(func[, args[, kwds[, callback[, error_callback]]]])
      • 建立 child process 並執行
      • nonblocking，直接進行下一步
      • callback
        • 處理回傳資料的函數
        • 必須夠快，不然會影響運作
      • error_callback
        • 處理錯誤的函數 
      1. from multiprocessing import Pool
      2. import traceback
      3.  
      4.  
      5. def f(x):
      6.     return x*x
      7.  
      8.     
      9. def show(x):
      10.     print(x+1, 'In show')
      11.  
      12.     
      13. def handle_error(e):
      14.     traceback.print_exception(type(e), e, e.__traceback__)
      15.     
      16.  
      17. # 因 window spawn 的緣故
      18. # 必須在 __name__ == '__main__' 之內執行    
      19. if __name__ == '__main__':
      20.     with Pool() as pool:
      21.         result = pool.apply_async(f, (10,), callback=show, error_callback=handle_error)
      22.         print(result.get(timeout=1))
      23.  
      24.         
      25. # 輸出結果
      26. # 101 In show
      27. # 100
    • map(func, iterable[, chunksize])
      • 同內建的 map，但只支援單一參數的 function
      • blocking，直到完成才會進行下一步
      • chunksize
        • 一次處理的數目
      1. from multiprocessing import Pool
      2. import time
      3.  
      4.  
      5. def f(x):
      6.     # 稍微暫停才看得出差異
      7.     time.sleep(0.1)
      8.     print(x)
      9.  
      10.     
      11. if __name__ == '__main__':
      12.     pool = Pool(2)
      13.     # chunksize 若改為 20，則會按順序輸出，因同時處理 20筆
      14.     pool.map(f, range(20), chunksize=10)
      15.     
      16.  
      17. # 輸出結果，因個別交錯輸出
      18. # 0
      19. # 10
      20. # 1
      21. # 11
      22. # 2
      23. # 12
      24. # 3
      25. # 13
      26. # 4
      27. # 14
      28. # 5
      29. # 15
      30. # 6
      31. # 16
      32. # 7
      33. # 17
      34. # 8
      35. # 18
      36. # 9
      37. # 19
    • map_async(func, iterable[, chunksize[, callback[, error_callback]]])
      • 參數同 apply_async & map
      • nonblocking
    • imap(func, iterable[, chunksize])
      • iterator 版的 map
      • blocking
    • imap_unordered(func, iterable[, chunksize])
      • 無序的 imap
      • blocking
    • starmap(func, iterable[, chunksize])
      • 類似 map，但傳入的參數視為 *x
      • blocking，直到完成才會進行下一步
      1. from multiprocessing import Pool
      2. import time
      3.  
      4.  
      5. def f(x, y):
      6.     print(x, y)
      7.  
      8.     
      9. if __name__ == '__main__':
      10.     pool = Pool()
      11.     pool.starmap(f, zip(range(20), range(20,40)), chunksize=1)
      12.  
      13.     
      14. # 輸出結果
      15. # 0 20 
      16. # 1 21 
      17. # 2 22 
      18. # 3 23 
      19. # 4 24 
      20. # 5 25 
      21. # 6 26 
      22. # 7 27 
      23. # 8 28 
      24. # 10 30
      25. # 11 31
      26. # 9 29 
      27. # 12 32
      28. # 13 33
      29. # 15 35
      30. # 14 34
      31. # 16 36
      32. # 17 37
      33. # 18 38
      34. # 19 39
    • starmap_async(func, iterable[, chunksize[, callback[, error_back]]])
      • 參數同 apply_async & map
      • nonblocking，直接執行下一步
    • close()
      • 等到所有任務完成，process 將退出，並關閉 Pool
    • terminate()
      • 立刻終止，無需等待任務完成
    • join()
      • 等待 child process 結束
      • 需先呼叫 close() 或 terminate()

class multiprocessing.pool.AsyncResult

  • 得 nonblocking 的回傳結果，例：Pool.apply_async() 和 Pool.map_async()
  • 方法
    • get([timeout])
      • 持續等待直到得到回傳的值
      • timeout (秒)
    • wait([timeout])
      • 持續等待直到回傳的值可用
    • ready()
      • child process 是否已完成
    • successful()
      • 同 ready()，但未完成會 raise AssertionError

Logging

  • debug 使用
  • multiprocessing.get_logger()
    • 回傳 logger
  • multiprocessing.log_to_stderr()
    • 呼叫 get_logger() 並回傳 logger
    • 同時加入處理程序，將訊息發送到 sys.stderr
      格式 '[%(levelname)s /%(processName)s] %(message)s' 
  1. import multiprocessing, logging
  2.  
  3.  
  4. logger = multiprocessing.log_to_stderr()
  5. logger.setLevel(logging.INFO)
  6.  
  7. # 因 window spawn 的緣故
  8. if __name__ == '__main__':
  9.     m = multiprocessing.Manager()
  10.     del m
  11.     
  12. # 輸出結果
  13. # [INFO/SyncManager-1] child process calling self.run()
  14. # [INFO/SyncManager-1] child process calling self.run()
  15. # [INFO/SyncManager-1] manager serving at '\\\\.\\pipe\\pyc-5984-0-ekn9ajo6'
  16. # [INFO/MainProcess] sending shutdown message to manager
  17. # [INFO/SyncManager-1] manager serving at '\\\\.\\pipe\\pyc-5984-0-ekn9ajo6'
  18. # [INFO/SyncManager-1] process shutting down
  19. # [INFO/SyncManager-1] process shutting down
  20. # [INFO/SyncManager-1] process exiting with exitcode 0
  21. # [INFO/SyncManager-1] process exiting with exitcode 0
  22. # [INFO/MainProcess] process shutting down

參考

Python 標準庫10 多進程初步 (multiprocessing包)
python 中的 Queue 與多進程（multiprocessing）
正确使用 Multiprocessing 的姿势
Python 多进程编程
multiprocessing--多線程

留言

張貼留言