pip install celery

pip install django-celery

pip install pymysql

- proj/

  - proj/__init__.py

  - proj/settings.py

  - proj/urls.py

- manage.py 

import pymysql

pymysql.install_as_MySQLdb()

# 是否启用celery任务

IS_USE_CELERY = True

# 本地开发的 celery 的消息队列（RabbitMQ）信息

# BROKER_URL_DEV = 'amqp://guest:guest@127.0.0.1:5672/'

BROKER_URL_DEV = 'redis://127.0.0.1:6379/1'

# TOCHANGE 调用celery任务的文件路径, List of modules to import when celery starts.

CELERY_IMPORTS = (

    'celerytest.task',

)

# ===============================================================================

# CELERY 配置

# ===============================================================================

if IS_USE_CELERY:

    try:

        import djcelery

        import sys

        INSTALLED_APPS += (

            'djcelery',            # djcelery

        )

        djcelery.setup_loader()

        CELERY_ENABLE_UTC = False

        CELERYBEAT_SCHEDULER = "djcelery.schedulers.DatabaseScheduler"

        if "celery" in sys.argv:

            DEBUG = False

        # celery 的消息队列（redis）信息

        BROKER_URL = os.environ.get('BK_BROKER_URL', BROKER_URL_DEV)

        #BROKER_URL = 'amqp://guest:guest@127.0.0.1:5672/'

    except:

        pass

# ===============================================================================

# 数据库设置, 本地开发数据库设置

# ===============================================================================

DATABASES = {

    'default': {

        'ENGINE': 'django.db.backends.mysql',  # 默认用mysql

        'NAME': 'celerytest',                 # 数据库名 (默认与APP_ID相同)

        'USER': 'root',                       # 你的数据库user

        'PASSWORD': '',                  # 你的数据库password

        'HOST': '127.0.0.1',               # 开发的时候，使用localhost

        'PORT': '',                       # 默认3306

    },

}

from django.conf.urls import include, url

from django.contrib import admin

import test

urlpatterns = [

    url(r'^admin/', include(admin.site.urls)),

    url(r'^aa/$', test.test),

    url(r'^add/$', test.addtest),

    url(r'^del/$', test.deltest),

    url(r'^tt/$', test.startandstop),

]

# -*- coding: utf-8 -*-

import django     #测试不加如这两行启动worker要报错,而且需要加在最前

django.setup()

import time

from celery import task,shared_task

#直接执行

@task()

def sayhello():

    print('hello ...')

    time.sleep(2)

    print('world ...')

#定时任务

@shared_task(name="celerytest")

def everysay(**kwargs):

    """定时任务调用执行作业"""

    sayadd(**kwargs)

def sayadd(**kwargs):

    print('add1 ...')

    time.sleep(2)

    print(int(kwargs['x'])+int(kwargs['y']))

# -*- coding: utf-8 -*-

from django.shortcuts import render,HttpResponse

from task import *

from djcelery import models as djmodels

import json

def getcrontabtime(minute,hour,day_of_month,month_of_year,day_of_week):

    mycrontab = {}

    mycrontab['minute'] = minute

    mycrontab['hour'] = hour

    mycrontab['day_of_month'] = day_of_month

    mycrontab['month_of_year'] = month_of_year

    mycrontab['day_of_week'] = day_of_week

    return mycrontab

def test(request):

    sayhello.delay()#如果有参数sayhello.delay(x,y,z)

    return HttpResponse("hello world")

def addtest(request):

    '''添加定时任务'''

    task, created = djmodels.PeriodicTask.objects.get_or_create(name='', task='celerytest')   #task对应的是task文件里的函数方法

    mycrontab = getcrontabtime(5,'*','*','*','*')

    crontab = djmodels.CrontabSchedule.objects.filter(**mycrontab).first()

    if crontab is None:

        crontab = djmodels.CrontabSchedule.objects.create(**mycrontab)

    task.crontab = crontab

    task.enabled = True

    param_dic = {}

    param_dic['x'] = 5

    param_dic['y'] = 2

    # print param_dic

    # if type(param_dic) == list:

    #     task.args = param_dic

    if type(param_dic) == dict:

        task.kwargs = json.dumps(param_dic)

    task.save()

    return HttpResponse("添加任务")

def deltest(request):

    '''删除定时任务'''

    try:

        djmodels.PeriodicTask.objects.filter(name='').delete()

    except:

        return HttpResponse("删除定时任务失败")

    return HttpResponse("删除定时任务成功")

def startandstop(request):

    '''启停定时作业'''

    status = request.GET.get('status','')

    task, created = djmodels.PeriodicTask.objects.get_or_create(name='', task='celerytest')

    if status == '':

        task.enabled = False

        task.save()

        return HttpResponse("停用成功")

    elif status == '':

        task.enabled = True

        task.save()

        return HttpResponse("启用成功")

    else:

        return HttpResponse("系统错误")

#配置数据库迁移，生成celery需要的数据表

python manage.py migrate

#启动Redis

sudo redis-server /etc/redis/redis.conf

#启动worker

python manage.py celery worker --loglevel=info

#启动心跳

python manage.py celery beat --max-interval=10 --loglevel=INFO

--max-interval=10 ：每十秒侦测一次任务 --loglevel=INFO：日志等级是INFO

#启动web

python manage.py runserver 0.0.0.0:9999

#直接执行

http://127.0.0.1:9999/aa/

#添加定时服务

http://127.0.0.1:9999/add/

#删除定时服务

http://127.0.0.1:9999/add/

#暂停执行定时服务

http://127.0.0.1:9999/tt/

http://127.0.0.1:9999/tt/?status=0

#启动执行定时服务

http://127.0.0.1:9999/tt/?status=1