碼上快樂

相關內容    簡體    繁體

一篇搞定Python3.6+Django+channels實現WebSocket的實時聊天室

本文轉載自   查看原文   2020-12-11 23:24   425 

個人博客,歡迎來撩 fangzengye.com

1.第一部分：基礎設置

Channels3.0支持Python3.6和Django2.2+

1.1項目結構

mysite/
    manage.py
    mysite/
        __init__.py
        asgi.py
        settings.py
        urls.py
        wsgi.py

1.2自建項目主要文件夾

chat/
    __init__.py
    admin.py
    apps.py
    migrations/
        __init__.py
    models.py
    tests.py
    views.py

1.3移除不必須文件，保留必須文件，像這樣

chat/
    __init__.py
    views.py

1.4在INSTALLED_APPS加入項目名

# mysite/settings.py
INSTALLED_APPS = [
    'chat',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]

1.5創建聊天首頁view

1.5.1在templates文件夾創建一個html文件

chat/
    __init__.py
    templates/
        chat/
            index.html
    views.py

html代碼

<!-- chat/templates/chat/index.html -->
<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8"/>
    <title>Chat Rooms</title>
</head>
<body>
    What chat room would you like to enter?<br>
    <input id="room-name-input" type="text" size="100"><br>
    <input id="room-name-submit" type="button" value="Enter">

&lt;script&gt;
    document.querySelector('#room-name-input').focus();
    document.querySelector('#room-name-input').onkeyup = function(e) {
        if (e.keyCode === 13) {  // enter, return
            document.querySelector('#room-name-submit').click();
        }
    };

    document.querySelector('#room-name-submit').onclick = function(e) {
        var roomName = document.querySelector('#room-name-input').value;
        window.location.pathname = '/chat/' + roomName + '/';
    };
&lt;/script&gt;



</body>
 </html>

1.6在view創建接受前端請求，返回html

# chat/views.py
from django.shortcuts import render

def index(request):
 return render(request, 'chat/index.html')

1.7在路由端chat/urls.py添加路徑

# chat/urls.py
from django.urls import path

from . import views


urlpatterns = [
 path('', views.index, name='index'),
 ]

1.8在路由端mysite/urls.py添加路徑

# mysite/urls.py
from django.conf.urls import include
from django.urls import path
from django.contrib import admin

urlpatterns = [
 path('chat/', include('chat.urls')),
 path('admin/', admin.site.urls),
 ]

1.9可以運行了

python3 manage.py runserver

這里有兩個連接

http://127.0.0.1:8000/chat/

http://127.0.0.1:8000/chat/lobby/這個連接可能會404，因為還沒有整合channels

1.10修改mysite/Saginaw.py

# mysite/asgi.py
import os

from channels.routing import ProtocolTypeRouter
 from django.core.asgi import get_asgi_application


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mysite.settings')


application = ProtocolTypeRouter({
 "http": get_asgi_application(),
 # Just HTTP for now. (We can add other protocols later.)
 })

1.11在mysite/settings.py的INSTALLED_APP 添加channels

# mysite/settings.py
INSTALLED_APPS = [
    'channels',
    'chat',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]

1.12在mysite/settings.py添加指向Channels

# mysite/settings.py
# Channels
ASGI_APPLICATION = 'mysite.asgi.application'

1.13再次運行

python3 manage.py runserver

 

 

2.聊天室服務器接口

2.1添加一個聊天頁面視圖view

新建一個空html頁面文件chat/templates/chat/room.html

因此項目結構如下：

chat/
    __init__.py
    templates/
        chat/
            index.html
            room.html
    urls.py
    views.py

html代碼如下

<!-- chat/templates/chat/room.html -->
<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8"/>
    <title>Chat Room</title>
