在Django中实现WebSocket通信常见的两种方法分别一种是通过dwebsocket和Channels，dwebsocket模块基于django-websocket开发（dwebsocket使用可以参考文章：Django实战018：dwebsocket成功实现WebSocket），但是由于django-websocket已经停止开发所以dwebsocket的更新也基本停滞了。Channels为提升Django的 ASGI 能力开发的插件，可以同时处理 HTTP、HTTP2、WebSocket 协议。channels在保留了原生django的同步和易用的特性上还带来了异步处理方式，并且将django自带的认证系统以及session集成到模块中，扩展性非常强（官方推荐的插件）。

安装Channels

Python安装模块很简单，直接用pip来在线安装：pip install -U channels。channels安装需要下载很多依赖包，这些包用官方源容易下载超时（建议换源再装，pip换源可以参考文章：Django实战012：PIP配置国内镜像源及使用）。这里-U指--upgrade，如果已安装就升级到最新版。

# 官方源安装很多依赖包下载的容易超时，建议换源再安装
pip install -U channels
# 也可以临时指定安装源，-U指--upgrade，如果已安装就升级到最新版
python -m pip install -U channels -i  https://pypi.tuna.tsinghua.edu.cn/simple/
------------------------------------------
Successfully installed Automat-20.2.0 PyHamcrest-2.0.2 attrs-20.3.0 
autobahn-20.12.3 cffi-1.14.4 channels-3.0.3 constantly-15.1.0 cryptography-3.3.1 
daphne-3.0.1 hyperlink-21.0.0 idna-3.1 incremental-17.5.0 pyasn1-0.4.8 
pyasn1-modules-0.2.8 pycparser-2.20 pyopenssl-20.0.1 service-identity-18.1.0 
twisted-20.3.0 txaio-20.12.1 zope.interface-5.2.0

集成Channels框架

Django2.x 不支持ASGI，所以实现WebSocket需要集成Channels框架（貌似Django3.x无依赖就支持WebSocket）。我们需要创建consumers.py和routing.py文件，routing.py相当于Django中的url用来映射路由和执行函数之间的关系。当监听到websocket的请求时，Django会根据routing.py来触发相应的执行函数。consumers.py则相当于Django中的views视图，在这里定义执行函数及相应的业务逻辑处理。在 wsgi.py 同级目录新增文件 asgi.py（如果你安装了channels在创建项目会自动生成该文件，但是还是需要修改），将Django运行切换到ASGI而不是WSGI服务。

import os, django
from channels.routing import get_default_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'channel_server.settings')
django.setup()
application = get_default_application()

配置Channels

要在Django项目中使用Channels就需要将Channels库添加到已安装的应用程序列表中，编辑 settings.py 件将channels添加到INSTALLED_APPS设置中。Channels走的是通道模式，所以我们需要设置Django以“通道模式”运行。通过ASGI_APPLICATION 来指定Django开启通道。如果有多个通道的话可以使用Redis作为我们的通道层，这时候就需要安装asgi_redis配合了（需安装redis）。

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'corsheaders',  # 跨域处理
    'channels',     # 注册Channels
]
# 启用 ASGI
ASGI_APPLICATION = "channel_server.routing.application"

定义routing.py

在项目settings文件同级目录中新增routing.py，利用Channels的routring模块中的方法来实现路由分发。如果你需要认证可以添加中间件AuthMiddlewareStack或者SessionMiddlewareStack，不需要则直接在URLRouter中定义路由指向。

from channels.auth import AuthMiddlewareStack
from channels.sessions import SessionMiddlewareStack
from channels.routing import ProtocolTypeRouter,URLRouter
from django.urls import path, re_path
from channel_server.consumers import consumers
application = ProtocolTypeRouter({
    'websocket':AuthMiddlewareStack(       # 'websocket':SessionMiddlewareStack(
        URLRouter ([
            # Empty for now (http->django views is added by default)
            # path('ws/parameter/<str:room_name>/', consumers),
            path('ncdev/',test())，
        ])
    )
})

定义consumers.py

这个是核心的功能模块，需要单独拿出来说明。这里我们就简单的测试下整个项目是否可以正常地运行和被访问，这里我们定义一个简单的test类来测试下Channels。当链接进来的时候我们打印“链接上了，当我们接收到客户端消息的时候打印“已发送数据”，当客户端断开链接的时候我们打印“链接断开啦”。

class test(WebsocketConsumer):
    # 当Websocket创建连接时
    def connect(self):
        self.accept()  # 保持连接！必须写这步
        print('连接上啦')
    # 当Websocket接收到消息时
    def receive(self, text_data=None, bytes_data=None):
        self.send(text_data)
        print('已发送数据')
    # 当Websocket发生断开连接时
    def disconnect(self, code):
        print('断开连接啦')

运行项目

当前项目运行环境为：python3.7.9 +Django2.2+Channels3.0.3，项目运行直接通过命令：python manage runserver来启动。然后通过Vue前端向Django发送一个websocket请求，我们可以看到后台已经打印了字符符：webscoket request！！！（说明请求已经进来了）。

Channels测试

这里我们用个WebScoket客户端测试工具来测试下请求结果，分别点击链接、发送数据、断开来测试刚才的几个接口是否能正常响应。在右侧我们可以看到服务端给我们返回了我们发送的数据，说明我们的WebScoket链接已经成功建立并可以发送数据。

以上内容是小编给大家分享的【Django实战021：Channels实现WebSocket】。希望对大家有所帮助，如果大家有任何疑问请给我留言，小编会及时回复大家的。