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

Python threading.Event() - Ensuring all waiting threads wake up on event.set()

提问by culhantr

I have a number of threads which wait on an event, perform some action, then wait on the event again. Another thread will trigger the event when it's appropriate.

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

I can't figure out a way to ensure that each waiting thread triggers exactly once upon the event being set. I currently have the triggering thread set it, sleep for a bit, then clear it. Unfortunately, this leads to the waiting threads grabbing the set event many times, or none at all.

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

I can't simply have the triggering thread spawn the response threads to run them once because they're responses to requests made from elsewhere.

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

In short: In Python, how can I have a thread set an event and ensure each waiting thread acts on the event exactly once before it gets cleared?

简而言之：在 Python 中，如何让线程设置事件并确保每个等待线程在事件被清除之前只对事件执行一次？

Update:

更新：

I've tried setting it up using a lock and a queue, but it doesn't work. Here's what I have:

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

# 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

The first time, itemsinq properly reflects how many calls are waiting, but only the first waiting thread to make the call will make it through. From then on, itemsinq is always 1, and the waiting threads take turns; each time the trigger call happens, the next goes through.

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

Update 2It appears as though only one of the event.wait() threads is waking up, and yet the queue.join() is working. This suggests to me that one waiting thread wakes up, grabs from the queue and calls task_done(), and that single get()/task_done() somehow empties the queue and allows the join(). The trigger thread then completes the join(), clears the event, and thus prevents the other waiting threads from going through. Why would the queue register as empty/finished after only one get/task_done call, though?

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

Only one seems to be waking up, even if I comment out the queue.get() and queue.task_done() and hang the trigger so it can't clear the event.

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

采纳答案by Jason Orendorff

You don't need an Event, and you don't need both a Lock and a Queue. All you need is a Queue.

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

Call queue.putto drop a message in without waiting for it to be delivered or processed.

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

Call queue.getin the worker thread to wait for a message to arrive.

调用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()

The messages don't have to be strings; they can be any Python object.

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

回答by bot403

I'm not a python programmer but if an event can only be processed once perhaps you need to switch to a message queue with the appropriate locking so that when one thread wakes up and receives the event message it will process it and remove it from the queue so its not there if other threads wake up and look in the queue.

我不是 Python 程序员，但如果一个事件只能处理一次，也许您需要切换到具有适当锁定的消息队列，以便当一个线程唤醒并接收到事件消息时，它将处理它并将其从队列所以如果其他线程醒来并查看队列，它就不存在。

回答by krs1

One solution I've used in the past is the Queueclass for interthread communication. It is threadsafe and can be used to easy communication between threads when using both the multiprocessing and threading libraries. You could have the child threads waiting for something to enter the queue and then process the new entry. The Queue class also has a get() method which takes a handy blocking argument.

我过去使用的一种解决方案是用于线程间通信的Queue类。它是线程安全的，可用于在同时使用多处理和线程库时轻松地在线程之间进行通信。您可以让子线程等待某些东西进入队列，然后处理新条目。Queue 类还有一个 get() 方法，它接受一个方便的阻塞参数。

回答by Jeremy Brown

If you want discrete, atomic events that can be processed sequentially by each thread, then do as krs1 & bot403 suggested and use a queue. The Python Queueclass is synchronized - you do not have to worry about locking to use it.

如果您想要可以由每个线程顺序处理的离散原子事件，请按照 krs1 和 bot403 建议并使用队列。PythonQueue类是同步的——你不必担心锁定来使用它。

If however your needs are simpler (the event tells you that you have data available to read, etc), you can subscribe/register your threads as observers of an object responsible for triggering the events. This object would maintain the list of observer threading.Eventobjects. On a trigger, it can then call set() on all of the threading.Eventobjects in the list.

然而，如果您的需求更简单（事件告诉您有数据可供读取等），您可以订阅/注册您的线程作为负责触发事件的对象的观察者。这个对象将维护观察者threading.Event对象的列表。在触发器上，它可以threading.Event对列表中的所有对象调用 set() 。