</head>
<body>
    <textarea id="chat-log" cols="100" rows="20"></textarea><br>
    <input id="chat-message-input" type="text" size="100"><br>
    <input id="chat-message-submit" type="button" value="Send">
    {{ room_name|json_script:"room-name" }}
    <script>
        const roomName = JSON.parse(document.getElementById('room-name').textContent);

    const chatSocket = new WebSocket(
        'ws://'
        + window.location.host
        + '/ws/chat/'
        + roomName
        + '/'
    );

    chatSocket.onmessage = function(e) {
        const data = JSON.parse(e.data);
        document.querySelector('#chat-log').value += (data.message + '\n');
    };

    chatSocket.onclose = function(e) {
        console.error('Chat socket closed unexpectedly');
    };

    document.querySelector('#chat-message-input').focus();
    document.querySelector('#chat-message-input').onkeyup = function(e) {
        if (e.keyCode === 13) {  // enter, return
            document.querySelector('#chat-message-submit').click();
        }
    };

    document.querySelector('#chat-message-submit').onclick = function(e) {
        const messageInputDom = document.querySelector('#chat-message-input');
        const message = messageInputDom.value;
        chatSocket.send(JSON.stringify({
            'message': message
        }));
        messageInputDom.value = '';
    };
&lt;/script&gt;



</body>
 </html>

這段前端Html挺重要的，實例化了WebSocket，添加onsend、onclosed、onmessage函數，與后端設置WebSocket連接起來了

 

2.2在chat/views.py添加romm函數

# chat/views.py
from django.shortcuts import render

def index(request):
 return render(request, 'chat/index.html', {})


def room(request, room_name):
 return render(request, 'chat/room.html', {
 'room_name': room_name
 })

2.3修改chat/urls.py

# chat/urls.py
from django.urls import path

from . import views


urlpatterns = [
 path('', views.index, name='index'),
 path('<str:room_name>/', views.room, name='room'),
 ]

2.4運行

python3 manage.py runserver

Go to http://127.0.0.1:8000/chat/我們可以看到首頁

http://127.0.0.1:8000/chat/lobby/可以看到展示一個空聊天記錄信息

2.5聊天室輸入hello回🚗沒有任何發生說明還沒啟動Websocket

到目前為止，只是在前端設置了WebSocket，還沒在后端設置WebSocket，因此還沒建立連接，接下來在后端設置WebSocket

2.6創建chat/consumer.py，像

chat/
    __init__.py
    consumers.py
    templates/
        chat/
            index.html
            room.html
    urls.py
    views.py

consumer.py代碼

# chat/consumers.py
import json
from channels.generic.websocket import WebsocketConsumer

class ChatConsumer(WebsocketConsumer):
 def connect(self):
 self.accept()


def disconnect(self, close_code):
    pass

def receive(self, text_data):
    text_data_json = json.loads(text_data)
    message = text_data_json['message']

    self.send(text_data=json.dumps({
        'message': message
    }))</code></pre> 



2.7創建chat/routing.py，像
 

chat/
    __init__.py
    consumers.py
    routing.py
    templates/
        chat/
            index.html
            room.html
    urls.py
    views.py
 

routing.py代碼
 

chat/routing.py
from django.urls import re_path

from . import consumers


websocket_urlpatterns = [
 re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()),
 ]


這里調用as_asgi（）為了適配ASGI應用
 

2.8修改mysite/asgi.py
 

mysite/asgi.py
import os

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


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


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


2.9使用redis作為消息存儲倉（backing store ）
 

2.10安裝channels_redis
 

Pip install channels_redis

 

2.11配置mysite/sttings.py，添加CHANNELS_LAYERS
 

mysite/settings.py
# Channels
ASGI_APPLICATION = 'mysite.asgi.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}
 

2.12打開python shell
 

$ python3 manage.py shell
>>> import channels.layers
>>> channel_layer = channels.layers.get_channel_layer()
>>> from asgiref.sync import async_to_sync
>>> async_to_sync(channel_layer.send)('test_channel', {'type': 'hello'})
>>> async_to_sync(channel_layer.receive)('test_channel')
{'type': 'hello'}
 

control+D退出shell
 

2.13現在有了chanel層，使用ChatConsumer函數，在chat/consumers.py上添加函數
 

# chat/consumers.py
import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer

class ChatConsumer(WebsocketConsumer):
 def connect(self):
 self.room_name = self.scope['url_route']['kwargs']['room_name']
 self.room_group_name = 'chat_%s' % self.room_name


    # Join room group
    async_to_sync(self.channel_layer.group_add)(
        self.room_group_name,
        self.channel_name
    )

    self.accept()

def disconnect(self, close_code):
    # Leave room group
    async_to_sync(self.channel_layer.group_discard)(
        self.room_group_name,
        self.channel_name
    )

