1.新进一个django项目
- proj/
- proj/__init__.py
- proj/settings.py
- proj/urls.py
- manage.py
2.在该项目创建一个proj / proj / celery.py模块来定义Celery实例
from __future__ import absolute_import import os from celery import Celery # set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'proj.settings') #注意这里的proj from django.conf import settings # noqa app = Celery('proj') #创建应用,再这之前要先设置上面的os.environ.setdefault，设置意味着celery命令行程序将知道Django项目的位置
# 你可以在这里直接传递对象，但是使用字符串更好，因为当使用Windows或execv时，worker不必序列化对象
app.config_from_object('django.conf:settings')
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS) #打开自动发现apps里面的那些应用有没有包含tasks.py文件，
#那么Celery应用就会自动去检索创建的任务。比如你添加了一个任务，在django中会实时地检索出来。
#例如：这样您就不必手动将各个模块添加到CELERY_IMPORTS设置中
#- app1/
# - app1/tasks.py
# - app1/models.py
#- app2/
# - app2/tasks.py
# - app2/models.py
@app.task(bind=True)
def debug_task(self):
print('Request: {0!r}'.format(self.request))
实例2
# -*- coding: utf-8 -*-
"""
celery 任务示例 本地启动celery命令: python manage.py celery worker --settings=settings
周期性任务还需要启动celery调度命令：python manage.py celerybeat --settings=settings
"""
import datetime from celery import task
from celery.schedules import crontab
from celery.task import periodic_task from common.log import logger @task()
def async_task(x, y):
"""
定义一个 celery 异步任务
"""
logger.error(u"celery 定时任务执行成功，执行结果：{:0>2}:{:0>2}".format(x, y))
return x + y def execute_task():
"""
执行 celery 异步任务 调用celery任务方法:
task.delay(arg1, arg2, kwarg1='x', kwarg2='y')
task.apply_async(args=[arg1, arg2], kwargs={'kwarg1': 'x', 'kwarg2': 'y'})
delay(): 简便方法，类似调用普通函数
apply_async(): 设置celery的额外执行选项时必须使用该方法，如定时（eta）等
详见 ：http://celery.readthedocs.org/en/latest/userguide/calling.html
"""
now = datetime.datetime.now()
logger.error(u"celery 定时任务启动，将在10s后执行，当前时间：{}".format(now))
# 调用定时任务
async_task.apply_async(args=[now.hour, now.minute], eta=now + datetime.timedelta(seconds=10)) @periodic_task(run_every=crontab(minute='*', hour='*', day_of_week="*"))
def get_time():
"""
celery 周期任务示例 run_every=crontab(minute='*/5', hour='*', day_of_week="*")：每 5 分钟执行一次任务
periodic_task：程序运行时自动触发周期任务
"""
execute_task()
now = datetime.datetime.now()
logger.error(u"celery 周期任务调用成功，当前时间：{}".format(now))
3.修改proj / proj /__init__.py
from __future__ import absolute_import #目的是拒绝隐士引入，celery.py和celery冲突。
#这将确保在Django启动时始终导入应用程序，以便@shared_task装饰器使用
from .celery import app as celery_app # noqa
4.在配置文件proj/settings.py里面添加
import djcelery
djcelery.setup_loader()
BROKER_URL = 'redis://127.0.0.1:6379/10'#BROKER_URL：broker是代理人，它负责分发任务给worker去执行。我使用的是Redis作为broker
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/11'#不填就是默认'djcelery.backends.database:DatabaseBackend'
# CELERY_RESULT_BACKEND = 'redis://localhost:6379'
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TIMEZONE = 'Asia/Shanghai'
CELERY_ENABLE_UTC = False #刚开始时区不准，一直是UTC时间，后来索性把utc禁用
CELERYD_CONCURRENCY = 10
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' # 定时任务
# 这是使用了django-celery默认的数据库调度模型,任务执行周期都被存在你指定的orm数据库中
CELERYD_MAX_TASKS_PER_CHILD = 1 # 每个worker最多执行1个任务就会被销毁，可防止内存泄露
INSTALLED_APPS += ( 'djcelery')
实例2
# 是否启用celery任务
IS_USE_CELERY = True
# 本地开发的 celery 的消息队列（RabbitMQ）信息
BROKER_URL_DEV = 'amqp://guest:guest@127.0.0.1:5672/'
# TOCHANGE 调用celery任务的文件路径, List of modules to import when celery starts.
CELERY_IMPORTS = (
'home_application.celery_tasks',
)
# ===============================================================================
# CELERY 配置
# ===============================================================================
if IS_USE_CELERY:
try:
import djcelery
INSTALLED_APPS += (
'djcelery', # djcelery
)
djcelery.setup_loader()
CELERY_ENABLE_UTC = False
CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"
if "celery" in sys.argv:
DEBUG = False
# celery 的消息队列（RabbitMQ）信息
BROKER_URL = os.environ.get('BK_BROKER_URL', BROKER_URL_DEV)
if RUN_MODE == 'DEVELOP':
from celery.signals import worker_process_init @worker_process_init.connect
def configure_workers(*args, **kwargs):
import django
django.setup()
except:
pass
5.创建任务 tasks，在某个应用中demoapp/tasks.py
@task() 装饰器说明
如果使用了多个装饰器那么需要task装饰器在最后即在最上面，一般情况使用的是从celeryapp中引入的app作为的装饰器：app.task()，如果是django那种在app中定义的task则需要使用@shared_task
from __future__ import absolute_import
from celery import shared_task @shared_task
def add(x, y):
#这里可以把x+y结果写入数据库
return x + y @shared_task
def mul(x, y):
return x * y @shared_task
def xsum(numbers):
return sum(numbers) @task(ignore_result=True,max_retries=1,default_retry_delay=10) #创建测试用的task
def just_print():
print "Print from celery task"
6.快速执行
只需要找到方法，用方法加delay就可。
比如上面的文件：
from demoapp.tasks import add
r = add.delay(3,5) # 执行这一行就是再下发任务在后台执行了，结果到时候在取就可以了，
使用技巧，一般会创建一个标识健,存到数据库，然后传到我们要执行的方法，在执行方法里面写上功能，把执行结果保存到指定数据库，等执行成功后，就用我们之前的标识键去查。
7.定时任务的使用（第一种方法-使用数据库）
使用其实就是修改数据库，修改后django-celery会实时推送到celery-beat里生效。所以只要再开发一个页面去配置djcelery_periodictask及其它表就可以了。
djcelery提供了一些Model（定义在djcelery/models.py文件），数据库模型如下，
| periodictask |
描述定时任务。重要字段有： name: 字符串，标识符 task，字符串，任务函数/类所在的路径，一般是celery_imports + function name。 interval：外键指向intervalschedule，表示每隔多少时间执行 crontab：外键指向crontabschedule，表示在某一时刻执行。 enabled：是否生效 expires: 任务过期时间 |
|
| intervalschedule |
表示时间间隔，有两个参数： every：正数；period间隔单位。比如intervalschedule(every=2, period='day')表示每隔2天 |
|
| crontabschedule | 表示某一时刻，有minute、hour、day_of-week、day_of_month、day_of_year它们的组合意义，参见cron时间表示法，比如0 0 * 10 * 表示每个月的10号凌晨。 |
说明：
任务和定时任务的区别：定时任务 = 任务 + intervalschedule/crontabschedule 。两个定时任务可以执行同一个任务。
任务没有相应的Model，用字符串表示，即periodictask模型的task字段
定时任务有相应的Model即periodictask。
通过它提供的Model Query API来操作，同平常的数据库查询一样。
from djcelery import models as celery_models celery_models.PeriodicTask.objects.create(...)
celery_models.PeriodicTask.ojects.get(name='add')
....
8.定时任务示例
admin毕竟是给后台管理人员使用的，它所有的参数都暴露给使用者了。下面是一个实际使用的例子。
需求：ajax实现月度定时任务monthly_reading_task的执行和控制，即
每个月的某一天执行该任务；可以选择开启或者关闭该定时任务；能够选择任务在哪一天（1-28日）执行。
界面看起来是这样的：
基本上就是Model的增删改查。就不是通过admin来操作了。
查询任务信息
def read(self, request, *args, **kwargs):
try:
task = celery_models.PeriodicTask.objects.get(name=self.TASK_NAME)
if task.enabled:
return {
'enabled': True,
'day_of_month': int(task.crontab.day_of_month),
'last_run_at': task.last_run_at if task.last_run_at else ''
}
else:
return {'enabled': False}
except celery_models.PeriodicTask.DoesNotExist:
return {'enabled': False}
更新日期
def create(self, request, *args, **kwargs):
enabled = request.POST.get('enabled', None)
if enabled not in [self.ENABLED_POST_VALUE, self.DISABLED_POST_VALUE]:
return self.operate_fail('无效参数')
if enabled == self.DISABLED_POST_VALUE:
self.disable_task(self.TASK_NAME)
return self.operate_success()
else:
try:
day_of_month = int(request.POST.get('day_of_month', ''))
if day_of_month > 28 or day_of_month < 1:
return self.operate_fail('日期必须在1-28日之间')
task, created = celery_models.PeriodicTask.objects.get_or_create(name="monthly_reading",
task="mrs_app.my_celery.tasks.monthly_reading_task")
if created:
crontab = celery_models.CrontabSchedule.objects.create(day_of_month=day_of_month,
hour=0,
minute=0)
crontab.save()
task.crontab = crontab
task.enabled = True
task.save()
else:
task.crontab.day_of_month = day_of_month
task.crontab.save()
task.enabled = True
task.save()
return self.operate_success()
except ValueError:
return self.operate_fail('抄表日不能为空')
新建定时任务
def celery_get_tag(request):
name = 'test'
task = 'home_application.celery_tasks.async_task'
task_args ={"x":1, "y":1}
crontab_time = {'month_of_year':'*','day_of_month':'*','day_of_week':'*',
'hour':'*','minute':'*'}
create_task(name, task , task_args, crontab_time)
result = return_result(status=True, code=200, message="添加任务成功")
return result #创建任务
def create_task(name, task, task_args, crontab_time):
'''
name # 任务名字
task # 执行的任务 "myapp.tasks.add"
task_args # 任务参数 {"x":1, "Y":1}
crontab_time # 定时任务时间 格式：
{
'month_of_year': 9 # 月份
'day_of_month': 5 # 日期
'hour': 01 # 小时
'minute':05 # 分钟
}
''' # task任务， created是否定时创建
task, created = celery_models.PeriodicTask.objects.get_or_create(name=name,task=task)
# 获取 crontab
crontab = celery_models.CrontabSchedule.objects.filter(**crontab_time).first()
if crontab is None:
# 如果没有就创建，有的话就继续复用之前的crontab
crontab = celery_models.CrontabSchedule.objects.create(**crontab_time)
task.crontab = crontab # 设置crontab
task.enabled = True # 开启task
task.kwargs = json.dumps(task_args) # 传入task参数
#expiration = timezone.now() + datetime.timedelta(day=1)
#task.expires = expiration # 设置任务过期时间为现在时间的一天以后
task.save()
result = return_result(status=True, code=200, message="添加任务成功")
关闭定时
def disable_task(self, name):
try:
task = celery_models.PeriodicTask.objects.get(name=name)
task.enabled = False
task.save()
return True
except celery_models.PeriodicTask.DoesNotExist:
return True
9.定时任务的使用（第二种方法-CELERYBEAT_SCHEDULE）
djcelery在初始化中主要完成两件：
在settings.CELERY_IMPORTS定义下的模块搜索所有任务。这个对数据库没有任何改变，只是用Admin添加定时任务时periodictask.task字段变成选择框，列出了所有定义的任务。
从settings.CELERYBEAT_SCHEDULE创建定时任务，这个会创建数据记录，相当于celery_models.PeriodicTask.objects.create(..)语句。
所以在settings文件中可添加配置信息，例如：
CELERYBEAT_SCHEDULE = {
'add-every-3-minutes': {
'task': 'mrs_app.my_celery.tasks.monthly_reading_task',
'schedule': timedelta(minutes=3)
# 'schedule': crontab(minute=u'40', hour=u'17',),
},
}
| celerybeat_schedule | 定义任务，上面注释表示每隔3分钟执行monthly_reading_task任务 |
schedule就是执行计划，可以用crontab格式，用这种配置，会自动更新数据库。一般不用这种方法。
10.启动
启动 python manage.py celery worker -l info
如果有定时任务的话，还需要启动心跳
另开一个cmd窗口 python manage.py celery beat （windows下-B选项不可用）
或者后台：
/home/python3/bin/python3 /project/manage.py celery worker --loglevel=info >/dev/null 2>&1 &
/home/python3/bin/python3 /project/manage.py celery beat >> /project/celery.log 2>&1 &
（参考）官网和https://my.oschina.net/kinegratii/blog/292395