检查任务是否正在进行 celery 热关闭

Question

长话短说

有没有办法判断我们的芹菜工人是否正在进入热关机状态？换句话说，我可以检查是否有待SIGTERM处理的情况吗？我有一个可以自行重新安排的任务，但如果有待关闭的任务，我想避免重新安排自己，以避免阻碍热关闭。像这样的东西：

if not self.shutdown_pending():
    self.retry(countdown=5, max_retries=3)

实际上，除了重新安排工作之外，我希望能够在获得工作许可后立即彻底摆脱当前的工作，SIGTERM以便我可以在新的代码部署中尽快重新启动我的工作人员：

@app.task(bind=True)
def my_work_task(self):
    work = get_work()
    for item in work:
        if self.shutdown_pending():
            logger.info("Shutdown detected. Bailing.")
            return
        item.process()

---

背景

我有一项任务需要不同的时间（从几秒到几分钟）。我最初使用的是每分钟 celery-beat 计划来调用该任务，但如果我只完成少量工作，例如只需要 10 秒即可完成，那么我想立即重新调用该任务次，以避免等待 50 秒等待下一个 celery 节拍到来，因为新工作很可能会在该时间段内可用。

所有这些都是为了最大限度地减少正在处理的工作项目的延迟。我想避免工作人员坐在那儿无所事事的 50 秒时间，因为在这段时间里可能有一些工作可以完成。请注意，工作会根据数据库中项目的“过期”而变得“准备就绪”，这就是为什么我使用 celerybeat 在可用时将其清理掉，而不是直接触发任务。

我的任务看起来是这样的：

@app.task(bind=True)
def my_work_task(self):
    work = get_work()
    do_some_work(work)
    # if this was just a short bit of work reschedule ourselves
    # immediately to avoid wasting time waiting for the
    # next celery beat.
    if len(work) < SMALL_WORK_THRESHOLD:
        self.retry(countdown=5, max_retries=3)

这一切都工作得很好，除了一件事：当我重新加载我的工作人员（通过发送SIGTERM）时，我最终可能会等待单个工作人员重新安排自己，可能每次都会处理大量工作。每次调用可能需要几分钟，直到达到我的max_retries值。这使得部署新代码成为一个问题，因为工作处理几乎会停止长达几分钟。

Answer 1

遗憾的是，没有像其他类似问题中提到的简单解决方案。

您唯一能做的就是改变方法或使用 a SIGKILL，在这种情况下，只需确保使用任务结果后端来了解您可能丢失了哪些任务，或者如果您在数据库上保留状态，您可能不需要它。

根据我个人的经验，我总是使用 mongodb 来注册任务的星星和结束。这使我能够看到由于机器崩溃而从未完成的任务（我使用的是 CELERY_ACKS_LATE），并且如果我希望某个任务在整个云上只运行一次，也可以执行全局锁定。这样，如果 SIGTERM 在一定时间后不起作用，我会发送 SIGKILL 而不会丢失工作的风险。

希望这可以帮助