按照Celery 官方文档，用 Django + Celery + Redis 写的一个简单项目
如需转载，标记出处
环境准备
1. 安装依赖
创建 Django 项目
1. 创建 Django 项目和 APP
django-admin startproject myproject
cd myproject
python manage.py startapp myapp
2. 注册 APP
在 myproject/settings.py 中添加 myapp：
INSTALLED_APPS = [
...,
'myapp',
]
配置 Celery
1. 创建 celery.py 文件（项目根目录）
在 myproject/celery.py：
import os
from celery import Celery
# 设置 Django 配置文件路径
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
# 实例化 Celery
app = Celery('myproject')
# 使用 Django 配置
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现任务
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print(f'Request: {self.request!r}')
2. 修改 __init__.py（项目目录 myproject/）
让 Django 项目启动时自动加载 Celery
# 导入 celery 应用
from .celery import app as celery_app
__all__ = ('celery_app',)
Celery 配置（在 settings.py）
# Redis 作为 Broker 和 Backend
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
如果你这里出现了错误，可以看下我的celery报错文章
创建异步任务
在 myapp/tasks.py 中定义任务
from celery import shared_task
@shared_task
def add(x, y):
return x + y
视图中调用任务
在 myapp/views.py：
from django.http import JsonResponse
from .tasks import add
def add_task(request):
# 异步调用 Celery 任务
result = add.delay(4, 6)
return JsonResponse({"task_id": result.id})
配置路由
在 myapp/urls.py：
from django.urls import path
from . import views
urlpatterns = [
path('add/', views.add_task, name='add_task'),
]
在 myproject/urls.py：
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path('admin/', admin.site.urls),
path('', include('myapp.urls')),
]
启动服务
1. 启动 Redis
redis-cli
2. 启动 Django
python manage.py runserver
3. 启动 Celery Worker
celery -A myproject worker --loglevel=info
测试项目
访问异步任务接口：
http://127.0.0.1:8000/add/
