Using Celery With Django for Background Task Processing_综合

用芹菜配合Django处理背景任务

Web应用程序通常一开始很简单，但可能变得相当复杂，而且大多数应用程序很快就超出了只响应HTTP请求的责任。

当发生这种情况时，必须区分必须立即发生的(通常是在HTTP请求生命周期中)和最终可能发生的事情。那是为什么？好吧，因为当你的应用程序被流量超载时，像这样的简单事情就会产生影响。

Web应用程序中的操作可以分为关键操作或请求时间操作和后台任务，这些操作发生在请求时间之外。这些地图与上文所述的地图相符：

需要立即发生：请求时间操作
最终需要：后台任务

请求时间操作可以在单个请求/响应周期内完成，而不必担心操作会超时或用户可能有不良体验。常见的例子包括CRUD(创建、读取、更新、删除)数据库操作和用户管理(Login/Logout例程)。

背景任务是不同的，因为它们通常很耗时，并且容易失败，主要是由于外部依赖。复杂Web应用程序中的一些常见场景包括：

发送确认或活动电子邮件
每天从不同的来源抓取和收集一些信息，并将它们储存起来。
执行数据分析
删除不需要的资源
以不同格式导出文件/照片

背景任务是本教程的主要重点。用于此场景的最常见的编程模式是生产者消费者体系结构。

简单地说，这个体系结构可以这样描述：

生产者创建数据或任务。
任务被放入称为任务队列的队列中。
使用者负责使用数据或运行任务。

通常，消费者以先进先出(FIFO)的方式或根据他们的优先级从队列中检索任务。消费者也被称为工人，这是我们将始终使用的术语，因为它与所讨论的技术所使用的术语是一致的。

在后台可以处理什么样的任务？任务：

并不是web应用程序的基本功能所必需的。
无法在请求/响应周期中运行，因为它们很慢(I/O密集型，等等)
依赖外部资源，这些资源可能不可用或不能按预期运行。
可能至少需要重审一次
必须按计划执行

芹菜是Python/Django生态系统中进行后台任务处理的事实上的选择。它有一个简单而清晰的API，并与Django完美地集成在一起。它支持任务队列的各种技术和工作人员的各种范例。

在本教程中，我们将创建一个Django玩具Web应用程序(处理现实世界的场景)，该应用程序使用后台任务处理。

把事情安排好

假设您已经熟悉Python包管理和虚拟环境，那么让我们安装Django：

$ pip ``install Django

我决定构建另一个博客应用程序。应用程序的重点将是简单性。用户可以简单地创建一个帐户，而不需要太多的小题大做就可以创建一个帖子并将其发布到平台上。

设置quick_publisherDjango项目：

$ django-admin startproject quick_publisher

让我们启动应用程序：

$ ``cd quick_publisher

$ .``/manage``.py startapp main

启动一个新Django项目时，我喜欢创建一个main包含自定义用户模型等内容的应用程序。通常情况下，我会遇到默认Django的限制User模特。有习俗的User模型给了我们灵活性的好处。

# main/models.py

from django.db ``import models

from django.contrib.auth.models ``import AbstractBaseUser, PermissionsMixin, BaseUserManager

class UserAccountManager(BaseUserManager):

use_in_migrations ``= True

def _create_user(``self``, email, password, ``*``*``extra_fields):

if not email:

raise ValueError(``'Email address must be provided'``)

if not password:

raise ValueError(``'Password must be provided'``)

email ``= self``.normalize_email(email)

user ``= self``.model(email``=``email, ``*``*``extra_fields)

user.set_password(password)

user.save(using``=``self``._db)

return user

def create_user(``self``, email``=``None``, password``=``None``,``*``*``extra_fields):

return self``._create_user(email, password,``*``*``extra_fields)

def create_superuser(``self``, email, password,``*``*``extra_fields):

extra_fields[``'is_staff'``] ``= True

extra_fields[``'is_superuser'``] ``= True

