Create tasks with celery and django installed by default and build-ins already configured with redis

import djcelery  # # #
djcelery.setup_loader()  # # #
CELERY_TIMEZONE='Asia/Shanghai' 
BROKER_URL='redis: / / 127.0.0.1:16379'  
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'  # # #
Copy the code

Add celery. Py under your own project

#! /usr/bin/env python3
# -*- coding:utf-8 -*-
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE'.'the app name. Settings')
app = Celery('the app name')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
Copy the code

Modify init.py in app

#! /bin/python
from __future__ import absolute_import
from .celery import app as celery_app
Copy the code

Add tesk in tasks.py

from celery import task
from celery import shared_task
@task()
def add(x, y) :
    return x + y
@shared_task
def mul(x, y) :
    return x * y
Copy the code

Implemented in interface functions

from django.http import JsonResponse
fromThe app nameimport tasks
def index(request,*args,**kwargs) :
    res=tasks.add.delay(1.3)
    # Task logic
    return JsonResponse({'status':'successful'.'task_id':res.task_id})
Copy the code

Start the celery

python manage.py runserver 0.0. 0. 0:8001
python manage.py celery beat 
python manage.py  celery worker -c 6 -l debug  
Copy the code