Django+celery 定时任务配置

更多关注：http://www.mknight.cn/

安装

pip install celery
pip install django-celery

配置

修改settings文件

第一行引入absolute_import

from __future__ import absolute_import

INSTALLED_APPS

添加'djcelery','appName'

INSTALLED_APPS = (
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'djcelery',
    'appdemo',
)

djcelery配置

加入配置，在该文件最下面

import djcelery
from celery.schedules import crontab
djcelery.setup_loader()
BROKER_URL = 'redis://:xxxx@192.168.1.10:6380/6'
# BROKER_URL = 'redis://:密码@主机地址:端口号/数据库号'
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'
#计划任务
CELERY_RESULT_BACKEND = 'djcelery.backends.database:DatabaseBackend'
CELERY_ENABLE_UTC = False # 不是用UTC
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_TASK_RESULT_EXPIRES = 10 #任务结果的时效时间

#日志配置
# CELERYD_LOG_FILE = BASE_DIR + "/logs/celery/celery.log" # log路径
# CELERYBEAT_LOG_FILE = BASE_DIR + "/logs/celery/beat.log" # beat log路径
# CELERY_ACCEPT_CONTENT = ['pickle', 'json', 'msgpack', 'yaml'] # 允许的格式

from datetime import timedelta


CELERYBEAT_SCHEDULE = {
    'add-every-3-seconds': {
        'task': 'appdemo.tasks.test_celery',
        # 'schedule': crontab(minute=u'40', hour=u'17',),
        'schedule': timedelta(seconds=3),
        'args': (16, 16)
    },
}

celery.py

示例： /Users/jack/github/Star/Star/celery.py

from __future__ import absolute_import

import os

from celery import Celery, platforms

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery_demo.settings')

from django.conf import settings  # noqa

app = Celery('Earth')
platforms.C_FORCE_ROOT = True

# Using a string here means the worker will not have to
# pickle the object when using Windows.
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

init.py

在项目文件init.py引入
示例：/Users/jack/github/Star/Star/init.py

from __future__ import absolute_import
from .celery import app as celery_app

创建task.py

在APP目录下创建改文件
示例：

from __future__ import absolute_import

import logging
from celery import task
from celery.utils.log import get_task_logger
from celery.schedules import crontab


@task
def test_celery(x, y):
    logger = get_task_logger(__name__)
    logger.info('func start  ----------------->')
    logger.info('application:%s', "TEST_APP")
    logger.info('func end -------------------->')
    print x + y
    return x + y


@task
def test_multiply(x, y):
    logger = get_task_logger(__name__)
    logger.info('func start  ----------------->')
    logger.info('application:%s', "TEST_APP")
    logger.info('func end -------------------->')
    print x * y
    return x * y

同步数据库

python3 manage.py makemigrations
python3 manage.py migrate

启动

python3  manage.py celery -A Star  worker -B

启动

定时任务示例

CELERYBEAT_SCHEDULE = {
    'add-every-3-seconds': {
        'task': 'appdemo.tasks.test_celery',
        # 'schedule': crontab(minute=u'40', hour=u'17',),
        'schedule': timedelta(seconds=3),
        'args': (16, 17)
    },
    'timing': {
        'task': 'appdemo.tasks.test_multiply',
        'schedule': crontab(minute=u'28', hour=u'11',),
        # 'schedule': timedelta(seconds=3),
        'args': (2, 3)
    },
}

更多关注：http://www.mknight.cn