Django中使用WebSocket（channels）_django使用ws

下载channels库

下载channels库和daphne库

python -m pip install -U channels["daphne"]
1

配置setting.py文件

在setting.py进行注册

INSTALLED_APPS = [
    'daphne',
      ···
      ···
    'channels',
]
1
2
3
4
5
6

并配置ASGI_APPLICATION

ASGI_APPLICATION = '项目名.asgi.application'
1

启动服务

启动django服务，若出现以下内容则表示配置成功

将asgi.py文件中的application = get_asgi_application()改成以下内容

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

1
2
3
4
5

创建ws路由

创建websocket服务端文件consumers.py文件

from channels.generic.websocket import AsyncWebsocketConsumer

#异步方法
class MyConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.accept()

    async def disconnect(self, close_code):
        pass

    async def receive(self, text_data):
        await self.send(text_data)




# 同步方式，仅作示例，不使用
class SyncConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()

    def disconnect(self, close_code):
        pass
    # 从WebSocket中接收消息
    def receive(self, text_data=None, bytes_data=None):
        pass

     
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

创建routing.py文件

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'websocket/$', consumers.MyConsumer.as_asgi()),
]

1
2
3
4
5
6
7

编写前端页面的websocket

在前端js处添加以下代码运行项目进行测试，至此则完成django与websocket的连接

 var socket = new WebSocket("ws:127.0.0.1:8000/websocket/");

 socket.onopen = function () {
     console.log('连接成功');//成功连接上Websocket
 };
1
2
3
4
5