Django 常用插件实战

前面我们已经简单介绍了 django-restframework 插件的基本用法，本节的插件实战主要以 django-celery 和 django-guardian 这两个插件进行演示。使用这两个插件完成一些简单的需求实验，体验这两个插件给我们带来的好处。

1. django-celery 框架实战

我们使用 django-celery 框架完成2个小实验，简单体验下 celery 框架的功能。首先准备好实验环境，包括安装 django-celery 等依赖的模块以及完成初始配置代码。按照以下步骤进行：

安装依赖模块：包括 django-celery 等；

(django-manual) [root@server first_django_app]# pip install django-celery

注意： 如果想要使用 Django 的 ORM 来存储结果，想要安装 django-celery-results 插件；

安装好 redis 服务，使用源码或者 yum 安装即可；

(django-manual) [root@server first_django_app]# yum install redis

推荐使用源码安装方式，可以安装高版本的 redis。最后我在本机上安装了 redis 5.0 版本的服务，如下：

(django-manual) [root@server first_django_app]# redis-server --version
Redis server v=5.0.7 sha=00000000:0 malloc=jemalloc-5.1.0 bits=64 build=5cb8e5612a04130d
(django-manual) [root@server first_django_app]# systemctl status redis
● redis.service - Redis
   Loaded: loaded (/etc/systemd/system/redis.service; disabled; vendor preset: disabled)
  Drop-In: /etc/systemd/system/redis.service.d
           └─limit.conf
   Active: active (running) since 一 2020-01-13 22:41:09 CST; 3 months 28 days ago
 Main PID: 1707 (redis-server)
   CGroup: /system.slice/redis.service
           └─1707 /usr/local/bin/redis-server 127.0.0.1:6379

新增一个处理异步任务的应用：async_tasks，另外设置相应的 settings.py 中的配置：

(django-manual) [root@server first_django_app]# django-admin startapp async_tasks

在 async_tasks 目录下准备一个 tasks.py 文件在，这个代码里面会放一些需要异步处理的任务：

# 代码位置：async\_tasks/tasks.py
# Create your tasks here
from __future__ import absolute_import, unicode_literals
import time
from celery import shared_task

@shared_task
def add(x, y):
    time.sleep(10)
    return x + y

最后，为了能启动 celery 服务，我们需要在 Django 项目下 settings.py 的同级目录下添加一个 celery.py 文件 (必须这样命名) :

# 代码位置：first\_django\_app/celery.py
from __future__ import absolute_import, unicode_literals 
import os
from celery import Celery
from django.conf import settings

# 设置环境变量
os.environ.setdefault("DJANGO\_SETTINGS\_MODULE", "first\_django\_app.settings")

#创建celery应用
app = Celery('first\_django\_app')
app.config_from_object('django.conf:settings', namespace='CELERY')

#如果在工程的应用中创建了 tasks.py 模块，那么Celery应用就会自动去检索创建的任务。
app.autodiscover_tasks(lambda:settings.INSTALLED_APPS)

# 代码位置：first\_django\_app/\_\_init\_\_.py
from __future__ import absolute_import

# This will make sure the app is always imported when
# Django starts so that shared\_task will use this app.
from .celery import app as celery_app

接下里我们在 settings.py 中增加 celery 的相关配置，包括设置 broker 等

# 代码位置：first\_django\_app/settings.py
INSTALLED_APPS = [
    # ...
    'djcelery',
    # 注册应用
    'hello\_app',
    'async\_tasks'
]  

import djcelery
djcelery.setup_loader()
CELERY_TIMEZONE='Asia/Shanghai'
CELERY_BROKER_URL='redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0' # BACKEND配置，这里使用redis
CELERY_RESULT_SERIALIZER = 'json' 
CELERY_IMPORTS = ('async\_tasks.tasks')
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler'  ###

测试环境。下面第一个以交互式方式启动 celery，后续则是以守护进程方式启动。

# 交互式方式启动
(django-manual) [root@server first_django_app]# celery -A first_django_app worker -l info
...
# 以守护进程方式启动
(django-manual) [root@server first_django_app]# celery multi start w1 -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid
celery multi v4.4.2 (cliffs)
> Starting nodes...
> w1@server: OK
(django-manual) [root@server first_django_app]# celery multi stop w1 -A first_django_app -l info
celery multi v4.4.2 (cliffs)
> w1@server: DOWN

启动服务之后，我们在另一个连接窗口进行测试，使用 Django 的 shell 交互模式调用 celery 的 task。

(django-manual) [root@server first_django_app]# python manage.py shell
Python 3.8.1 (default, Dec 24 2019, 17:04:00) 
[GCC 4.8.5 20150623 (Red Hat 4.8.5-39)] on linux
Type "help", "copyright", "credits" or "license" for more information.
(InteractiveConsole)
>>> from async_tasks.tasks import add
>>> add.delay(8, 10)
<AsyncResult: c41bee82-f17d-4556-8c48-b246b36822b8>

# 当然在等待10s之后，我们也能通过如下方式获取结果
>>> from celery.result import AsyncResult
>>> res = AsyncResult("c41bee82-f17d-4556-8c48-b246b36822b8")
>>> res.result
18

执行该异步任务后，我们在 redis 中查看执行的结果。刚开始是空的，等到10s 之后，结果就会写入 redis 中，默认的 key 是 celery-task-meta-[任务id]：

