Python编程：Django中使用Celery执行异步任务和定时任务

pip install django django-celery

$ django-admin.py startproject celery_project
$ python manage.py startapp course
$ cd celery_project

├── celery_project
│   ├── __init__.py
│   ├── celery_config.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── course
│   ├── __init__.py
│   ├── admin.py
│   ├── apps.py
│   ├── migrations
│   │   ├── __init__.py
│   ├── models.py
│   ├── tasks.py
│   ├── tests.py
│   └── views.py
├── db.sqlite3
└── manage.py

# -*- coding: utf-8 -*-
import time
from celery.task import Task
class CourseTask(Task):
    name = "course-task"
    def run(self, *args, **kwargs):
        print("start course task")
        time.sleep(4)  # 模拟耗时
        print(args, kwargs)
        print("end course task")

# -*- coding: utf-8 -*-
import djcelery
from datetime import timedelta
djcelery.setup_loader()
BROKER_BACKEND = "redis"
BROKER_URL = 'redis://localhost:6379/1'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/2'
# 设置任务队列
CELERY_QUEUES = {
    "beat_queue": {
        "exchange": "beat_queue",
        "exchange_type": "direct",
        "binding_key": "beat_queue"
    },
    "work_queue": {
        "exchange": "work_queue",
        "exchange_type": "direct",
        "binding_key": "work_queue"
    }
}
# 设置默认队列
CELERY_DEFAULT_QUEUE = "work_queue"
# 注册任务函数
CELERY_IMPORTS = (
    "course.tasks",
)
# 设置时区，默认UTC
CELERY_TIMEZONE = 'Asia/Shanghai'
# 有些情况下防止死锁
CELERYD_FORCE_EXECV = True
# 设置并发的worker数量
CELERYD_CONCURRENCY = 4
# 允许重试
CELERYD_ACKS_LATE = True
# 每个worker最多执行100个任务，防止内存泄漏
CELERYD_MAX_TASKS_PER_CHILD = 100
# 单个任务最大运行时间
CELERYD_TASK_TIME_LIMIT = 6 * 60

# 添加app
INSTALLED_APPS = [
    ...
    'djcelery',
    'course'
]
# 引入 Celery设置
from .celery_config import *

# -*- coding: utf-8 -*-
from course.tasks import CourseTask
from django.http import JsonResponse
def do(reqeust):
    # 执行异步任务
    print("start do")
    CourseTask.apply_async(args=("hello", ), queue="work_queque")
    print("end do")
    return JsonResponse({"result": "ok"})

from course import views
urlpatterns = [
    url(r'^do/$', views.do)
]

$ python manage.py help
$ python manage.py runserver
$ python manage.py celery worker -l INFO
# 或者
$ python manage.py celery worker -Q queue

# 设置定时任务
CELERYBEAT_SCHEDULE = {
    "course-task": {
        "task": "course-task",
        "schedule": timedelta(seconds=5),
        "options": {
            "queue": "beat_queue"
        }
    }
}

$ python manage.py celery beat -l INFO

pip install flower

celery flower --address=0.0.0.0 --port=5555 --broker=redis://localhost:6379/1 --basic_auth=user:123

python manage.py celery flower