# -*- coding: utf-8 -*-from __future__ import absolute_import, unicode_literalsimport osfrom celery import Celery# 设置环境变量os.environ.setdefault(‘DJANGO_SETTINGS_MODULE‘, ‘project_name.settings‘)# 注册Celery的APPapp = Celery(‘project_name‘)# 绑定配置文件app.config_from_object(‘django.conf:settings‘, namespace=‘CELERY‘)# 自动发现各个app下的tasks.py文件app.autodiscover_tasks()

修改settings.py同级目录的init.py文件

from __future__ import absolute_import, unicode_literalsfrom .celery import app as celery_app__all__ = [‘celery_app‘]

在某个APP中创建tasks.py文件

# -*- coding: utf-8 -*-from celery.task import task# 自定义要执行的task任务@taskdef print_hello():    return ‘hello celery and django...‘

from celery.schedules import crontabCELERY_BEAT_SCHEDULE = {    # 周期性任务    ‘task-one‘: {        ‘task‘: ‘app.tasks.print_hello‘,        ‘schedule‘: 5.0, # 每5秒执行一次        # ‘args‘: ()    },    # 定时任务    ‘task-two‘: {        ‘task‘: ‘app.tasks.print_hello‘,        ‘schedule‘: crontab(minute=0, hour=‘*/3,10-19‘),        # ‘args‘: ()    }}

celery -A project_name worker -l info -P eventlet

celery -A project_name beat -l info

INSTALLED_APPS = (    ...,    ‘django_celery_results‘,)