使用步骤
1、安装
pip install django django-celery
2、新建工程
$ 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
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")
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
5、配置settings
celery_project/settings.py
# 添加app INSTALLED_APPS = [ ... 'djcelery', 'course' ] # 引入 Celery设置 from .celery_config import *
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"})
7、配置路由
celery_project/urls.py
from course import views urlpatterns = [ url(r'^do/$', views.do) ]
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
访问测试：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" } } }
10、启动beat
$ python manage.py celery beat -l INFO
监控工具flower
安装
pip install flower
启动
celery flower --address=0.0.0.0 --port=5555 --broker=redis://localhost:6379/1 --basic_auth=user:123
django中启动
python manage.py celery flower
访问管理界面 http://localhost:5555
总结
Django中使用Celery 只是多了3个步骤：
1、task任务编写
2、celery配置
3、启动celery-worker
