使用 celery 任务创建新进程

Question

在 Celery 中，最终重新加载模块的唯一方法是重新启动所有 celery 进程。我知道如何远程关闭工作人员 (broadcast('shutdown', destination = [<workers>]))，但不知道如何让他们恢复工作。

我有一段 Python 代码可以创建一个包含新工作程序的守护进程，但是当我尝试在 Celery 中将它作为 celery 任务运行时，我得到了AssertionError: daemonic processes are not allowed to have children，我猜是与工作人员池的设置方式有关。

1. 有没有办法以某种方式覆盖 Celery 中的这个错误？
2. 如果没有，是否有其他方法可以让 Celery 启动另一个工人来替换自己？也许将整个事情包装在一个 bash 脚本中（尽管在 Python 中这样做的目的是避免使用 Python 调用 bash 来调用 Python）。
3. 如果没有，是否有其他方法可以说服 Celery 重新加载最新版本的代码？ --autoreload 标志和broadcast('pool_restart') 都什么都不做。

示例行为：

1. 创建：

   @任务 定义添加（x，y）： 返回 x+y

2. 加载 celery，运行 add.delay(4,4)，返回 8。

3. 更改添加到：

   @任务 定义添加（x，y）： 返回 x*y

4. 变魔术（目前是“重启芹菜”）

5. 再次运行 add.delay(4,4)

你应该返回 16。我总是返回 8，除非我关闭 celery 并重新加载它，如果没有办法从远程机器上调出工作进程，我就无法远程完成，最好使用脚本.

Answer 1

在主管下运行 Celery 守护程序，例如 supervisord。当 celeryd 进程死亡时，主管会重新启动它。

Answer 2

如果您使用的是 Flask，您可以在带有子进程的 shell 中的另一个进程中打开 celery，并使用 Werkzeug 重新加载器重新启动文件：

if __name__ == "__main__":
    celery.control.broadcast("shutdown") # Kill the previous workers
    import subprocess
    subprocess.Popen("celery worker", shell=True) # Or celeryd
    app.run(use_reloader=True) # From Flask/Werkzeug

您不妨使用debug=True 在 Flask/Werkzeug 中安装重新加载器（在生产服务器中不会​​这样做）。 您还可以使用 watchdog 包查看更改。 这在某种程度上是蛮力......但有效。