如何从 shell 手动运行芹菜周期性任务?

我用的是芹菜和姜戈芹菜。我定义了一个我想测试的周期性任务。是否可以从 shell 手动运行周期性任务,以便查看控制台输出?

67055 次浏览

Have you tried just running the task from the Django shell? You can use the .apply method of a task to ensure that it is run eagerly and locally.

Assuming the task is called my_task in Django app myapp in a tasks submodule:

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

The result instance has the same API as the usual AsyncResult type, except that the result is always evaluated eagerly and locally and the .apply() method will block until the task is run to completion.

I think you'll need to open two shells: one for executing tasks from the Python/Django shell, and one for running celery worker (python manage.py celery worker). And as the previous answer said, you can run tasks using apply() or apply_async()

I've edited the answer so you're not using a deprecated command.

If you means just trigger a task when the condition is not satisfied, for example, the periodic time not meet. You can do it in two steps.

1.Get your task id.

You can do it by typing.

celery inspect registered

You will see something like app.tasks.update_something. If nothing, it is probably that celery was not started. Just run it.

2.Run the task with celery call

celery call app.tasks.update_something

For more details, just type

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