如何从外壳手动运行芹菜定期任务

Question

我正在使用芹菜和姜戈芹菜。我已经定义了一个要测试的定期任务。是否可以从 shell 手动运行定期任务，以便查看控制台输出？

Answer 1

你有没有试过从 Django shell 运行任务？您可以使用任务的 .apply 方法来确保它在本地紧急运行。

假设任务在 Django 应用程序中myapp在 tasks 子模块中调用my_task：

$ python manage.py shell
>>> from myapp.tasks import my_task
>>> eager_result = my_task.apply()

结果实例具有与通常的AsyncResult类型相同的 API，只是始终在本地急切地评估结果，并且 .apply() 方法将阻塞，直到任务运行完成。

Answer 2

如果您的意思是仅在不满足条件时触发任务，例如，周期性时间不满足。您可以通过两个步骤完成。

1.获取您的任务 ID。

您可以通过键入来做到这一点。

celery inspect registered

你会看到类似 app.tasks.update_something .如果没有，可能是celery没有启动。只需运行它。

2.使用celery call运行任务

celery call app.tasks.update_something

有关更多详细信息，只需键入

celery --help
celery inspect --help
celery call --help

Answer 3

我认为

你需要打开两个shell：一个用于执行来自Python/Django shell的任务，另一个用于运行celery worker（python manage.py celery worker）。正如前面的答案所说，您可以使用apply()或apply_async()运行任务

我已经编辑了答案，因此您没有使用已弃用的命令。