Django中使用Celery异步任务

在Django企业级应用开发中，Celery异步任务是处理耗时操作的核心组件。本文将介绍如何在Django项目中集成Celery。

环境准备 首先安装必要的依赖包：

pip install celery redis django-celery-beat

配置步骤

在Django项目根目录创建celery.py文件：

import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
celery_app = Celery('myproject')
celery_app.config_from_object('django.conf:settings', namespace='CELERY')
celery_app.autodiscover_tasks()

在__init__.py中导入：

from .celery import celery_app

__all__ = ('celery_app',)

在settings.py中配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_SERIALIZER = 'json'

使用示例 创建异步任务：

from celery import shared_task

@shared_task
def send_email_task(email, subject):
    # 模拟耗时操作
    time.sleep(5)
    print(f'发送邮件给{email}')
    return f'邮件已发送至{email}'

在视图中调用：

from .tasks import send_email_task

def my_view(request):
    task = send_email_task.delay('user@example.com', '测试')
    return JsonResponse({'task_id': task.id})

启动Celery服务：celery -A myproject worker --loglevel=info

企业级应用中，建议使用Redis作为消息代理，并配置任务重试机制和监控工具。