# Receive message from WebSocket
def receive(self, text_data):
    text_data_json = json.loads(text_data)
    message = text_data_json['message']

    # Send message to room group
    async_to_sync(self.channel_layer.group_send)(
        self.room_group_name,
        {
            'type': 'chat_message',
            'message': message
        }
    )

# Receive message from room group
def chat_message(self, event):
    message = event['message']

    # Send message to WebSocket
    self.send(text_data=json.dumps({
        'message': message
    }))</code></pre> 



更深層次對函數解釋
 

self.scope['url_route']['kwargs']['room_name']
 

從已經建立好連接WebSocket的URL路由地址chat/routing.py獲取room_name參數
 

每個用戶都有一個保存自己連接的信息的scope
 

self.room_group_name = 'chat_%s' % self.room_name
 

建立一個Channels聊天組名
 

async_to_sync(self.channel_layer.group_add)(...)
 

加入一個group
 

任何一個聊天室都需要實時同步的async WebSocket方法
 

Group name 限制在ASCII字符
 

self.accept()
 

接收WebSocket連接
 

如果你沒有在connect（）中調用accept（）可能會導致連接關閉或被拒絕
 

如果你想要使用accept（），建議你在connect（）函數中最后使用（放在最后一行）
 

async_to_sync(self.channel_layer.group_discard)(...)
 

退出group群行為
 

async_to_sync(self.channel_layer.group_send)
 

發送一個事件到群group
 

如果返回的事件包含type鍵可能會引起用戶接受到次事件信息
 

 
 

2.14運行runserver
 

使用兩個瀏覽器同時訪問http://127.0.0.1:8000/chat/lobby/，在上面發送消息，可以看到另外瀏覽器收到消息
 

[refers](https://channels.readthedocs.io/en/latest/tutorial/part_2.html)
 

 
 

3.用Asynchronous重寫聊天室
 

3.1重寫ChatConsumer類，修改chat/consumers.py
 

chat/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
 async def connect(self):
 self.room_name = self.scope['url_route']['kwargs']['room_name']
 self.room_group_name = 'chat_%s' % self.room_name


    # Join room group
    await self.channel_layer.group_add(
        self.room_group_name,
        self.channel_name
    )

    await self.accept()

async def disconnect(self, close_code):
    # Leave room group
    await self.channel_layer.group_discard(
        self.room_group_name,
        self.channel_name
    )

# Receive message from WebSocket
async def receive(self, text_data):
    text_data_json = json.loads(text_data)
    message = text_data_json['message']

    # Send message to room group
    await self.channel_layer.group_send(
        self.room_group_name,
        {
            'type': 'chat_message',
            'message': message
        }
    )

# Receive message from room group
async def chat_message(self, event):
    message = event['message']

    # Send message to WebSocket
    await self.send(text_data=json.dumps({
        'message': message
    }))





簡化上面代碼，寫成如下框架
 

chat/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer

class ChatConsumer(AsyncWebsocketConsumer):
 async def connect():


    await self.channel_layer.group_add()

    await self.accept()

async def disconnect():

    await self.channel_layer.group_discard()

async def receive():

    await self.channel_layer.group_send()

async def chat_message(self, event):

    await self.send()





await 是調用同步asynshronous函數的I/O平台
 

 
 

[refers](https://channels.readthedocs.io/en/latest/tutorial/part_3.html)
 

 
 

4.自動化測試
 

4.1安裝selenium
 

pip install selenium
 

4.2創建新文件chat/test.py，文件架構如下
 

chat/
    __init__.py
    consumers.py
    routing.py
    templates/
        chat/
            index.html
            room.html
    tests.py
    urls.py
    views.py
 

修改chat/test.py
 

# chat/tests.py
from channels.testing import ChannelsLiveServerTestCase
from selenium import webdriver
from selenium.webdriver.common.action_chains import ActionChains
from selenium.webdriver.support.wait import WebDriverWait

class ChatTests(ChannelsLiveServerTestCase):
 serve_static = True # emulate StaticLiveServerTestCase


@classmethod
def setUpClass(cls):
    super().setUpClass()
    try:
        # NOTE: Requires "chromedriver" binary to be installed in $PATH
        cls.driver = webdriver.Chrome()
    except:
        super().tearDownClass()
        raise

@classmethod
def tearDownClass(cls):
    cls.driver.quit()
    super().tearDownClass()

def test_when_chat_message_posted_then_seen_by_everyone_in_same_room(self):
    try:
        self._enter_chat_room('room_1')

        self._open_new_window()
        self._enter_chat_room('room_1')

        self._switch_to_window(0)
        self._post_message('hello')
        WebDriverWait(self.driver, 2).until(lambda _:
            'hello' in self._chat_log_value,
            'Message was not received by window 1 from window 1')
        self._switch_to_window(1)
        WebDriverWait(self.driver, 2).until(lambda _:
            'hello' in self._chat_log_value,
            'Message was not received by window 2 from window 1')
    finally:
        self._close_all_new_windows()

def test_when_chat_message_posted_then_not_seen_by_anyone_in_different_room(self):
    try:
        self._enter_chat_room('room_1')

        self._open_new_window()
        self._enter_chat_room('room_2')

        self._switch_to_window(0)
        self._post_message('hello')
        WebDriverWait(self.driver, 2).until(lambda _:
            'hello' in self._chat_log_value,
            'Message was not received by window 1 from window 1')

        self._switch_to_window(1)
        self._post_message('world')
        WebDriverWait(self.driver, 2).until(lambda _:
            'world' in self._chat_log_value,
            'Message was not received by window 2 from window 2')
        self.assertTrue('hello' not in self._chat_log_value,
            'Message was improperly received by window 2 from window 1')
    finally:
        self._close_all_new_windows()

# === Utility ===

def _enter_chat_room(self, room_name):
    self.driver.get(self.live_server_url + '/chat/')
    ActionChains(self.driver).send_keys(room_name + '\n').perform()
    WebDriverWait(self.driver, 2).until(lambda _:
        room_name in self.driver.current_url)

def _open_new_window(self):
    self.driver.execute_script('window.open("about:blank", "_blank");')
    self.driver.switch_to_window(self.driver.window_handles[-1])

def _close_all_new_windows(self):
    while len(self.driver.window_handles) &gt; 1:
        self.driver.switch_to_window(self.driver.window_handles[-1])
        self.driver.execute_script('window.close();')
    if len(self.driver.window_handles) == 1:
        self.driver.switch_to_window(self.driver.window_handles[0])

def _switch_to_window(self, window_index):
    self.driver.switch_to_window(self.driver.window_handles[window_index])

def _post_message(self, message):
    ActionChains(self.driver).send_keys(message + '\n').perform()

@property
def _chat_log_value(self):
    return self.driver.find_element_by_css_selector('#chat-log').get_property('value')</code></pre> 



4.3搭載數據庫sqlite3，需要告訴Django要使用數據庫，在mysite/settings.py修改
 

mysite/settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': os.path.join(BASE_DIR, 'db.sqlite3'),
        'TEST': {
            'NAME': os.path.join(BASE_DIR, 'db_test.sqlite3')
        }
    }
}
 

 
 

