阻塞主线程直到python后台线程完成副任务

Question

我有一个线程化的 python 应用程序，在后台线程中有一个长时间运行的主循环。这个后台主循环实际上是对pyglet.app.run() 的调用，它驱动一个GUI 窗口，也可以配置为定期调用其他代码。我需要从主线程随意调用一个do_stuff(duration)函数来触发GUI中的动画，等待动画停止，然后返回。实际动画必须在后台线程中完成，因为 GUI 库无法处理由单独线程驱动的情况。

我认为我需要这样做：

import threading

class StuffDoer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.max_n_times = 0
        self.total_n_times = 0
        self.paused_ev = threading.Event()

    def run(self):
        # this part is outside of my control
        while True:
            self._do_stuff()
            # do other stuff

    def _do_stuff(self):
        # this part is under my control
        if self.paused_ev.is_set():
            if self.max_n_times > self.total_n_times:
                self.paused_ev.clear()
        else:
            if self.total_n_times >= self.max_n_times:
                self.paused_ev.set()
        if not self.paused_ev.is_set():
            # do stuff that must execute in the background thread
            self.total_n_times += 1

sd = StuffDoer()
sd.start()

def do_stuff(n_times):
    sd.max_n_times += n_times
    sd.paused_ev.wait_for_clear()   # wait_for_clear() does not exist
    sd.paused_ev.wait()
    assert (sd.total_n_times == sd.max_n_times)

编辑：使用max_n_times 而不是stop_time 来说明为什么Thread.join(duration) 不起作用。

来自threading.Event的文档：

等待（[超时]）

阻塞直到内部标志为真。 如果进入时内部标志为真， 立即返回。否则，阻塞 直到另一个线程调用 set() 将标志设置为真，或者直到 发生可选超时。

我发现如果我有一对事件paused_ev 和not_paused_ev 并使用not_paused_ev.wait()，我可以获得我正在寻找的行为。我几乎可以只使用Thread.join(duration)，但它只需要在后台线程实际上 注册时间已到时准确返回。我应该使用其他同步对象或其他策略吗？

我也愿意接受我以错误的方式处理整个事情的论点，只要它们是好的论点。

Answer 1

希望我从我的评论中得到一些修订或附加信息，但我有点想知道你是否没有通过子类化 Thread.你可以这样做：

class MyWorker(object):
  def __init__(self):
    t = Thread(target = self._do_work, name "Worker Owned Thread")

    t.daemon = True

    t.start()

  def _do_work(self):
    While True:
      # Something going on here, forever if necessary.  This thread
      # will go away if the other non-daemon threads terminate, possibly
      # raising an exception depending this function's body.

我发现当您要运行的方法是某个其他类的成员函数而不是线程上的 run 方法时，这更有意义。此外，这使您不必在线程内封装一堆业务逻辑。当然，所有 IMO。

Answer 2

您的 GUI 动画线程似乎在其while True 循环中使用了自旋锁。这可以使用线程安全队列来防止。根据我对您问题的阅读，这种方法在功能上是等效且有效的。

我省略了您上面代码的一些细节，这些细节不会改变。我在这里还假设您无法控制的 run() 方法使用 self.stop_time 值来完成其工作；否则不需要线程安全队列。

from Queue import Queue
from threading import Event

class StuffDoer:
  def __init__(self, inq, ready):
    self.inq = inq
    self.ready = ready
  def _do_stuff(self):
    self.ready.set()
    self.stop_time = self.inq.get()

GUIqueue = Queue()
control = Event()

sd = StuffDoer(GUIqueue, control)

def do_stuff(duration):
  control.clear()
  GUIqueue.put(time.time() + duration)
  control.wait()

Answer 3

我最终使用了类似于@wberry 建议的队列，并使用了Queue.task_done 和Queue.wait：

import Queue
import threading

class StuffDoer(threading.Thread):
    def __init__(self):
        threading.Thread.__init__(self)
        self.setDaemon(True)
        self.max_n_times = 0
        self.total_n_times = 0
        self.do_queue = Queue.Queue()

    def run(self):
        # this part is outside of my control
        while True:
            self._do_stuff()
            # do other stuff

    def _do_stuff(self):
        # this part is under my control
        if self.total_n_times >= self.max_n_times:
            try:
                self.max_n_times += self.do_queue.get(block=False)
            except Queue.Empty, e:
                pass
        if self.max_n_times > self.total_n_times:
            # do stuff that must execute in the background thread
            self.total_n_times += 1
            if self.total_n_times >= self.max_n_times:
                self.do_queue.task_done()

sd = StuffDoer()
sd.start()

def do_stuff(n_times):
    sd.do_queue.put(n_times)
    sd.do_queue.join()
    assert (sd.total_n_times == sd.max_n_times)

Answer 4

我根据@g.d.d.c advice 为这个问题制定了解决方案。有我的代码：

threads = []
# initializing aux thread(s) in the main thread ...
t = threading.Thread(target=ThreadF, args=(...))
#t.setDaemon(True) # I'm not sure does it really needed
t.start()
threads.append(t.ident)

# Block main thread
while filter(lambda thread: thread.ident in threads, threading.enumerate()):
    time.sleep(10)

另外，您可以使用Thread.join 来阻塞主线程 - 这是更好的方法。