如何在 Django 和 Celery 中使用异步任务？

我们将帮助您举例说明如何在 Django 中使用Celery。我们将使用带有 celery 的 django 异步任务。

当我们处理长时间运行的任务时，我们的项目现在Celery对我们的任务时间非常有用。如电子邮件发送、图像处理等。

Celery workers 是 Celery 的基础。即使您打算使用 Celery beat 来安排重复性任务，Celery worker 也会接收您的指令并在预定时间处理它们。Celery Beat 为 Celery 工作人员添加了一个基于时间的调度程序。

在这里我会给你我们将帮助你举例说明如何使用 django 和 celery 的异步任务。那么让我们看看下面的例子：

第 1 步：创建项目

在这一步中，我们将使用 django-admin 创建一个新的 django 项目。返回命令行界面并运行以下命令：

 django-admin startproject example 

第 2 步：创建应用

现在我们将创建一个名为core的应用程序来存储帖子名称列表。我们故意保持基本的东西。使用Control+c停止本地服务器并使用startapp命令创建这个新应用程序。

1. cd example
2. django-admin startapp core

第三步：安装 Celery

接下来，在我们的项目中安装 Celery 。

 python -m pip install celery 

第四步：安装Redis

在我们的本地系统中安装 redis：

• sudo apt update
• sudo apt install redis
• redis-server
• redis-cli
• 127.0.0.1:6379> ping
• PONG
• 127.0.0.1:6379>

在使用 redis-cli 启动 Redis 服务器后，您向它发送了ping命令，它以权威的PONG响应。如果您收到此响应，则您的 Redis 安装成功。

第五步：安装python Redis

在此步骤中，我们需要通过以下命令要求安装 python redis 库。

 python -m pip install redis 

第 6 步：创建 Celery.py 文件

接下来，在示例目录中创建一个 celery.py 文件，打开您的文件并粘贴下面的代码。
example/celery.py

1. import os
2. from celery import Celery
4. os.environ.setdefault(“DJANGO_SETTINGS_MODULE”, “django_celery.settings”)
5. app = Celery(“django_celery”)
6. app.config_from_object(“django.conf:settings”, namespace=”CELERY”)
7. app.autodiscover_tasks()
8. import core.tasks

第 7 步：更新初始化文件

接下来，更新示例目录中的 __init__.py 文件，打开您的文件并粘贴以下代码。
example/__init__.py

1. from .celery import app as celery_app
3. __all__ = (“celery_app”,)

第 8 步：更新 setting.py

然后在我们的settings.py文件中更新INSTALLED_APPS以通知 Django 有关该应用程序的信息。
settings.py

1. ….
2. INSTALLED_APPS = [
3. ‘django.contrib.admin’,
4. ‘django.contrib.auth’,
5. ‘django.contrib.contenttypes’,
6. ‘django.contrib.sessions’,
7. ‘django.contrib.messages’,
8. ‘django.contrib.staticfiles’,
9. ‘core’,
10. ]
12. CELERY_BROKER_URL = “redis://localhost:6379”
13. CELERY_RESULT_BACKEND = “redis://localhost:6379”
15. EMAIL_BACKEND = ‘django.core.mail.backends.smtp.EmailBackend’
16. EMAIL_HOST = ‘smtp.gmail.com’
17. EMAIL_USE_TLS = True
18. EMAIL_PORT = 587
19. EMAIL_HOST_USER = ‘your_email@gmail.com’
20. EMAIL_HOST_PASSWORD = ‘YOUR_APP_PASSWORD’

第 9 步：创建表单

在这一步中，我们需要创建一个将被使用的表单。比如添加引导程序类和验证等。另外，我们需要添加自定义样式。
core/forms.py

1. from time import sleep
2. from django.core.mail import send_mail
3. from django import forms
4. from core.tasks import send_contactus_email_task
6. class ContactForm(forms.Form):
7. email = forms.EmailField(
8. widget=forms.EmailInput(
9. attrs={
10. “placeholder”: “Email”,
11. “class”: “form-control”
12. }
13. ),error_messages={
14. ‘required’: ‘The email field is required.’
15. })
16. message = forms.CharField(
17. widget=forms.Textarea(
18. attrs={
19. “placeholder”: “Message”,
20. “class”: “form-control”,
21. “rows”: 3
22. }
23. ),error_messages={
24. ‘required’: ‘The message field is required.’
25. })
27. def send_email(self):
28. “””Sends an email when the contact form has been submitted.”””
29. # Simulate expensive operation(s) that freeze Django
30. send_contactus_email_task.delay(
31. self.cleaned_data[“email”], self.cleaned_data[“message”]
32. )