[root@server ~]# redis-cli
127.0.0.1:6379> get celery-task-meta-c41bee82-f17d-4556-8c48-b246b36822b8
(nil)
# 10s过后，结果才会出来
127.0.0.1:6379> get celery-task-meta-c41bee82-f17d-4556-8c48-b246b36822b8
"{\"status\": \"SUCCESS\", \"result\": 18, \"traceback\": null, \"children\": [], \"date_done\": \"2020-05-12T09:51:53.716003\", \"task_id\": \"c41bee82-f17d-4556-8c48-b246b36822b8\"}"

有了上面的基础后，我们就可以实现两个简单的工作了。

实验1：异步发送邮件。

准备好邮件发送代码，放到 async_tasks/utils.py 中。

# 代码位置：async\_tasks/utils.py
import smtplib
from email.mime.text import MIMEText
from email.header import Header

# 发送方邮箱
mail_from = '2894577759@qq.com'  
# 大家可以使用自己的邮箱作为发送邮箱
passwd = 'xxxxx' 

def send\_email(subject_text, content='', receivers=['2894577759@qq.com']):
    send_html = "<html><head></head><body><h3>{}</h3></body></html>".format(content)
    msg = MIMEText(send_html, 'html', 'utf-8')

    # 放入邮件主题
    msg['Subject'] = subject_text
    msg['From'] = mail_from

    try:
        s = smtplib.SMTP_SSL("smtp.qq.com", 465)
        # 登录到邮箱
        s.login(mail_from, passwd)
        # 发送邮件：发送方，收件方，要发送的消息
        s.sendmail(mail_from, receivers, msg.as_string())
        print("发送邮件成功")
    except s.SMTPException as e:
        print("发送邮件失败:{}，请手工查看生成的巡检报告".format(str(e)))
        return False
    finally:
        s.quit()
    return True

if __name__ == "\_\_main\_\_":
    send_email('Django-Celery测试', "慕课网的朋友们，大家好!")

可以直接使用 python 测试，结果如下：

准备异步发送邮件的 task：

# 代码位置: async\_tasks/tasks.py

from celery import shared_task
from async_tasks.utils import send_email
from first_django_app import celery_app

@celery_app.task
def send\_email\_task(title, content="慕课网的兄弟们，大家好!"):
    time.sleep(10)
    return send_email(title, content)

准备好视图函数，比较简单，直接使用异步处理即可：

# 代码位置：async\_tasks/views.py
from django.http.response import HttpResponse
from async_tasks.tasks import send_email_task

# Create your views here.
def send\_email(request, \*args, \*\*kwargs):
    try: # res = send\_email\_task.apply\_async((), countdown=10) 
        res = send_email_task.delay('这是来自django-celery的测试', '慕课网的朋友们，大家好，我是xxx，来自xxx!')  
    except Exception as e:
        print('异常:{}'.format(str(e)))
    return HttpResponse('发送邮件成功，请耐心等待')

准备好 URLConf 配置，注意我们这里新建了一个 app，所以需要在主入口也要加上 URLConf 配置，此外还需要在该 app 下新建对应的 urls.py 文件并在其中添加相应的映射关系：

# 代码位置：first\_django\_app/urls.py

# ...

# 所有url入口
urlpatterns = [
    url('admin/', admin.site.urls),
    url('hello/', include('hello\_app.urls')),
    # 添加async\_tasks应用的映射入口
    url('celery/', include('async\_tasks.urls')),
    url('hello\_world/', hello_world),
    path('api-auth/', include('rest\_framework.urls', namespace='rest\_framework'))
]

# 代码位置：async\_tasks/urls.py
from django.urls import path
from async_tasks import views

urlpatterns = [
    path('send\_email/', views.send_email),
]

在项目目录下，启动 celery 服务(最好以守护进程方式)，这一步非常重要，千万不能漏掉！！

(django-manual) [root@server first_django_app]# celery multi start w1 -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid
(django-manual) [root@server first_django_app]# ps -ef | grep celery
root      7592 23339  0 23:35 pts/0    00:00:00 grep --color=auto celery
root     18503     1  0 22:49 ?        00:00:06 /root/.pyenv/versions/3.8.1/envs/django-manual/bin/python3.8 -m celery worker -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid --hostname=w1@server
root     18851 18503  0 22:49 ?        00:00:00 /root/.pyenv/versions/3.8.1/envs/django-manual/bin/python3.8 -m celery worker -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid --hostname=w1@server
root     18875 18503  0 22:49 ?        00:00:00 /root/.pyenv/versions/3.8.1/envs/django-manual/bin/python3.8 -m celery worker -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid --hostname=w1@server
root     18890 18503  0 22:49 ?        00:00:00 /root/.pyenv/versions/3.8.1/envs/django-manual/bin/python3.8 -m celery worker -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid --hostname=w1@server
root     18903 18503  0 22:49 ?        00:00:00 /root/.pyenv/versions/3.8.1/envs/django-manual/bin/python3.8 -m celery worker -A first_django_app --concurrency=4 -l info --logfile=/root/test/celerylog.log --pidfile=celerypid.pid --hostname=w1@server

最后，我们启动服务，然后请求这个异步的接口，效果如下视频所示。可以看到，请求会立马返回，而任务会交给 celery 去执行，并最后将结果放到 redis 服务中。

效果演示：