python - Celery 的 app.control.broadcast 期望什么 "command"？

Question

我认为 app.control.broadcast 会接受一个 @task，但是当运行以下命令时:

app.send_task("workerTasks_supervisor.task_supervisor_test", args=[], queue='supervisor')

app.control.broadcast("workerTasks_supervisor.task_supervisor_test", args=[], queue="supervisor")

第一个成功，第二个失败:

[2019-08-01 12:10:52,260: ERROR/MainProcess] pidbox command error: KeyError('task_supervisor_test',)
Traceback (most recent call last):
  File "/usr/local/lib/python3.5/dist-packages/kombu/pidbox.py", line 104, in dispatch
    reply = handle(method, arguments)
  File "/usr/local/lib/python3.5/dist-packages/kombu/pidbox.py", line 126, in handle_cast
    return self.handle(method, arguments)
  File "/usr/local/lib/python3.5/dist-packages/kombu/pidbox.py", line 120, in handle
    return self.handlers[method](self.state, **arguments)
KeyError: 'task_supervisor_test'

工作线程启动

celery worker -A workerTasks_supervisor -n Supervisor --concurrency=1 --loglevel=info -Q supervisor -f /logs/celery_supervisor.log --pidfile=/logs/supervisor_pid.pid

任务本身很简单:

@app.task()
def task_supervisor_test():
    print("working")

我做错了什么？ 谢谢。

Answer 1

你的假设是错误的。

你的第二行试图广播一个你没有实现的命令，它自然会抛出异常。

Celery 的优点(其中之一)是它允许您实现自己的 commands 。您可以像上面尝试的那样以编程方式执行它们，或者通过类似 celery -A my.project.app <command> [params...] 的命令行来执行它们。 。这是一个非常强大的概念，我建议每个 Celery 高级用户都应该了解一下。