第 10 步：创建视图

在这一步中，我们需要配置视图。打开 core/views.py 文件并添加。 core/views.py

1. from core.forms import ContactForm
2. from django.views.generic.edit import FormView
3. from django.views.generic.base import TemplateView
5. class ContactFormView(FormView):
6. template_name = “core/contact.html”
7. form_class = ContactForm
8. success_url = “/success/”
10. def form_valid(self, form):
11. form.send_email()
12. return super().form_valid(form)
14. class SuccessView(TemplateView):
15. template_name = “core/success.html”

第 11 步：创建 tasks.py

在这一步中，我们需要配置任务。打开 core/tasks.py 文件并添加。 core/tasks.py

1. from time import sleep
2. from django.core.mail import send_mail
3. from celery import shared_task
5. @shared_task()
6. def send_contactus_email_task(email_address, message):
7. “””Sends an email when the contact form has been submitted.”””
8. # Simulate expensive operation(s) that freeze Django
9. send_mail(
10. “Contact Us”,
11. f”\t{message}\n\nThank you!”,
12. “example@gmail.com”,
13. [email_address],
14. fail_silently=False,
15. )

第 12 步：创建模板

接下来，然后使用您的文本编辑器创建新的模板文件：core/templates/contact.html文件并添加： core/templates/contact.html

1. <!DOCTYPE html>
2. <html lang=”en”>
3. <head>
4. <meta charset=”UTF-8″>
5. <title>Tuts-Station.com</title>
6. <link rel=”stylesheet” href=”https://cdn.jsdelivr.net/npm>bootstrap@4.6.1/dist/css/bootstrap.min.css”>
7. <style type=”text/css”>
8. body{
9. background-color: #f7fcff;
10. }
11. </style>
12. </head>
13. <body>
14. <div class=”container mt-5 pt-5″>
15. <div class=”row d-flex justify-content-center”>
16. <div class=”col-md-8″>
17. <div class=”card”>
18. <div class=”card-header”>
19. <h5>How to Use Asynchronous Tasks with Django>and Celery? – <span class=”text-primary”>Tuts-Station.com</span></h5>
20. </div>
21. <div class=”card-body”>
22. <form method=”post”>
23. {% csrf_token %}
24. {{ form.as_p }}
25. <button type=”submit” class=”btn btn-success>>Submit</button>
26. </form>
27. </div>
28. </div>
29. </div>
30. </div>
31. </div>
32. </body>
33. </html>

核心/模板/success.html

 <h1>Mail Send Successfully!</h1> 

第 13 步：创建 URL

在本节中，我们需要核心应用程序中的urls.py文件，但 Django 不会使用 startapp 命令为我们创建一个文件。使用文本编辑器创建core/urls.py并粘贴下面的代码。
core/urls.py

1. from django.urls import path
2. from . import views
4. urlpatterns = [
5. path(”, views.ContactFormView.as_view()),
6. path(‘success/’, views.SuccessView.as_view()),
7. ]

接下来，我们需要为我们的核心应用程序添加一个 URL 路径，这可以通过导入 include 并为其设置路径来完成。
example/urls.py

1. from django.contrib import admin
2. from django.urls import path, include
4. urlpatterns = [
5. path(‘admin/’, admin.site.urls),
6. path(”, include(‘core.urls’)),
7. ]

运行服务器

在此步骤中，我们将运行本地开发服务器来使用我们的应用程序，而无需将其部署到 Web。

 python manage.py runserver 

在不停止 Django 服务器的情况下在终端中打开第二个选项卡。

 python -m celery -A example worker -l info 

接下来，使用 Web 浏览器转到 http://localhost:8000/home 地址。

因此，在此示例中，主要目标是发送电子邮件而无需花费任何额外时间使用后端 celery 发送电子邮件。

我希望它能帮助你……