1. 程式人生 > >Python+Django+channels實現websocket

Python+Django+channels實現websocket

        目前網路上多數基於django實現websocket都是老版本,而django和channels都升級到2.0+版本,導致在搭建過程中填坑無數,好在看了諸多官網文章總算搞定,都是英文,著實好了一點時間,底部附上原始碼,歡迎大家一起交流。

專案版本:

Python 3.6
Django 2.0.2
Channels 2.0.2
channels_redis 2.0.2
Daphne 2.0.4
asgi_redis 1.4.3
asgiref   2.1.6
pypiwin32 2.2.3

 注意事項:

1,安裝channels之前,先把剩下的幾個庫全部安裝,用PIP即可,若安裝過程出現UnicodeDecodeError:‘utf-8’ codec can not decode byte 0xd6 in position 2:invalid continuation byte錯誤則將D:\Python36-32\Lib\site-packages\pip\compat\_init_.py中return s.decode('utf-8')修改為return s.decode('gbk')。

2,專案基於redis,各位自行安裝並執行redis,執行channels時,保證redis已啟動Redis啟動命令,cmd進入redis安裝目錄(c:\\redis),Redis-server.exe redis.windows.conf  回車,正常會顯示no error。預設redis已經做成服務並開機自啟動,所以只需啟動一次即可。

配置過程:

settings.py檔案

redis_host = os.environ.get('REDIS_HOST', 'localhost')

# Channel layer definitions
# http://channels.readthedocs.io/en/latest/topics/channel_layers.html
CHANNEL_LAYERS = { "default": { # This example app uses the Redis channel layer implementation channels_redis "BACKEND": "channels_redis.core.RedisChannelLayer", "CONFIG": { "hosts": [(redis_host, 6379)],}, }, } ASGI_APPLICATION = "NetWorkMonitor.routing.application"
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'channels',
    'AppMain',
    'WebSocket',
]

新增asgi.py檔案,內容如下:

import os
import django
from channels.routing import get_default_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "NetWorkMonitor.settings")
django.setup()
application = get_default_application()

修改wsgi.py檔案,內容如下:

import os

from django.core.wsgi import get_wsgi_application

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "NetWorkMonitor.settings")

application = get_wsgi_application()

修改routing.py檔案,EchoConsumer是websocket的響應類內容如下:

from django.urls import path
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from WebSocket.consumers import EchoConsumerapplication = ProtocolTypeRouter({
"websocket": AuthMiddlewareStack(
        URLRouter([
            # URLRouter just takes standard Django path() or url() entries.
path("websockettest/", EchoConsumer),
        ]),
    ),
})
url檔案增加websockettest.html訪問處理,在該頁面js中發起websocket連線,內容如下:
urlpatterns = [
    path('admin/', admin.site.urls),
    path('',AppMain_views.login),
    path('loginfunc/', AppMain_views.loginfunc, name='loginfunc'),
    path('websockettest/',AppMain_views.websockettest),
    path(r'^favicon\.ico$', RedirectView.as_view(url='/static/devifile_psp.ico')),
    url(r'^crossdomain.xml$',allow_domains,{'domains': ['172.20.0.42']}),
]
js中發起websocket連線,很簡單:
socket = new WebSocket("ws://127.0.0.1:9873/websockettest/");
socket.onmessage = function(e) {
    if (socket.readyState == WebSocket.OPEN)
    {
        alert("success... content="+e.data);
    }
    else
{
        alert("failed");
    }
    consoe.log(e.data);
}
socket.onopen = function() {
    socket.send("hello world");
}
最後還有一個響應類,實現websocket通訊,注意json資料一定要使用json.dumps,這個也是版本更新所致:
class EchoConsumer(JsonWebsocketConsumer):

    def websocket_connect(self, event):
        self.accept()

    def websocket_receive(self, event):
        self.send(json.dumps({
            "type": "websocket.send supcon",
            "text": event["text"],
        }))


    def websocket_message(self, event):
        self.send(json.dumps({
            "type": "websocket.send jackie",
            "text": event["text"],
        }))

    def websocket_disconnect(self, event):
        self.send(json.dumps({
            "type": "websocket.send",
            "text": event["text"],
        }))
到此,基本上是配置完成,服務執行即可實踐最新版本的websocket。
    原始碼連結