4.4運行
 

python3 manage.py test chat.tests
 

您可以看到
 

Creating test database for alias 'default'...
System check identified no issues (0 silenced).
..
----------------------------------------------------------------------
Ran 2 tests in 5.014s

OK
 Destroying test database for alias 'default'...


4.5大功完成
 

[refers](https://channels.readthedocs.io/en/latest/tutorial/part_4.html)
 

4.6如您想要更深的了解可以訪問https://channels.readthedocs.io/en/latest/index.html#topics
 

后記
 

一直在做Django后台開發，但是，一開始並不知道WebSocket是啥，更不知道python的Django還有第三方庫Channels庫實現網頁全雙工WebSocket，是Amanda小姐姐告訴還有WebSocket可以聊解一下，沒想到，才意識到現在使用的微信、高德實時導航，QQ等這些實時通信工具都是基於這種原理實現。
 

https://channels.readthedocs.io/en/latest/tutorial/index.html
 

https://blog.csdn.net/weixin_43486863/article/details/83344368
 

 
 

 
 

 
 

 
 

 
 

 


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 Django實現WebSocket在線聊天室（channels庫） Django用websocket實現聊天室之築基篇 Django實現websocket完成實時通訊，聊天室，在線客服等 Django實現websocket完成實時通訊、聊天室、在線客服等 Django2.0.4 + websocket 實現實時通信，主動推送，聊天室及客服系統 Django實現websocket完成實時通訊、聊天室、在線客服等 Django中使用websocket並實現簡易聊天室 python 實現聊天室 php+websocket 實現聊天室 websocket實現多房間聊天室
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM