Celery：停止后台线程中的无限循环

Question

我正在尝试使用 threading.Thread 和 Celery，让它们像守护进程一样工作。简单来说，我的目标是让线程去监测硬件传感器，这个功能是作为一个网页界面的温控器的一部分。不过现在，我在这个过程中遇到了一些问题：

from celery import Celery
from celery.signals import worker_init
from celery.signals import worker_process_shutdown

from threading import Thread
from threading import Event

from time import sleep


class ThisClass(object):
  def __init__(self):
    self.shutdown = Event()
    self.thread = Thread(target=self.BackgroundMethod)

  def Start(self):
    self.thread.start()

  def Stop(self):
    self.shutdown.set()

  def BackgroundMethod(self):
    while not self.shutdown.is_set():
      print("Hello, world!")
      sleep(1)


this_class = ThisClass()
celery_app = Celery("tasks", broker="amqp://guest@localhost//")


@worker_init.connect
def WorkerReady(**kwargs):
  this_class.Start()


@worker_process_shutdown.connect
def StopPollingSensors(**kwargs):
  this_class.Stop()

这个 Celery 脚本应该创建一个名为 ThisClass 的实例，叫做 this_class，并在 Celery 启动时运行 this_class.Start()。当 Celery 关闭时，它应该调用 this_class.Stop()，这样可以优雅地退出 ThisClass 中的线程，并让 Celery 干净利落地退出。

但是，当我在 Celery 中按 Ctrl-C 来发送一个 SIGINT 信号时，this_class 的线程仍然在运行，Celery 也没有退出，即使我多次发送 SIGTERM 信号。让我困惑的是，如果我在 ThisClass.Stop 中加一个打印语句，我是能看到这个打印的。此外，如果我在 this_class.Start() 后面加上 sleep(5); this_class.Stop()，线程就会按预期启动和停止，Celery 也会在收到 SIGINT 时正常退出。

我该如何在基于 Celery 的脚本中终止 threading.Thread 实例呢？

Answer 1

可以考虑在外部创建一个 Event 对象，然后把它传递给一个 Thread 的子类。这个线程会一直循环检查这个事件对象。当外部的某个函数调用 Event.set() 时，线程就会顺利地退出这个 while 循环。

import sys, time
from threading import Event, Thread, Timer


class ThisClass2(Thread):
    def __init__(self, event):
        super(ThisClass2,self).__init__()
        self.event = event

    def run(self):
        print 'thread start'
        while not self.event.wait(timeout=1.0):
            print("thread: Hello, world!")
        print 'thread done'


event2 = Event()
x = ThisClass2(event2)
x.start()

print 'okay'
time.sleep(1)

print 'signaling event'
event2.set()

print 'waiting'
x.join()

示例输出：

thread start
okay
thread: Hello, world!
signaling event
waiting
thread done