- #CeleryTest/celery.py
- from __future__ import absolute_import,unicode_literals
- import os
- from celery import Celery
- #set the default Django settings module for the 'celery' program.
- #'CeleryTest'项目名
- os.environ.setdefault('DJANGO_SETTINGS_MODULE','CeleryTest.setting')
- App = Celery('CeleryTest')
- #Using a string here means the worker don't to serialize
- #the configuration object to child processes
- #Using a string here means the worker don't to serialize
- App.config_from_object('django.conf:settings',namespace='CELERY')
- #load task modules from all registered Django App configs
- App.autodiscover_tasks()
- @App.task(bind=True)
- def debug_task(self):
- print('Request: {0!r}'.format(self.request))
- #apple/tasks.py
- #create your tasks here
- from __future__ import absolute_import,unicode_literals
- from celery import shared_task
- #@shared_task 去找这个项目下所有 App 的任务
- @shared_task
- def add(x,y):
- return x+y
- @shared_task
- def mul(x,y):
- return x*y
- @shared_task
- def xsum(number):
- return sum(number)
- #CeleryTest/settings.py
- #for celery
- CELERY_BROKER_URL = 'redis://:@192.168.14.41'
- CELERY_RESULT_BACKEND = 'redis://:@192.168.14.41'
- #CeleryTest/urls.py
- from django.conf.urls import url
- from django.contrib import admin
- from ..apple import views
- urlpatterns = [
- url(r'^admin/', admin.site.urls),
- url(r'^index/$',views.index),
- ]
- #apple/views.py
- from django.shortcuts import render,HttpResponse
- from ..apple import tasks
- from celery.result import AsyncResult
- # Create your views here.
- def index(request):
- """会返回一个任务 id, 然后有一个函数一直去查询他的状态"""
- res = tasks.add.delay(4,777)
- print("res:",res)
- return HttpResponse(res.task_id)
- def task_res(request):
- #获取任务 id 的状态
- result = AsyncResult(id="1111111")
- #return HttpResponse(result.get())
- return HttpResponse(result.status)
- https://www.cnblogs.com/alex3714/articles/6351797.html
来源: http://www.bubuko.com/infodetail-3275999.html