芹菜周期性任务不起作用

Question

我试图跟随芹菜doc为django.这是我的项目结构:

??? hiren
?   ??? celery_app.py
?   ??? __init__.py
?   ??? settings.py
?   ??? urls.py
?   ??? wsgi.py
??? manage.py
??? reminder
?   ??? admin.py
?   ??? __init__.py
?   ??? migrations
?   ??? models.py
?   ??? serializers.py
?   |?? task.py
?   |?? tests.py
?   |?? views.py

这是我的settings.py文件:

BROKER_URL = 'redis://localhost:6379/4'
CELERYBEAT_SCHEDULE = {
    'run-every-5-seconds': {
        'task': 'reminder.task.run',
        'schedule': timedelta(seconds=5),
        'args': (16, 16)
    },
}

和reminder/task.py文件:

def run():
    print('hello')

当我运行celery -A hiren beat -l debug命令时,我没有在终端中看到"hello"文本.我错过了什么？

Answer 1

要从任何可调用的任务创建任务,您需要使用task()装饰器.这将创建一个芹菜任务run().

提醒/ task.py:

from celery import Celery

app = Celery('tasks', broker='redis://localhost')

@app.task
def run():
    print('hello')

Celery库必须在使用前进行实例化,此实例称为应用程序(或app简称).

如果您使用基于"旧"模块的celery API,那么您可以像这样导入任务装饰器:

from celery import task

@task
def run():
    print('hello')

即使这会像第一种方法一样创建芹菜任务,但不建议这样做.