使用步骤

1、安装

pip install django django-celery

  

 

  
1
 

2、新建工程

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

  

 

  
1
  
2
  
3
 

项目结构

├── 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


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
 

3、新建任务

course/tasks.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")


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
 

4、配置celery
celery_project/celery_config.py

# -*- 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


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
  
14
  
15
  
16
  
17
  
18
  
19
  
20
  
21
  
22
  
23
  
24
  
25
  
26
  
27
  
28
  
29
  
30
  
31
  
32
  
33
  
34
  
35
  
36
  
37
  
38
  
39
  
40
  
41
  
42
  
43
  
44
  
45
  
46
  
47
  
48
  
49
  
50
  
51
  
52
  
53
  
54
  
55
 

5、配置settings
celery_project/settings.py

# 添加app
INSTALLED_APPS = [ ... 'djcelery', 'course'
]

# 引入 Celery设置
from .celery_config import *


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
 

6、编写视图函数
course/views.py

# -*- 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"})


  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
  
11
  
12
  
13
 

7、配置路由
celery_project/urls.py

from course import views

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

  

 

  
1
  
2
  
3
  
4
  
5
 

8、启动Django服务和 worker

$ python manage.py help
$ python manage.py runserver
$ python manage.py celery worker -l INFO

# 或者
$ python manage.py celery worker -Q queue

  

 

  
1
  
2
  
3
  
4
  
5
  
6
 

访问测试：http://127.0.0.1:8000/do/

9、设置定时任务
celery_project/celery_config.py

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

  

 

  
1
  
2
  
3
  
4
  
5
  
6
  
7
  
8
  
9
  
10
 

10、启动beat

$ python manage.py celery beat -l INFO

  

 

  
1
 

监控工具flower

安装

pip install flower

  

 

  
1
 

启动

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

  

 

  
1
 

django中启动

python manage.py celery flower

  

 

  
1
 

访问管理界面 http://localhost:5555

总结

Django中使用Celery 只是多了3个步骤：
1、task任务编写
2、celery配置
3、启动celery-worker

文章来源: pengshiyu.blog.csdn.net，作者：彭世瑜，版权归原作者所有，如需转载，请联系作者。

原文链接：pengshiyu.blog.csdn.net/article/details/88079097