python - RQ 超时不会终止多线程作业

Question

我在使用 python RQ 运行多线程任务时遇到问题(在 v0.5.6 和 v0.6.0 上测试)。

考虑以下代码，作为我想要实现的目标的简化版本:

事物.py

from threading import Thread

class MyThing(object):
    def say_hello(self):
        while True:
            print "Hello World"

    def hello_task(self):
        t = Thread(target=self.say_hello)
        t.daemon = True # seems like it makes no difference
        t.start()
        t.join()

main.py

from rq import Queue
from redis import Redis
from thing import MyThing

conn = Redis()

q = Queue(connection=conn)

q.enqueue(MyThing().say_hello, timeout=5)

执行 main.py 时(当 rqworker 在后台运行时)，作业会按预期因超时而中断，在 5 秒内。

问题是，当我设置包含线程的任务(例如 MyThing().hello_task)时，线程会永远运行，并且在 5 秒超时结束后什么也不会发生。

如何使用 RQ 运行多线程任务，以便超时会杀死该任务及其子孙和他们的妻子？

Answer 1

当您运行t.join()时，hello_task线程会阻塞并等待，直到say_hello线程返回 - 因此不会收到超时来自 rq 的信号。您可以使用 Thread.join 来允许主线程运行并正确接收超时信号，并在等待线程完成运行的同时设置等待时间。就像这样:

def hello_task(self):
    t = Thread(target=self.say_hello)
    t.start()
    while t.isAlive():
        t.join(1)  # Block for 1 second

这样，如果您愿意，您还可以捕获超时异常并处理它:

def hello_task(self):
    t = Thread(target=self.say_hello)
    t.start()
    try:
        while t.isAlive():
            t.join(1)  # Block for 1 second
    except JobTimeoutException:  # From rq.timeouts.JobTimeoutException
        print "Thread killed due to timeout"
        raise