在不中斷的情況下將對象添加到隊列中

Question

我想將兩個對象放入隊列中，但我必須確保對象同時處於兩個隊列中，因此不應該在兩者之間中斷 - 類似一個原子塊。有人有解決方案嗎？非常感謝...

queue_01.put(car) 
queue_02.put(bike) 

Answer 1

您可以使用Condition object。您可以告訴線程等待cond.wait()，並在隊列準備就緒時發出信號cond.notify_all()。例如，請參閱Doug Hellman的精彩Python Module of the Week blog。他的代碼使用multiprocessing;在這裏，我已經適應它threading：

import threading 
import Queue 
import time 

def stage_1(cond,q1,q2): 
    """perform first stage of work, then notify stage_2 to continue""" 
    with cond: 
     q1.put('car') 
     q2.put('bike') 
     print 'stage_1 done and ready for stage 2' 
     cond.notify_all() 
def stage_2(cond,q): 
    """wait for the condition telling us stage_1 is done""" 
    name=threading.current_thread().name 
    print 'Starting', name 
    with cond: 
     cond.wait() 
     print '%s running' % name 
def run(): 
    # http://www.doughellmann.com/PyMOTW/multiprocessing/communication.html#synchronizing-threads-with-a-condition-object 
    condition=threading.Condition() 
    queue_01=Queue.Queue() 
    queue_02=Queue.Queue()  
    s1=threading.Thread(name='s1', target=stage_1, args=(condition,queue_01,queue_02)) 
    s2_clients=[ 
     threading.Thread(name='stage_2[1]', target=stage_2, args=(condition,queue_01)), 
     threading.Thread(name='stage_2[2]', target=stage_2, args=(condition,queue_02)), 
     ] 
    # Notice stage2 processes are started before stage1 process, and yet they wait 
    # until stage1 finishes 
    for c in s2_clients: 
     c.start() 
     time.sleep(1) 
    s1.start() 
    s1.join() 
    for c in s2_clients: 
     c.join() 

run() 

運行腳本產生

Starting stage_2[1] 
Starting stage_2[2] 
stage_1 done and ready for stage 2 <-- Notice that stage2 is prevented from running until the queues have been packed. 
stage_2[2] running 
stage_2[1] running 

Answer 2

以原子添加到兩個不同的隊列，獲取鎖兩個隊列第一位。通過使用遞歸鎖定的Queue的子類最容易做到這一點。

import Queue # Note: module renamed to "queue" in Python 3 
import threading 

class MyQueue(Queue.Queue): 
    "Make a queue that uses a recursive lock instead of a regular lock" 
    def __init__(self): 
     Queue.Queue.__init__(self) 
     self.mutex = threading.RLock() 

queue_01 = MyQueue() 
queue_02 = MyQueue() 

with queue_01.mutex: 
    with queue_02.mutex: 
     queue_01.put(1) 
     queue_02.put(2)