Django-celery-redis 最小化例子
-
安装Django
pip install Django
-
安装redis
pip install redis
-
安装celery
pip install celery django-celery
-
创建Django项目
django-admin.py startproject myproject
-
新建APP
python manage.py startapp celerytest
-
编辑 settings.py
import djcelery djcelery.setup_loader() BROKER_URL = "redis://127.0.0.1:6379/0" BACKEND_URL = "redis://127.0.0.1:6379/0" CELERY_RESULT_BACKEND = "redis" INSTALLED_APPS = [ ...... 'djcelery', 'celerytest', ]
-
编辑 myproject/settings.py
import djcelery djcelery.setup_loader() BROKER_URL = "redis://127.0.0.1:6379/0" BACKEND_URL = "redis://127.0.0.1:6379/0" CELERY_RESULT_BACKEND = "redis"
INSTALLED_APPS = [ ...... 'djcelery', 'celerytest', ]
-
编辑 celerytest/urls.py
from django.conf.urls import url from celerytest import views
urlpatterns=[ url(r'^(?P<param>\d+)/$', views.index, name='index'), ]
-
编辑 views.py
from django.http import HttpResponse from celerytest.tasks import add
def index(request,param): r = add.delay(int(param),int(param)) return HttpResponse('param plus param is :'+ " " + str(r.get()))
-
编辑celerytest/tasks.py
from djcelery import celery from celery import Celery
@celery.task(trail=True) def add(x, y): return x + y @celery.task def sleeptask(i): from time import sleep sleep(i) return i @celery.task def raisetask(): raise KeyError("foo")
-
新建 myproject/celery.py
from future import absolute_import, unicode_literals import os from celery import Celery, platforms
# set the default Django settings module for the 'celery' program. os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings') app = Celery('myproject') # Using a string here means the worker don't have to serialize # the configuration object to child processes. # - namespace='CELERY' means all celery-related configuration keys # should have a `CELERY_` prefix. app.config_from_object('django.conf:settings', namespace='CELERY') # Load task modules from all registered Django app configs. app.autodiscover_tasks() platforms.C_FORCE_ROOT = True @app.task(bind=True) def debug_task(self): print('Request: {0!r}'.format(self.request))
-
编辑 myproject/init.py
from future import absolute_import
# This will make sure the app is always imported when # Django starts so that shared_task will use this app. from .celery import app as celery_app __all__ = ['celery_app']
-
启动 celery的worker
celery -A myproject worker -l info
-
测试