Django和Celery的异步任务案例
推荐超级课程：
@TOC
在Django中进行缓存是一种通过存储频繁访问的数据来提高网页应用程序性能的重要技术。 异步任务允许Django应用程序在HTTP请求-响应周期之外执行耗时的操作。这提高了性能和用户体验。
我们将探讨：
- 在Django中设置Celery。
- 使用消息代理（Redis）进行任务排队。
- 创建和运行异步任务。
- 一个小型项目，集成Celery用于邮件通知。
先决条件
- 系统上安装了Python。
- 在系统上创建并激活虚拟环境。
- 创建了Django项目（
django-celery-demo）。 - 安装并运行了Redis服务器。
步骤1：安装依赖项
运行以下命令以安装Celery和Redis包以及Django：
pip install celery[redis] django
步骤2：配置Celery
2.1 创建celery.py
在您的Django项目文件夹中（与settings.py同一级别），创建一个名为celery.py的文件。并将以下代码粘贴进去。可能需要根据您的项目名称进行更改。
from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
# 设置Celery的默认Django设置模块
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_celery_demo.settings')
app = Celery('django_celery_demo')
# 使用字符串在这里意味着工作进程不需要序列化对象
app.config_from_object('django.conf:settings', namespace='CELERY')
# 在Django应用中自动发现任务
app.autodiscover_tasks()
@app.task(bind=True)
def debug_task(self):
print(f'请求：{self.request!r}')
2.2 更新 __init__.py
修改你的项目 __init__.py 以导入 Celery：
from __future__ import absolute_import, unicode_literals
# 这将确保当Django启动时总是导入app
from .celery import app as celery_app
__all__ = ('celery_app',)
步骤3：配置Django设置
在 settings.py文件中，添加Celery和Redis配置：
# Redis设置
CELERY_BROKER_URL = 'redis://localhost:6379/0'
# 任务设置
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'
步骤4：定义Celery任务
4.1 创建 tasks.py
你需要先创建app，然后在你的Django应用之一（例如，app_name）中创建一个 tasks.py 文件。
from celery import shared_task
import time
@shared_task # 不要忘记添加这个装饰器
def add(x, y):
time.sleep(5) # 模拟一个耗时的任务
return x + y
@shared_task # 不要忘记添加这个装饰器
def send_email_task(email):
print(f'向 {email} 发送邮件')
time.sleep(5) # 模拟发送邮件
print('邮件发送成功！')
return "邮件已发送"
步骤5：启动Celery工作进程
在你的终端中像下面这样运行Celery工作进程：
celery -A django_celery_demo worker --loglevel=info
步骤6：测试任务
6.1 调用任务
你可以在Django shell中测试任务。
python manage.py shell
运行以下代码：
from app_name.tasks import add
result = add.delay(4, 5)
print(result.id) # 任务ID
6.2 检查任务状态
要检查任务状态：
result.status # 例如，'PENDING', 'SUCCESS'
result.result # 如果完成，任务结果
步骤7：创建一个小项目
我们将构建一个使用异步发送邮件通知的项目。
7.1 邮件队列模型
在 models.py 中我们需要添加这个模型。
from django.db import models
class Email(models.Model):
email = models.EmailField()
subject = models.CharField(max_length=255)
message = models.TextField()
status = models.CharField(max_length=50, default='PENDING')
7.2 邮件提交表单
在 forms.py 中：
from django import forms
from .models import Email
class EmailForm(forms.ModelForm):
class Meta:
model = Email
fields = ['email', 'subject', 'message']
7.3 处理邮件提交的视图
在 views.py 中：
from django.shortcuts import render, redirect
from .forms import EmailForm
from .models import Email
from .tasks import send_email_task
def send_email_view(request):
if request.method == 'POST':
form = EmailForm(request.POST)
if form.is_valid():
email_instance = form.save()
# 触发Celery任务
send_email_task.delay(email_instance.email)
email_instance.status = 'SENT'
email_instance.save()
return redirect('success')
else:
form = EmailForm()
return render(request, 'email_form.html', {'form': form})
7.4 邮件提交模板
创建一个 templates/email_form.html:
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>发送邮件</title>
</head>
<body>
<h1>发送邮件</h1>
<form method="post">
{% csrf_token %}
{{ form.as_p }}
<button type="submit">发送</button>
</form>
</body>
</html>
步骤8：运行项目
- 启动Redis服务器：
redis-server
- 启动Django服务器：
python manage.py runserver
- 启动Celery工作进程：
celery -A django_celery_demo worker --loglevel=info
- 打开
/send-email/的表单，填写详细信息，然后提交。
步骤9：监控Celery任务
要监控任务，使用Flower，一个用于监控Celery的基于Web的工具：
pip install flower
celery -A django_celery_demo flower
感谢阅读。如果你发现有什么错误或更好的方法，请在下面的评论中告诉我。
如果你喜欢这篇文章，请点击下面的👍按钮，以便其他人也能发现它有用。你可以关注我