Python threading.Event() - 确保所有等待的线程在 event.set() 上唤醒

Question

我有许多线程等待一个事件，执行一些操作，然后再次等待该事件。另一个线程将在适当的时候触发该事件。

我想不出一种方法来确保每个等待线程在设置事件时恰好触发一次。我目前设置了触发线程，睡了一会儿，然后清除它。不幸的是，这会导致等待线程多次获取 set 事件，或者根本没有。

我不能简单地让触发线程生成响应线程来运行它们一次，因为它们是对来自其他地方的请求的响应。

简而言之:在 Python 中，我怎样才能让一个线程设置一个事件并确保每个等待线程在事件被清除之前恰好对事件执行一次？

更新:

我试过使用锁和队列来设置它，但它不起作用。这是我拥有的:

# Globals - used to synch threads
waitingOnEvent = Queue.Queue
MainEvent = threading.Event()
MainEvent.clear()    # Not sure this is necessary, but figured I'd be safe
mainLock = threading.Lock()

def waitCall():
    mainLock.acquire()
    waitingOnEvent.put("waiting")
    mainLock.release()
    MainEvent.wait()
    waitingOnEvent.get(False)
    waitingOnEvent.task_done()
    #do stuff
    return

def triggerCall():
    mainLock.acquire()
    itemsinq = waitingOnEvent.qsize()
    MainEvent.set()
    waitingOnEvent.join()
    MainEvent.clear()
    mainLock.release()
    return

第一次，itemsinq 正确反射(reflect)了有多少调用正在等待，但只有第一个进行调用的等待线程才能通过。从此itemsinq一直为1，等待线程轮流；每次触发调用发生时，下一个通过。

更新 2似乎只有一个 event.wait() 线程正在唤醒 ，而 queue.join() 仍在工作。这向我表明，一个等待线程醒来，从队列中获取并调用 task_done()，并且单个 get()/task_done() 以某种方式清空队列并允许 join()。然后触发器线程完成 join()，清除事件，从而阻止其他等待线程通过。但是，为什么仅在一次 get/task_done 调用后队列就会注册为空/已完成？

即使我注释掉 queue.get() 和 queue.task_done() 并挂起触发器以使其无法清除事件，但似乎只有一个正在醒来。

Answer 1

您不需要事件，也不需要锁和队列。您只需要一个队列。

调用 queue.put 将消息放入，而无需等待它被传递或处理。

在工作线程中调用queue.get等待消息到达。

import threading
import Queue

active_queues = []

class Worker(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.mailbox = Queue.Queue()
        active_queues.append(self.mailbox)

    def run(self):
        while True:
            data = self.mailbox.get()
            if data == 'shutdown':
                print self, 'shutting down'
                return
            print self, 'received a message:', data

    def stop(self):
        active_queues.remove(self.mailbox)
        self.mailbox.put("shutdown")
        self.join()


def broadcast_event(data):
    for q in active_queues:
        q.put(data)

t1 = Worker()
t2 = Worker()
t1.start()
t2.start()
broadcast_event("first event")
broadcast_event("second event")
broadcast_event("shutdown")

t1.stop()
t2.stop()

消息不必是字符串；它们可以是任何 Python 对象。