window 7 + python2.7 + pycharm5 + celery3.1.25 + django1.9.4

$ pip install django-celery

#celery setting
import djcelery
djcelery.setup_loader()  #加載djcelery
BROKER_URL = 'pyamqp://guest@localhost//' #配置broker
BROKER_POOL_LIMIT = 0
CELERY_RESULT_BACKEND='djcelery.backends.database:DatabaseBackend' #配置backend

INSTALLED_APPS = [
 'django.contrib.admin',
 'django.contrib.auth',
 'django.contrib.contenttypes',
 'django.contrib.sessions',
 'django.contrib.messages',
 'django.contrib.staticfiles',
 'djcelery',  #django-celery必須添加
 'test_celery', 
]
DATABASES = { 'default': {  'ENGINE': 'django.db.backends.mysql',  'NAME': 'celery',  'USER':'username',  'PASSWORD':'password',  'HOST':'ip',  'PORT':3306, }}
 

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', 'test_celery.settings')

from django.conf import settings # noqa

app = Celery('test_celery')

# 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))

from __future__ import absolute_import
from celery import shared_task
import time

@shared_task(track_started=True)
def add(x, y):
 time.sleep(30) #模擬長(zhǎng)時(shí)間執(zhí)行
 return x + y

from django.db import models
#在數(shù)據(jù)庫(kù)中生成test_celery_add表
class Add(models.Model):
 task_id = models.CharField(max_length=128) #存儲(chǔ)taskid
 first = models.IntegerField()    #存儲(chǔ)第一個(gè)加數(shù)
 second = models.IntegerField()    #存儲(chǔ)第二個(gè)加數(shù)
 log_date = models.DateTimeField()   #存儲(chǔ)開(kāi)始時(shí)間

python manage.py makemigrations test_celery
python manage.py migrate

celery worker -A test_celery -l info