下面是详解Django-channels 实现WebSocket实例的完整攻略。
一、前言
Django框架自带的视图(View)只支持HTTP协议,对于websocket等实时通信场景,需要用到Django-channels库。本文将详细讲解如何使用Django-channels实现WebSocket实例。
二、安装配置
- 安装Django-channels
pip install channels
-
配置Django
-
在settings.py中添加如下代码:
INSTALLED_APPS = [
...
'channels',
...
]
# 配置ASGI
# 键值asgi.Application等效应为启动Channels工作方式,Channels在Django中作为中间件加载
ASGI_APPLICATION = 'your_project_name.routing.application'
# WebSocket协议的请求路径,可以自定义,这里以chat为例
# 前缀为路径,后缀为处理类名和方法名,chat.consumer是处理类的路径
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels.layers.InMemoryChannelLayer",
"ROUTING": "your_project_name.routing.websocket_routing",
},
}
- 在根目录下新建routing.py文件,添加如下代码:
from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path
from your_app_name import consumers
websocket_urlpatterns = [
path('chat/', consumers.ChatConsumer.as_asgi()),
]
application = ProtocolTypeRouter({
# 建立这个ProtocolTypeRouter就是告诉Channels,你可以支持对应协议的WebSocket连接,使用URLRouter分发路径
"websocket": URLRouter(websocket_urlpatterns),
})
- 在your_app_name下新建consumers.py文件,添加如下代码:
import json
from channels.generic.websocket import WebsocketConsumer
class ChatConsumer(WebsocketConsumer):
def connect(self):
# 连接时调用,接受WebSocket连接
self.accept()
def disconnect(self, close_code):
# 断开连接时调用
pass
def receive(self, text_data):
# 接收到数据时调用,text_data是Unicode编码的文本
text_data_json = json.loads(text_data)
message = text_data_json['message']
# 发送数据给WebSocket客户端
self.send(text_data=json.dumps({
'message': message
}))
三、使用WebSocket
示例1:在模板中使用JavaScript连接WebSocket
<script>
// 实例化一个WebSocket对象,参数地址在url.py中指定
const socket = new WebSocket('ws://' + window.location.host + '/chat/');
socket.onopen = function(event) {
console.log('WebSocket已连接');
};
socket.onmessage = function(event) {
console.log('接收到数据:' + event.data);
};
socket.onclose = function(event) {
console.log('WebSocket已断开');
};
</script>
示例2:Python客户端连接WebSocket
import websocket
import thread
import time
def on_message(ws, message):
print message
def on_error(ws, error):
print error
def on_close(ws):
print "### closed ###"
def on_open(ws):
def run(*args):
for i in range(3):
# 发送数据到WebSocket客户端
ws.send("Hello %d" % i)
time.sleep(1)
ws.close()
print "thread terminating..."
thread.start_new_thread(run, ())
if __name__ == "__main__":
websocket.enableTrace(True)
ws = websocket.WebSocketApp("ws://localhost:8000/chat/",
on_message = on_message,
on_error = on_error,
on_close = on_close)
ws.on_open = on_open
ws.run_forever()
四、总结
本文详细介绍了使用Django-channels实现WebSocket实例的完整攻略,包括安装配置、使用WebSocket等内容。通过以上示例,相信您已经掌握了如何在Django中使用WebSocket实现实时通信功能。
本站文章如无特殊说明,均为本站原创,如若转载,请注明出处:详解Django-channels 实现WebSocket实例 - Python技术站