開發環境：window10，python2.7，django1.11.13，

安裝：pip install celery-with-redis，pip install celery，pip install django-celery，pip install redis

          redis安裝連結：https：//github.com/MSOpenTech/redis。找到Redis-x64-3.2.100.msi

目錄結構：

- djangoDemo
  - djangoDemo
    setting.py
    wsgi.py
    urls.py
  - app
    models.py
    views.py
    ...
 

配置setting.py

INSTALLED_APPS = (
    ...
    'app',
    'djcelery',
)


BROKER_URL = 'redis://127.0.0.1:6379/0'
BROKER_TRANSPORT = 'redis'

CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'  

建立PY檔案在djangoDemo / djangoDemo / celery.py目錄下：

# encoding: utf-8
from __future__ import absolute_import,unicode_literals
import os
from celery import Celery
from django.conf import settings
# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'djangoDemo.settings')
app = Celery('djangoDemo')
# 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))
 

建立djangoDemo /應用程式/ tasks.py

from __future__ import absolute_import
from djangoDemo.celery import app
@app.task
def add(x, y):
   return x+y

view.py中使用

from django.shortcuts import render, HttpResponse

from app.tasks import add

from djangoDemo.celery import app
def tests(request):
    result = add.delay(2,4)
    #獲取id
    ids = result.id
    #獲取執行狀態
    sget = result.state
    #返回執行狀態
    state = app.AsyncResult(ids).state
    #返回執行結果
    results = app.AsyncResult(ids).result
    return HttpResponse(results)
 

url.py配置

  url(r'^tests/', views.tests),

啟動django

python manage.py runserver

啟動celery

python manage.py celery worker -c 4 --loglevel=info

 測試：