multiprocessing.Queue()和queue.Queue()的區別

Queue.Queue是程序內非阻塞佇列，用於程序內的各函式模組或執行緒間通訊。

multiprocess.Queue是跨程序通訊佇列。但是不能用於multiprocessing.Pool多程序的通訊。

multiprocessing.Manager

程序池multiprocessing.Pool()的多程序之間的通訊要用multiprocessing.Manager().Queue()

一個multiprocessing.Manager物件會控制一個伺服器程序，其他程序可以通過代理的方式來訪問這個伺服器程序。從而達到多程序間資料通訊且安全。

Manager支援的型別有list,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。

個人理解：Manager()是一箇中轉站，用於把一些不能垮程序的物件共享給多程序。個人認為其效率不會很高。

多程序使用 multiprocessing.Queue通訊

自己寫了個簡單的測試程式。 我擔心multiprocessing.Queue同時有多個程序讀或者多個程序寫，是否是安全的，資料是否會串擾。經過測試我的擔心是多餘的。

另外我擔心multiprocessing.Queue的效率問題。後來使用圖片來測試，發現相比寫磁碟，multiprocessing.Queue的效率要高很多。所以我決定採用multiprocessing.Queue來傳輸圖片。

from multiprocessing import Process, Queue, Pool,Manager
import time, random, cv2

# 寫資料程序執行的程式碼
def write(q, w_cnt, index):
    print('write task:{}'.format(index))
    for num in range(w_cnt):
        q.put('write task-{}, cnt-{}'.format(index, num))
    print( 'write task:{} done'.format( index )
 
 )

# 讀資料程序執行的程式碼
def read(q, read_cnt):
    cnt = 0
    while True:
        if not q.empty():
            value = q.get(True)
            print ('Get:[%s] from queue.' % value)
            cnt += 1
            if cnt >= read_cnt:
                break


if __name__=='__main__':
    r_num = 1  # 讀程序的數量
    w_num = 20 # 寫程序的數量
    w_cnt = 20 # 寫程序迴圈的次數
    q = Queue(10)
    r_list = [Process( target=read, args=(q,w_num*w_cnt,) ) for i in range(r_num)]
    w_list = [Process( target=write, args=(q,w_cnt,i') ) for i in range( w_num )]

    time_start = time.time()
    [task.start() for task in w_list]
    [task.start() for task in r_list]

    [task.join() for task in w_list]
    [task.join() for task in r_list]
    time_used = time.time() - time_start

    print('time_used:{}s'.format(time_used))

參考資料