django通过channels实现websocket

Mrterrific / 2023-05-12 / 原文

WebSocket协议是基于TCP的一种新的协议。WebSocket最初在HTML5规范中被引用为TCP连接，作为基于TCP的套接字API的占位符。它实现了浏览器与服务器全双工(full-duplex)通信。其本质是保持TCP连接，在浏览器和服务端通过Socket进行通信。

当客户端向服务端发送连接请求时，不仅连接还会发送【握手】信息，并等待服务端响应，至此连接才创建成功！

请求和响应的【握手】信息需要遵循规则：

从请求【握手】信息中提取 Sec-WebSocket-Key
利用magic_string 和 Sec-WebSocket-Key 进行hmac1加密，再进行base64加密
将加密结果响应给客户端

注：magic string为：258EAFA5-E914-47DA-95CA-C5AB0DC85B11

如果返回的结果和客户端的加密结果一致，说明服务端支持websocket请求，连接才能建立成功。

当服务端想要主动向客户端发送信息的时候，可以考虑使用websocket

客户端和服务端传输数据时，需要对数据进行【封包】和【解包】。很多库已经封装【封包】和【解包】过程，但有些没有进行封装的例如Socket服务端需要手动实现。

根据第二个字节的后7位的大小不同，数据的长度也不同。

   info = conn.recv(8096)

    payload_len = info[1] & 127
    if payload_len == 126:
        extend_payload_len = info[2:4]
        mask = info[4:8]
        decoded = info[8:]
    elif payload_len == 127:
        extend_payload_len = info[2:10]
        mask = info[10:14]
        decoded = info[14:]
    else:
        extend_payload_len = None
        mask = info[2:6]
        decoded = info[6:]

    bytes_list = bytearray()
    for i in range(len(decoded)):
        chunk = decoded[i] ^ mask[i % 4]
        bytes_list.append(chunk)
    body = str(bytes_list, encoding='utf-8')
    print(body)

Django实现websocket

django channels 是django支持websocket的一个模块。

settings.py

INSTALLED_APPS = [
    xxx,
    'corsheaders',
    'rest_framework',
    'channels',
]

ASGI_APPLICATION = "big_model_code.asgi.application"
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)], # 你的redis地址和端口
        },
    },
}

settings.py同级目录下创建asgi.py

假设app为chatOperation，在app下创建routtings.py , 作用等同于urls.py

urls.py处理http请求，routtings.py处理websocket请求

import os

from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application

from chatOperation.routings import websocket_urlpatterns       

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "big_model_code.settings")  # big_model_code为项目名

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    'websocket':AuthMiddlewareStack(URLRouter(websocket_urlpatterns)),
})

chatOperation.routings.py

from django.urls import re_path
from chatOperation import consumers

websocket_urlpatterns = [
    re_path(r"ws/start/(?P<user_room>\w+)", consumers.PushStream.as_asgi()),  # 这里可以定义自己的路由
    # 如果是传参的路由在连接中获取关键字参数方法：self.scope['url_route']['kwargs']['user_room']
]

app下创建consumers.py，作用等同于views.py

from channels.generic.websocket import WebsocketConsumer
from channels.exceptions import StopConsumer
from asgiref.sync import async_to_sync   # 异步转为同步

class PushStream(WebsocketConsumer):
    """推流"""
    def websocket_connect(self, message):
        #有客户端向后端改善websocket请求时自动触发

        #服务端允许执行下行代码，如果不允许可以用 raise StopConsumner()拒绝客户端的连接请求

        self.room_name = self.scope['url_route']['kwargs'].get('user_room', 'group1')

        # room_group_name 这个是一个分组，最好你主动推送时是要用到这个名字的，所以这个名字命名一定要有点规则，规则自己定义，需要注意的是，这个分组里面是可以放多个连接的，然后推送时，会推送到这个组里面所有用户
        self.room_group_name = '%s%s' % (PUSH_RULE,self.room_name)

        # 固定写法，加入组，如果组已存在，则加入，不存在，则先创建后加入
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name,  # 组名
            self.channel_name      
        )

        # 开启连接
        self.accept()


    def websocket_receive(self, message):
        #客户端发来数据时触发，message是客户端发来的数据（一个字典）

        text_data_json = json.loads(message)
        message = text_data_json['message']

        # Send message to room group
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name,
            {
                'type': 'send_message',   # 推送消息时调用的方法
                'event': message
            }
        )


    def send_message(self, event):
　　　　"""向某个组推送消息时调用此方法"""
        message = event['event']
        self.send(message)

    def websocket_disconnect(self, message):
        # 断开连接时触发
        async_to_sync(self.channel_layer.group_discard)(
            self.room_group_name,
            self.channel_name
        )
        raise StopConsumer()

主动推送信息，例如发起了某个http请求，接收到请求就通过websocket推送信息

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer

def handle_push_stream(user,path):
    """推流"""
    try:
        res = {}
        info = {"message":"plug flow success","path":path}
        # 发送消息
        channel_layer = get_channel_layer()
        async_to_sync(channel_layer.group_send)(
            get_rule_room_name(user),  # 组名
            {
                "type": "send_message",  # 这个就是你上面 consumers.py 定义的 send_message方法
                "event": json.dumps(info) # 是send_message方法接受的参数
            }
        )

        res['message'] = 'success'
        return res
    except:
        logger.error(traceback.format_exc())

nginx配置

location /ws {
                         proxy_pass http://127.0.0.1:8999;
                         proxy_http_version 1.1;
                         proxy_set_header Upgrade $http_upgrade;
                         proxy_set_header Connection "upgrade";
                         proxy_redirect off;
                         proxy_set_header Host $host;
                         proxy_set_header X-Real-IP $remote_addr;
                         proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
                         proxy_set_header X-Forwarded-Host $server_name;
                
		}

服务器需要通过 daphne -p 端口号项目名.asgi:application 来启动websocket

例如: daphne -p 8999 big_model_code.asgi:application