Django3（二）异步编程

一、使用多线程

二、启用ASGI服务

三、异步视图

异步与同步的转换

其他：消息队列(基本概念)

异步编程：使用协程、线程、进程、消息队列等方式实现。

Django支持多线程、内置异步和消息队列方式实现。

多线程：在当前运行的Django服务中开启新的线程执行。

内置异步：django3，使用asyncio和关键词Async/Await实现，异步功能主要在视图中实现（异步视图）

消息队列：使用celery框架和消息队列中间件搭建，解决了应用耦合、异步消息、流量削锋等问题，实现高性能、高可用、可伸缩和一致性的系统架构。

一、使用多线程

D:.
├─.idea
│ workspace.xml
│
└─MyDjango
│ db.sqlite3
│ manage.py
│
├─index
│ │ admin.py
│ │ apps.py
│ │ models.py
│ │ tests.py
│ │ urls.py
│ │ views.py
│ │ __init__.py
│ │
│ └─migrations
│ 0001_initial.py
│ __init__.py
│
├─MyDjango
│ asgi.py
│ settings.py
│ urls.py
│ wsgi.py
│ __init__.py
│
└─templates
index.html

django的多线程主要应用在视图函数中，当用户在浏览器访问某个路由的时候，就是向django发送Http请求，收到请求后，路由对应的视图函数执行业务处理，若某些业务需要耗时处理，可交给多线程执行，加快网站的响应速度。

INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','index'
]

index/models.pyfrom django.db import modelsclass PersonInfo(models.Model):id = models.AutoField(primary_key=True)name = models.CharField(max_length=20)age = models.IntegerField()

数据迁移

python manage.py migrate

定义路由index和threadIndex

Mydjango/urls.pyfrom django.urls import path, includeurlpatterns = [path('', include(('index.urls', 'index'), namespace='index')),
]

index/urls.pyfrom django.urls import path
from .views import *urlpatterns = [path('', indexView, name='index'),path('thread/', threadIndexView, name='threadIndex'),
]

路由index和threadIndex分别指向视图函数indexView()和ThreadIndexView()，两个视图函数都是用于查询模型PersonInfo的数据，并把查询结果传递给模板文件index.html，再由模板文件生成网页。

index/views.pyfrom django.shortcuts import render
from .models import PersonInfo
from concurrent.futures import ThreadPoolExecutor
import datetime, timedef indexView(request):startTime = datetime.datetime.now()print(startTime)title = '单线程'results = []for i in range(2):person = PersonInfo.objects.filter(id=int(i+1)).first()time.sleep(3)results.append(person)endTime = datetime.datetime.now()print(endTime)print('单线程查询所花费时间', endTime-startTime)return render(request, 'index.html', locals())# 定义多线程任务
def get_info(id):person = PersonInfo.objects.filter(id=int(id)).first()time.sleep(3)return persondef threadIndexView(request):# 计算运行时间startTime = datetime.datetime.now()print(startTime)title = '多线程'Thread = ThreadPoolExecutor(max_workers=2)results = []fs = []# 执行多线程for i in range(2):t = Thread.submit(get_info, i + 1)fs.append(t)# 获取多线程的执行结果for t in fs:results.append(t.result())# 计算运行时间endTime = datetime.datetime.now()print(endTime)print('多线程查询所花费时间', endTime-startTime)return render(request, 'index.html', locals())

视图函数的业务处理是查询模型PersonInfo的id=1和id=2的数据，每次查询间隔时间3秒，然后将所有查询结果写入列表result，再由模板文件解析列表的数据。

视图函数indexView()将模型PersonInfo的两次数据查询是单线程执行，每次查询间隔时间3秒，一共6秒；
视图函数ThreadIndexView()将模型PersonInfo的两次数据查询分别交给两条线程执行，假设两条线程并发执行，数据查询后的等待时间同时发生，一共3秒。

模板文件对视图函数传递的列表解析生成网页信息。

<!DOCTYPE html>
<html lang="zh-hans">
<head><title>{{ title }}</title>
</head>
<body>{% for r in results %}<div>姓名：{{ r.name }}</div><div>年龄：{{ r.age }}</div>{% endfor %}
</body>
</html>

手动新增几条数据

运行python manage.py runserver 8005

如果报错You’re seeing this error because you have DEBUG = True in your Django

可能是django版本不一样，这里使用的是django3.1

单线程和多线程路由：

http://127.0.0.1:8005/

http://127.0.0.1:8005/thread/

二、启用ASGI服务

