Python多重處理的輸出隊列提供了比預期更多的結果

Question

從下面的代碼中，我期望結果列表的長度與多進程所在的項目範圍之一相同：

import multiprocessing as mp 

def worker(working_queue, output_queue): 
    while True: 
     if working_queue.empty() is True: 
      break #this is supposed to end the process. 
     else: 
      picked = working_queue.get() 
      if picked % 2 == 0: 
       output_queue.put(picked) 
      else: 
       working_queue.put(picked+1) 
    return 

if __name__ == '__main__': 
    static_input = xrange(100)  
    working_q = mp.Queue() 
    output_q = mp.Queue() 
    for i in static_input: 
     working_q.put(i) 
    processes = [mp.Process(target=worker,args=(working_q, output_q)) for i in range(mp.cpu_count())] 
    for proc in processes: 
     proc.start() 
    for proc in processes: 
     proc.join() 
    results_bank = [] 
    while True: 
     if output_q.empty() is True: 
      break 
     else: 
      results_bank.append(output_q.get()) 
    print len(results_bank) # length of this list should be equal to static_input, which is the range used to populate the input queue. In other words, this tells whether all the items placed for processing were actually processed. 
    results_bank.sort() 
    print results_bank 

有沒有人有關如何使此代碼正常運行的任何想法？

Answer 1

此代碼永遠不會停止：

每個工人都從隊列中獲得一個項目，只要它不是空的：

picked = working_queue.get() 

，並提出爲每個它得到了一個新問題：

working_queue.put(picked+1) 

因此，隊列將永遠不會爲空，除非進程之間的時間恰好是隊列爲空時，其中一個進程調用empty()。由於隊列長度最初爲100，並且您擁有與cpu_count()一樣多的進程，所以如果這種情況在任何實際系統上停止，我都會感到驚訝。

那麼稍微修改一下代碼就證明我錯了，它在某個時刻停止了，這實際上讓我感到驚訝。用一個進程執行代碼似乎存在一個錯誤，因爲一段時間後進程凍結但不返回。多個進程的結果是不同的。

在循環迭代中添加一個短暫的休眠期使得代碼按照我預期的方式工作，並在上面進行了解釋。雖然它們應該是線程安全的，但似乎在Queue.put,Queue.get和Queue.empty之間有一些時間問題。除去empty測試也給出了預期的結果（沒有陷入空隊列）。

找到不同行爲的原因。放入隊列的對象不會立即刷新。因此，empty可能會返回False，但隊列中的項目正在等待刷新。

從documentation：

注：當一個對象被放在一個隊列，該目的是酸洗和 後臺線程以後的酸洗數據刷新到底層 管。這有一些令人驚訝的後果，但 不應該導致任何實際困難 - 如果他們真的打擾 你，那麼你可以改爲使用由經理創建的隊列。

1. 把一個物體在一個空隊列後可能會有一個微小的延遲隊列的空（）方法返回False和get_nowait（前）可以在不提高Queue.Empty返回。

2. 如果有多個進程正在排列對象，那麼可能會在另一端無序地接收對象。然而，由相同過程入隊的對象將始終按預期順序相對於彼此。