return self``._create_user(email, password,``*``*``extra_fields)

class User(AbstractBaseUser, PermissionsMixin):

REQUIRED_FIELDS ``= []

USERNAME_FIELD ``= 'email'

objects ``= UserAccountManager()

email ``= models.EmailField(``'email'``, unique``=``True``, blank``=``False``, null``=``False``)

full_name ``= models.CharField(``'full name'``, blank``=``True``, null``=``True``, max_length``=``400``)

is_staff ``= models.BooleanField(``'staff status'``, default``=``False``)

is_active ``= models.BooleanField(``'active'``, default``=``True``)

def get_short_name(``self``):

return self``.email

def get_full_name(``self``):

return self``.email

def __unicode__(``self``):

return self``.email

一定要检查Django文献资料如果您不熟悉自定义用户模型的工作方式。

现在我们需要告诉Django使用这个用户模型而不是默认的模型。将这一行添加到quick_publisher/settings.py档案：

AUTH_USER_MODEL = 'main.User'

我们还需要添加main向INSTALLED_APPS列表中的quick_publisher/settings.py档案。现在我们可以创建迁移，应用它们，并创建一个超级用户，以便能够登录到Django管理面板：

$ .``/manage``.py makemigrations main

$ .``/manage``.py migrate

$ .``/manage``.py createsuperuser

现在让我们创建一个单独的Django应用程序，负责POST：

$ .``/manage``.py startapp publish

让我们定义一个简单的POST模型publisher/models.py:

from django.db ``import models

from django.utils ``import timezone

from django.contrib.auth ``import get_user_model

class Post(models.Model):

author ``= models.ForeignKey(get_user_model())

created ``= models.DateTimeField(``'Created Date'``, default``=``timezone.now)

title ``= models.CharField(``'Title'``, max_length``=``200``)

content ``= models.TextField(``'Content'``)

slug ``= models.SlugField(``'Slug'``)

def __str__(``self``):

return '"%s" by %s' % (``self``.title, ``self``.author)

钩住Post使用Django管理的模型在publisher/admin.py文件如下：

from django.contrib ``import admin

from .models ``import Post

@admin``.register(Post)

class PostAdmin(admin.ModelAdmin):

pass

最后，让我们把publisher应用程序与我们的项目一起使用，方法是将它添加到INSTALLED_APPS名单。

我们现在可以运行服务器并转到http://localhost:8000/admin/然后创建我们的第一个帖子，这样我们就可以玩一些东西了：

$ .``/manage``.py runserver

我相信你已经做好了功课，你创造了这些帖子。

让我们继续前进。下一个显而易见的步骤是创建一种查看已发布文章的方法。

# publisher/views.py

from django.http ``import Http404

from django.shortcuts ``import render

from .models ``import Post

def view_post(request, slug):

try``:

post ``= Post.objects.get(slug``=``slug)

except Post.DoesNotExist:

raise Http404(``"Poll does not exist"``)

return render(request, ``'post.html'``, context``=``{``'post'``: post})

让我们将新视图与URL关联起来：quick_publisher/urls.py

# quick_publisher/urls.py

from django.conf.urls ``import url

from django.contrib ``import admin

from publisher.views ``import view_post

urlpatterns ``= [

url(r``'^admin/'``, admin.site.urls),

url(r``'^(?P<slug>[a-zA-Z0-9\-]+)'``, view_post, name``=``'view_post'``)

]

最后，让我们创建呈现POST的模板：publisher/templates/post.html

<!``DOCTYPE html>

<``html``>

<``head lang``=``"en"``>

<``meta charset``=``"UTF-8"``>

<``title``></``title``>

</``head``>

<``body``>

<``h1``>{ { post.title }}</``h1``>

<``p``>{ { post.content }}</``p``>

<``p``>Published by { { post.author.full_name }} on { { post.created }}</``p``>

</``body``>

</``html``>

