目录结构
taskproj
├── app01
│ ├── __init__.py
│ ├── apps.py
│ ├── migrations
│ │ └── __init__.py
│ ├── models.py
│ ├── tasks.py
│ └── views.py
├── manage.py
├── taskproj
│ ├── __init__.py
│ ├── settings.py
│ ├── urls.py
│ └── wsgi.py
└── templates
在项目目录taskproj/taskproj/目录下新建celery.py:
```python
#!/usr/bin/env python3
# -*- coding:utf-8 -*-
# Author:wd
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'taskproj.settings') # 设置django环境
app = Celery('taskproj')
app.config_from_object('django.conf:settings', namespace='CELERY') # 使用CELERY_ 作为前缀，在settings中写配置
app.autodiscover_tasks() # 发现任务文件每个app下的task.py
taskproj/taskproj/init.py：
from __future__ import absolute_import, unicode_literals
from .celery import app as celery_app
__all__ = ['celery_app']
taskproj/taskproj/settings.py:
CELERY_BROKER_URL = 'redis://10.1.210.69:6379/0' # Broker配置，使用Redis作为消息中间件
CELERY_RESULT_BACKEND = 'redis://10.1.210.69:6379/0' # BACKEND配置，这里使用redis
CELERY_RESULT_EXPIRES = None # 存储结果过期时间
CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
进入项目的taskproj目录启动worker：
celery worker -A taskproj -l debug
任务定义在每个tasks文件中，app01／tasks.py：
from __future__ import absolute_import, unicode_literals
from celery import shared_task
@shared_task
def add(x, y):
return x + y
@shared_task
def mul(x, y):
return x * y