内置异步是在原有Django服务的基础上再创建一个Django服务，两者是独立运行，互不干扰的。实质是开启一个ASGI服务，ASGI是异步网关协议接口，一个介于网络协议服务和python应用之间的标准接口，能够处理通用的协议类型，包括HTTP、HTTP2和WebSocket，

使用python manage.py runserver运行是在WSGI模式下运行，WSGI是基于HTTP协议模式，不支持WebSocket，ASGI就是为了解决这个问题。

大于等于django3版本，文件目录结构都会有这两个文件

执行WSGI：python manage.py runserver

执行ASGI：第三方模块，可由Daphne、Hypercorn或Uvicorn启动。

Daphne、Hypercorn或Uvicorn可pip安装

daphne -b 127.0.0.1 -p 8005 MyDjango.asgi:application

ASGI服务是WSGI服务的扩展，互不干扰，django中定义的路由都可以在这两个服务访问。

三、异步视图

D:.
└─MyDjango
│ db.sqlite3
│ manage.py
│
├─.idea
│ workspace.xml
│
├─index
│ │ admin.py
│ │ apps.py
│ │ models.py
│ │ tasks.py
│ │ tests.py
│ │ urls.py
│ │ views.py
│ │ __init__.py
│ │
│ └─migrations
│ 0001_initial.py
│ __init__.py
│
└─MyDjango
asgi.py
settings.py
urls.py
wsgi.py
__init__.py

1、 settings.py

INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','index',
]

2、定义TaskInfo模型，是在异步执行过程中实现数据库操作，

from django.db import modelsclass TaskInfo(models.Model):id = models.AutoField(primary_key=True)task = models.CharField(max_length=50)

3、使用Django内置指令执行数据迁移

4、创建数据表

5、定义路由命名空间index,指向项目应用index的urls.py,并在项目应用index的urls.py中分别定义路由syn和asyn

路由syn和asyn 分别对应视图syns()和asyns()，syns()执行同步任务，asyns()执行异步任务

MyDjango/urls.pyfrom django.contrib import admin
from django.urls import path, includeurlpatterns = [path('admin/', admin.site.urls),path('', include(('index.urls', 'index'), namespace='index')),
]

index/urls.pyfrom django.urls import path
from .views import synView, asynViewurlpatterns = [path('syn', synView, name='syn'),path('asyn', asynView, name='asyn'),
]

执行相同任务，循环5次，每次延时1s，结束就对TaskInfo模型进行数据写入操作，整个执行过程会自动计算所消耗时间，对比一下

tasks.pyimport time
import asyncio
from asgiref.sync import sync_to_async
from .models import TaskInfo# 异步任务
async def asyns():start = time.time()for num in range(1, 6):await asyncio.sleep(1)print('异步任务:', num)await sync_to_async(TaskInfo.objects.create,thread_sensitive=True)(task='异步任务')print('异步任务Done, time used:', time.time()-start)# 同步任务
def syns():start = time.time()for num in range(1, 6):time.sleep(1)print('同步任务:', num)TaskInfo.objects.create(task='同步任务')print('同步任务Done, time used:', time.time()-start)

syn()和asyn ()分别由SynsView()和AsynsView()调用

views.pyimport asyncio
from django.http import HttpResponse
from .tasks import syns, asyns
from asgiref.sync import async_to_sync, sync_to_async# 同步视图 -调用同步任务
def synView(request):# # 同步视图 - 调用异步任务# # 异步任务转为同步任务# w = async_to_sync(asyns)# # 调用函数# w()syns()return HttpResponse("Hello, This is syns!")# 异步视图 - 调用异步任务
async def asynView(request):# # 异步视图 - 调用同步任务# # 同步任务转为异步任务# a = sync_to_async(syns)# # 调用函数# loop = asyncio.get_event_loop()# loop.create_task(a())loop = asyncio.get_event_loop()loop.create_task(asyns())return HttpResponse("Hello, This is asyns!")

http://127.0.0.1:8002/syn

http://127.0.0.1:8002/asyn

syn请求和响应同步执行，同一线程依次执行，5s
asyn 延时功能交给异步处理，请求和延时功能在不同线程中执行，减少堵塞，提高响应速度。

异步与同步的转换

为了使同步视图和异步视图能够调用同一个函数，可使用asgiref模块将函数转换为同步任务或者异步任务，再由同步视图和异步视图进行调用。

都需要5s，同步转异步，同步任务没有添加异步特性，只满足了异步视图调用。

本文总结自《Django3 web 应用开发实战-黄永祥》第11章 P355-P363

以下补充：

消息队列(基本概念)

消息队列及常见消息队列介绍 - 腾讯云开发者社区-腾讯云