我们现在可以去http://localhost:8000/the-slug-of-the-post-you-created/在浏览器中。这并不是一个奇迹的网页设计，但作出漂亮的帖子是超出了本教程的范围。

发送确认电子邮件

以下是经典场景：

在平台上创建一个帐户。
您提供一个电子邮件地址，以便在平台上唯一标识。
该平台通过发送带有确认链接的电子邮件来检查您确实是电子邮件地址的所有者。
在执行验证之前，您无法(完全)使用该平台。

让我们添加一个is_verified旗帜和verification_uuid在User模型：

# main/models.py

import uuid

class User(AbstractBaseUser, PermissionsMixin):

REQUIRED_FIELDS ``= []

USERNAME_FIELD ``= 'email'

objects ``= UserAccountManager()

email ``= models.EmailField(``'email'``, unique``=``True``, blank``=``False``, null``=``False``)

full_name ``= models.CharField(``'full name'``, blank``=``True``, null``=``True``, max_length``=``400``)

is_staff ``= models.BooleanField(``'staff status'``, default``=``False``)

is_active ``= models.BooleanField(``'active'``, default``=``True``)

is_verified ``= models.BooleanField(``'verified'``, default``=``False``) ``# Add theis_verifiedflag

verification_uuid ``= models.UUIDField(``'Unique Verification UUID'``, default``=``uuid.uuid4)

def get_short_name(``self``):

return self``.email

def get_full_name(``self``):

return self``.email

def __unicode__(``self``):

return self``.email

让我们利用这个机会将用户模型添加到管理员中：

from django.contrib ``import admin

from .models ``import User

@admin``.register(User)

class UserAdmin(admin.ModelAdmin):

pass

让我们在数据库中反映这些更改：

$ .``/manage``.py makemigrations

$ .``/manage``.py migrate

我们现在需要编写一段代码，在创建用户实例时发送电子邮件。这就是Django信号的意义，这是一个触及这个主题的完美时机。

在应用程序中发生某些事件之前/之后触发信号。我们可以定义在信号被触发时自动触发的回调函数。要使回调触发器，我们必须首先连接到一个信号。

我们将创建一个回调，该回调将在创建用户模型之后触发。我们将在User模式定义如下：main/models.py

from django.db.models ``import signals

from django.core.mail ``import send_mail

def user_post_save(sender, instance, signal, ``*``args, ``*``*``kwargs):

if not instance.is_verified:

# Send verification email

send_mail(

'Verify your QuickPublisher account'``,

'Follow this link to verify your account: '

'[http://localhost:8000%s](http://localhost:8000%s/)' %``reverse(``'verify'``, kwargs``=``{``'uuid'``:``str``(instance.verification_uuid)}),

'from@quickpublisher.dev'``,

[instance.email],

fail_silently``=``False``,

)

signals.post_save.connect(user_post_save, sender``=``User)

我们在这里所做的就是我们定义了一个user_post_save函数并将其连接到post_save对象发送的信号(在保存模型后触发的信号)。User模特。

Django不只是自己发送电子邮件，它需要绑定到电子邮件服务。为了简单起见，可以将gmail凭据添加到quick_publisher/settings.py或者你可以添加你最喜欢的电子邮件提供商。

Gmail配置如下所示：

EMAIL_USE_TLS ``= True

EMAIL_HOST ``= 'smtp.gmail.com'

EMAIL_HOST_USER ``= '<YOUR_GMAIL_USERNAME>@gmail.com'

EMAIL_HOST_PASSWORD ``= '<YOUR_GMAIL_PASSWORD>'

EMAIL_PORT ``= 587

为了测试，进入管理面板，并创建一个有效的电子邮件地址的新用户，您可以快速检查。如果一切顺利，您将收到一封带有验证链接的电子邮件。验证程序还没有准备好。

以下是如何验证该帐户：

# main/views.py

from django.http ``import Http404

from django.shortcuts ``import render, redirect

from .models ``import User

