【发布时间】：2017-04-28 09:26:18
【问题描述】：
这不是一个问题，而是帮助那些会发现 celery 4.0.1 文档中描述的周期性任务声明很难集成到 django 中的人： http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html#entries
复制粘贴 celery 配置文件main_app/celery.py:
from celery import Celery
from celery.schedules import crontab
app = Celery()
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test('hello') every 10 seconds.
sender.add_periodic_task(10.0, test.s('hello'), name='add every 10')
# Calls test('world') every 30 seconds
sender.add_periodic_task(30.0, test.s('world'), expires=10)
# Executes every Monday morning at 7:30 a.m.
sender.add_periodic_task(
crontab(hour=7, minute=30, day_of_week=1),
test.s('Happy Mondays!'),
)
@app.task
def test(arg):
print(arg)
问题
但是如果我们使用 django 并且我们的任务被放置在另一个应用程序中呢？使用 celery 4.0.1 我们不再有 @periodic_task 装饰器。所以让我们看看我们能做些什么。
第一种情况
如果您希望将任务及其日程安排彼此靠近：
main_app/some_app/tasks.py
from main_app.celery import app as celery_app
@celery_app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test('hello') every 10 seconds.
sender.add_periodic_task(10.0, test.s('hello'))
@celery_app.task
def test(arg):
print(arg)
我们可以在调试模式下运行beat：
celery -A main_app beat -l debug
我们会看到没有这样的周期性任务。
第二种情况
我们可以尝试在配置文件中这样描述所有周期性任务：
main_app/celery.py
...
app = Celery()
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test('hello') every 10 seconds.
from main_app.some_app.tasks import test
sender.add_periodic_task(10.0, test.s('hello'))
...
结果是一样的。但是它的行为会有所不同，您可以通过 pdb 手动调试看到。在第一个示例中，setup_periodic_tasks 回调根本不会被触发。但在第二个例子中，我们会得到django.core.exceptions.AppRegistryNotReady: Apps aren't loaded yet.（这个异常不会被打印出来）
【问题讨论】：
-
SO 绝对欢迎以问答形式分享信息。但是，您在这里得到的并不是一个写得很好的问题。请重写此代码，使其读作面临实际问题的人所写的实际问题。您已经知道解决方案，但是从不知道的人的角度写问题。 （在我看来，您可以从 p.o.v. 中提出一个问题，即某人从 3.x 迁移到 4.x 并发现过去有效的方法不再有效。）
-
此外，从“问题”标题开始的所有内容都是一个解决方案，应该是一个正式的答案。 （您可以发布截然不同的解决方案作为不同的答案。然后人们可以独立投票。）