RabbitMQ的应用场景以及基本原理介绍_杨龙飞的博客的博客-CSDN博客_rabbitmq

消息队列，是一种异步通讯的中间件。可以理解为邮局，发送者将消息投递到邮局，然后邮局帮我们发送给具体的接收者，具体发送过程和时间与我们无关。消息队列是分布式系统中重要的组件。

消息队列在实际应用中四个场景：

应用耦合：多应用间通过消息队列对同一消息进行处理，避免调用接口失败导致整个过程失败；

异步处理：多应用对消息队列中同一消息进行处理，应用间并发处理消息，相比串行处理，减少处理时间；

限流削峰：广泛应用于秒杀或抢购活动中，避免流量过大导致应用系统挂掉的情况；

消息驱动的系统：系统分为消息队列、消息生产者、消息消费者，生产者负责产生消息，消费者(可能有多个)负责对消息进行处理；

应用耦合：用户下单后，订单系统需要通知库存系统。

订单系统：用户下单后，订单系统完成持久化处理，将消息写入消息队列，返回用户订单下单成功

库存系统：订阅下单的消息，采用拉/推的方式，获取下单信息，库存系统根据下单信息，进行库存操作

假如：在下单时库存系统不能正常使用。也不影响正常下单，因为下单后，订单系统写入消息队列就不再关心其他的后续操作了。实现订单系统与库存系统的应用解耦

解耦是消息中间件的一个主要作用，标准的用法是：

生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。

异步处理：用户为了使用某个应用，进行注册，系统需要发送注册邮件并验证短信。
假设三个业务节点分别使用50ms,串行方式使用时间150ms,并行使用时间100ms。虽然并性已经提高的处理时间,但是,前面说过,邮件和短信对我正常的使用网站没有任何影响，客户端没有必要等着其发送完成才显示注册成功,应该是写入数据库后就返回.
引入消息队列，将不是必须的业务逻辑，异步处理。

用户的响应时间相当于是注册信息写入数据库的时间，也就是50毫秒。注册邮件，发送短信写入消息队列后，直接返回，因此写入消息队列的速度很快，基本可以忽略，因此用户的响应时间可能是50毫秒。因此架构改变后，系统的吞吐量提高到每秒20 QPS。比串行提高了3倍，比并行提高了两倍。

限流削峰

购物网站开展秒杀活动，一般由于瞬时访问量过大，服务器接收过大，会导致流量暴增，相关系统无法处理请求甚至崩溃。而加入消息队列后，系统可以从消息队列中取数据，相当于消息队列做了一次缓冲。

可以控制活动的人数

可以缓解短时间内高流量压垮应用

用户的请求，服务器接收后，首先写入消息队列。假如消息队列长度超过最大数量，则直接抛弃用户请求或跳转到错误页面

秒杀业务根据消息队列中的请求信息，再做后续处理

比方说一个秒杀需求，一用有1万件商品，如果每笔秒杀订单，都去访问一次数据库，查一查库存，那得花费多长时间啊。

我们可以这样做，用一个消息队列，定制它的长度为1万，1万以内可以存到消息队列，并立马反馈秒杀成功，之后再去做减库存等一系列操作。一万以后不再近消息队列，并立马反馈秒杀失败。

消息驱动的系统

用户新上传了一批照片，人脸识别系统需要对这个用户的所有照片进行聚类，聚类完成后由对账系统重新生成用户的人脸索引(加快查询)。这三个子系统间由消息队列连接起来，前一个阶段的处理结果放入队列中，后一个阶段从队列中获取消息继续处理。

避免了直接调用下一个系统导致当前系统失败；

每个子系统对于消息的处理方式可以更为灵活，可以选择收到消息时就处理，可以选择定时处理，也可以划分时间段按不同处理速度处理；

消息队列的两种模式

消息队列包括两种模式，点对点模式和发布/订阅模式。

点对点模式特点：

每个消息只有一个接收者（Consumer）(即一旦被消费，消息就不再在消息队列中)；

发送者和接收者间没有依赖性，发送者发送消息之后，不管有没有接收者在运行，都不会影响到发送者下次发送消息；

接收者在成功接收消息之后需向队列应答成功，以便消息队列删除当前接收的消息；

发布订阅式：

发布者将消息发送到Topic,系统将这些消息传递给多个订阅者。

发布/订阅模式特点：

每个消息可以有多个订阅者；

发布者和订阅者之间有时间上的依赖性。针对某个主题（Topic）的订阅者，它必须创建一个订阅者之后，才能消费发布者的消息。

为了消费消息，订阅者需要提前订阅该角色主题，并保持在线运行；