当前位置：首页 > article >正文

celery入门

article 2025/3/14 10:21:15

按照Celery 官方文档，用 Django + Celery + Redis 写的一个简单项目

如需转载，标记出处

环境准备

1. 安装依赖

pip install django celery redis

创建 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')),

]