在不中断的情况下将对象添加到队列中

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)