def home(request):

return render(request, ``'home.html'``)

def verify(request, uuid):

try``:

user ``= User.objects.get(verification_uuid``=``uuid, is_verified``=``False``)

except User.DoesNotExist:

raise Http404(``"User does not exist or is already verified"``)

user.is_verified ``= True

user.save()

return redirect(``'home'``)

将视图连接起来：quick_publisher/urls.py

# quick_publisher/urls.py

from django.conf.urls ``import url

from django.contrib ``import admin

from publisher.views ``import view_post

from main.views ``import home, verify

urlpatterns ``= [

url(r``'^$'``, home, name``=``'home'``),

url(r``'^admin/'``, admin.site.urls),

url(r``'^verify/(?P<uuid>[a-z0-9\-]+)/'``, verify, name``=``'verify'``),

url(r``'^(?P<slug>[a-zA-Z0-9\-]+)'``, view_post, name``=``'view_post'``)

]

此外，请记住创建一个home.html文件下main/templates/home.html...它将由home视野。

尝试再次运行整个场景。如果一切顺利，您将收到一封带有有效验证URL的电子邮件。如果您将遵循URL，然后签入管理员，您可以看到帐户是如何被验证的。

异步发送电子邮件

这是我们目前所做的事情的问题所在。您可能已经注意到，创建一个用户有点慢。这是因为Django在请求时间内发送验证邮件。

这就是它的工作原理：我们将用户数据发送到Django应用程序。应用程序创建一个User模型，然后创建到Gmail(或您选择的其他服务)的连接。Django等待响应，然后它才会将响应返回到我们的浏览器。

芹菜来了。首先，确保安装了：

$ pip ``install Celery

我们现在需要在Django应用程序中创建一个芹菜应用程序：

# quick_publisher/celery.py

import os

from celery ``import Celery

os.environ.setdefault(``'DJANGO_SETTINGS_MODULE'``,``'quick_publisher.settings'``)

app ``= Celery(``'quick_publisher'``)

app.config_from_object(``'django.conf:settings'``)

# Load task modules from all registered Django app configs.

app.autodiscover_tasks()

芹菜是一种任务队列。它接收来自Django应用程序的任务，并在后台运行它们。芹菜需要与充当经纪人的其他服务搭配。

代理在web应用程序和芹菜之间传递消息。在本教程中，我们将使用Redis。Redis很容易安装，而且我们可以很容易地开始使用它，而不需要太多的小题大做。

您可以按照Redis快速开始页面上的说明安装Redis。你需要安装Redis Python库，pip install redis，以及使用红花和芹菜所需的捆：pip install celery[redis].

在一个单独的控制台中启动Redis服务器，如下所示：$ redis-server

让我们把芹菜/红宝石相关的秘密加入到quick_publisher/settings.py:

# REDIS related settings

REDIS_HOST ``= 'localhost'

REDIS_PORT ``= '6379'

BROKER_URL ``= '[redis://](redis://)' + REDIS_HOST ``+ ':' + REDIS_PORT ``+ '/0'

BROKER_TRANSPORT_OPTIONS ``= {``'visibility_timeout'``: ``3600``}

CELERY_RESULT_BACKEND ``= '[redis://](redis://)' + REDIS_HOST ``+ ':' + REDIS_PORT ``+``'/0'

在芹菜中进行任何操作之前，必须将其宣布为一项任务。

以下是如何做到这一点：

# main/tasks.py

import logging

from django.urls ``import function reverse() { [native code] }

from django.core.mail ``import send_mail

from django.contrib.auth ``import get_user_model

from quick_publisher.celery ``import app

@app``.task

def send_verification_email(user_id):

UserModel ``= get_user_model()

try``:

user ``= UserModel.objects.get(pk``=``user_id)

