celery在django项目中的运用和定时任务

#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

celery在django项目中的运用和定时任务

原文：https://www.cnblogs.com/venvive/p/11809431.html