send_mail(

'Verify your QuickPublisher account'``,

'Follow this link to verify your account: '

'[http://localhost:8000%s](http://localhost:8000%s/)' %``reverse(``'verify'``, kwargs``=``{``'uuid'``: ``str``(user.verification_uuid)}),

'from@quickpublisher.dev'``,

[user.email],

fail_silently``=``False``,

)

except UserModel.DoesNotExist:

logging.warning(``"Tried to send verification email to non-existing user '%s'" % user_id)

我们在这里所做的是：将发送验证电子邮件功能移到另一个名为tasks.py.

几个注意事项：

文件的名称很重要。所有的应用程序INSTALLED_APPS中注册任务。tasks.py档案。
注意我们是如何装饰send_verification_email功能与@app.task...这告诉芹菜这是一个将在任务队列中运行的任务。
注意我们期望如何作为参数user_id而不是User对象。这是因为当将任务发送到芹菜时，我们可能很难序列化复杂的对象。最好保持简单。

回到main/models.py，信号代码转化为：

from django.db.models ``import signals

from main.tasks ``import send_verification_email

def user_post_save(sender, instance, signal, ``*``args, ``*``*``kwargs):

if not instance.is_verified:

# Send verification email

send_verification_email.delay(instance.pk)

signals.post_save.connect(user_post_save, sender``=``User)

注意我们如何调用.delay方法处理任务对象。这意味着我们要把任务交给芹菜，而不是等待结果。如果我们用send_verification_email(instance.pk)相反，我们仍然会把它送去芹菜，但是会等待任务完成，这不是我们想要的。

在开始创建新用户之前，有一个陷阱。芹菜是一种服务，我们需要开始。打开一个新控制台，确保激活适当的虚拟主机，并导航到项目文件夹。

$ celery worker -A quick_publisher --loglevel=debug --concurrency=4

这就启动了四个芹菜加工工人。是的，现在您终于可以去创建另一个用户了。注意如何没有延迟，并确保观察芹菜控制台中的日志，并查看任务是否正确执行。这个应该是这样的：

[2017-04-28 15:00:09,190: DEBUG``/MainProcess``] Task accepted: main.tasks.send_verification_email[f1f41e1f-ca39-43d2-a37d-9de085dc99de] pid:62065

[2017-04-28 15:00:11,740: INFO``/PoolWorker-2``] Task main.tasks.send_verification_email[f1f41e1f-ca39-43d2-a37d-9de085dc99de] succeeded ``in 2.5500912349671125s: None

芹菜定期作业

下面是另一个常见的场景。大多数成熟的Web应用程序都会发送用户生命周期的电子邮件，以保持用户的参与。一些常见的生命周期电子邮件示例：

每月报告
活动通知(喜欢、友谊请求等)
提醒您完成某些操作(“不要忘记激活您的帐户”)

下面是我们在应用程序中要做的事情。我们将统计每一篇文章被浏览多少次，并发送一份每日报告给作者。每一天，我们都要检查所有的用户，获取他们的帖子，并发送一封电子邮件，其中有一个包含帖子和视图计数的表格。

让我们改变一下Post模型，以便我们能够适应视图计数场景。

class Post(models.Model):

author ``= models.ForeignKey(User)

created ``= models.DateTimeField(``'Created Date'``, default``=``timezone.now)

title ``= models.CharField(``'Title'``, max_length``=``200``)

content ``= models.TextField(``'Content'``)

slug ``= models.SlugField(``'Slug'``)

view_count ``= models.IntegerField(``"View Count"``, default``=``0``)

def __str__(``self``):

return '"%s" by %s' % (``self``.title, ``self``.author)

与往常一样，当我们更改模型时，我们需要迁移数据库：

$ .``/manage``.py makemigrations

$ .``/manage``.py migrate

让我们也修改view_postDjango视图以计数视图：

def view_post(request, slug):

try``:

post ``= Post.objects.get(slug``=``slug)

except Post.DoesNotExist:

raise Http404(``"Poll does not exist"``)

post.view_count ``+``= 1

post.save()

return render(request, ``'post.html'``, context``=``{``'post'``: post})

显示view_count在模板里。加上这个<p>Viewed { { post.view_count }} times</p>在某个地方publisher/templates/post.html档案。现在对一个帖子做一些看法，看看计数器是如何增加的。

让我们创造一个芹菜任务。因为它是关于帖子的，所以我要把它放在publisher/tasks.py:

from django.template ``import Template, Context

from django.core.mail ``import send_mail

from django.contrib.auth ``import get_user_model

from quick_publisher.celery ``import app

from publisher.models ``import Post

REPORT_TEMPLATE ``= """

Here's how you did till now:

{% for post in posts %}

"{ { post.title }}": viewed { { post.view_count }} times |

{% endfor %}

"""

@app``.task

def send_view_count_report():

for user ``in get_user_model().objects.``all``():

posts ``= Post.objects.``function filter() { [native code] }``(author``=``user)

if not posts:

continue

template ``= Template(REPORT_TEMPLATE)

send_mail(

'Your QuickPublisher Activity'``,

template.render(context``=``Context({``'posts'``: posts})),

'from@quickpublisher.dev'``,

[user.email],

fail_silently``=``False``,

)

每次你改变芹菜的任务，记得重新开始芹菜的过程。芹菜需要发现和重新装填任务。在创建定期任务之前，我们应该在Django shell中测试这一点，以确保一切正常运行：

$ .``/``manage.py shell

In [``1``]: ``from publisher.tasks ``import send_view_count_report

In [``2``]: send_view_count_report.delay()

希望你在电子邮件中收到了一份漂亮的报告。

现在让我们创建一个周期性的任务。敞开quick_publisher/celery.py并登记定期任务：

# quick_publisher/celery.py

import os

from celery ``import Celery

from celery.schedules ``import crontab

os.environ.setdefault(``'DJANGO_SETTINGS_MODULE'``,``'quick_publisher.settings'``)

app ``= Celery(``'quick_publisher'``)

app.config_from_object(``'django.conf:settings'``)

# Load task modules from all registered Django app configs.

app.autodiscover_tasks()

app.conf.beat_schedule ``= {

'send-report-every-single-minute'``: {

'task'``: ``'publisher.tasks.send_view_count_report'``,

'schedule'``: crontab(), ``# change tocrontab(minute=0, hour=0)if you want it to run daily at midnight

},

}

到目前为止，我们创建了一个运行该任务的计划。publisher.tasks.send_view_count_report每分钟，如crontab()符号。您还可以指定各种芹菜时刻表.

打开另一个控制台，激活适当的环境，并启动芹菜节拍服务。

$ celery -A quick_publisher beat

节拍服务的工作是按照日程安排芹菜中的任务。考虑到时间表使send_view_count_report任务根据设置每分钟运行一次。这是好的测试，但不推荐在现实世界的web应用程序。

使任务更加可靠

任务通常用于执行不可靠的操作、依赖外部资源的操作或由于各种原因很容易失败的操作。下面是一个使它们更可靠的指南：

使任务成为幂等的。幂等任务是一种任务，如果中途停止，它不会以任何方式改变系统的状态。该任务要么对系统进行完全更改，要么根本不做任何更改。
重试任务。如果任务失败，最好一次又一次地尝试，直到成功执行为止。你可以用芹菜芹菜重试...另一件有趣的事情是指数退避算法。当考虑将服务器上不必要的负载限制在重试任务时，这可能会派上用场。

结论

我希望这是一个有趣的教程，为您和Django一起使用芹菜做了一个很好的介绍。

以下是我们可以得出的一些结论：

在请求时间之外保持不可靠和耗时的任务是很好的做法。
长期运行的任务应该在后台由工作进程(或其他范例)执行。
背景任务可以用于对应用程序的基本功能不重要的各种任务。
芹菜还可以使用celery beat服务。
任务可以更可靠，如果使幂等和重试(可能使用指数退避)。