Django Channels Tutorial – 运维那点事

最近在实现数据库运维平台时需要用到 WebSocket，而原生的 Django 又不支持 WebSocket，仅有 Django Channels 库支持 WebSocket。

正常情况下，Django 使用 HTTP 请求实现客户端和服务器端的通信：

1. 客户端发送 HTTP 请求到服务器端

2. Django 解析请求，提取 URL，并将其和 View 进行匹配

3. View 处理请求并返回 HTTP Response 至客户端

不同于HTTP请求，WebSocket是一种在单个TCP连接上进行全双工通讯的协议。WebSocket允许服务端主动向客户端推送数据。在WebSocket协议中，客户端浏览器和服务器只需要完成一次握手就可以创建持久性的连接，并在浏览器和服务器之间进行双向的数据传输。

已经有网友整理翻译了 Django Channels Tutorial，个人也是摘抄测试了一遍。

一、教程

Channels 允许您在 Django 站点中使用 Websockets 和其他非 HTTP 协议。例如，您可能希望 Websockets 允许网站上的页面立即从 Django 服务器接收更新，而无需使用 HTTP 长轮询或其他昂贵的技术。

在本教程中，我们将建立一个简单的聊天服务器，在那里你可以加入一个在线房间，发送一些消息到这个房间，并让其他人在同一房间立即看到这些消息。

本教程分为四个部分：

教程部分一：基本设置
教程部分二：实现聊天服务器
教程部分三：将聊天服务器重写为异步
教程部分四：自动化测试

二、基本设置

在本教程中，我们将构建一个简单的聊天服务器。它将有两个页面：

一个 index 视图，用于输入要加入的聊天室的名称。
一个可以查看在特定聊天室中发送的消息的房间视图。

房间视图将使用 WebSocket 与 Django 服务器进行通信，并监听任何发送出来的消息。

我们假设您熟悉构建 Django 站点的基本概念。如果不是，建议您先完成 Django 教程，然后再回到本教程。

我们假设你已经安装了 Django。您可以通过在 shell 提示符下运行以下命令 (用 $ 前缀表示) 来查看您安装的 Django 版本：

$ python3 -m django --version

1	$ python3 -m django --version

我们还假设您已经安装了 Channels，您可以通过运行以下命令来查看 Channels 安装与否：

$ python3 -c 'import channels; print(channels.__version__)'

1	$ python3 -c 'import channels; print(channels.__version__)'

本教程是为 Channels 2.0 编写的，它支持 Python 3.5 + 和 Django 1.11 +。如果 Channels 版本不匹配，你可以使用本页左下角的版本切换器，或将 Channels 更新到最新版本，以参考您的 Channels 版本的教程。

本教程还使用 Docker 安装和运行 Redis，我们使用 Redis 作为 Channels 层的后端存储，它是我们在教程中使用的 Channels 库的可选组件。

1. 新建一个项目

如果您还没有 Django 项目，您将需要创建一个。

$ django-admin startproject mysite

1	$ django-admin startproject mysite

这将在当前目录中创建一个 mysite 目录，其中有以下内容：

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

mysite/

manage.py

mysite/

__init__.py

settings.py

urls.py

wsgi.py

2. 创建 chat 应用程序

我们会将聊天服务器的代码放在它自己的应用程序中。请确保您位于与 manage.py 相同的目录中，然后输入以下命令：

$ python3 manage.py startapp chat

1	$ python3 manage.py startapp chat

这将创建一个 chat 文件夹，它是像这样的：

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

chat/

__init__.py

admin.py

apps.py

migrations/

__init__.py

models.py

tests.py

views.py

为了达到本教程的目的，我们将只使用 chat/views.py 和 chat/__init__.py。因此，从 chat 目录中删除所有其他文件。删除不必要的文件后，chat 目录应如下所示：

chat/
    __init__.py
    views.py

chat/

__init__.py

views.py

我们需要告诉我们的项目 chat app 已经安装，编辑 mysite/settings.py 文件并将 ‘chat’ 添加到 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',
]

# mysite/settings.py

INSTALLED_APPS = [

'chat',

'django.contrib.admin',

'django.contrib.auth',

'django.contrib.contenttypes',

'django.contrib.sessions',

'django.contrib.messages',

'django.contrib.staticfiles',

]

3. 添加 index 视图

现在，我们将创建第一个视图，这个 index 视图允许您输入要加入的聊天室的名称。

在 chat 目录中创建 templates 目录，在刚刚创建的 templates 目录中，创建另一个名为 chat 的目录，并在其中创建一个名为 index.html 的文件。

您的 chat 目录现在应该看起来像：

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

chat/

__init__.py

templates/

chat/

index.html

views.py

将下面的代码写进 chat/templates/chat/index.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"/>
</body>
<script>
    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 + '/';
    };
</script>
</html>

<!DOCTYPE html>

<html>

<head>

<title>Chat Rooms</title>

</head>

<body>

What chat room would you like to enter?<br/>

</body>

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 + '/';

};

</script>

</html>

为 room 视图创建视图函数，将下面的代码写进 chat/views.py 文件中：

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

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

# chat/views.py

from django.shortcuts import render

def index(request):

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

为了调用这个视图，我们需要把它映射到一个 URL — 因此我们需要一个 URL 配置文件。

为了在 chat 目录下创建一个 URL 配置文件，我们需要新建一个名为 urls.py 的文件。你的 app 目录应该像现在这样子：

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

chat/

__init__.py

templates/

chat/

index.html

urls.py

views.py

在 chat/urls.py 文件中包含以下代码：

# chat/urls.py
from django.conf.urls import url

from . import views

urlpatterns = [
    url(r'^$', views.index, name='index'),
]

# chat/urls.py

from django.conf.urls import url

from . import views

urlpatterns = [

url(r'^$', views.index, name='index'),

]

下一步是将根目录下的 URLconf 文件指向 chat.urls 模块，在 mysite/urls.py 中导入 django.conf.urls.include模块，并在 urlpatterns 列表中插入一个 include() 函数，因此您需要写入以下代码：

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

urlpatterns = [
    url(r'^chat/', include('chat.urls')),
    url(r'^admin/', admin.site.urls),
]

# mysite/urls.py

from django.conf.urls import include, url

from django.contrib import admin

urlpatterns = [

url(r'^chat/', include('chat.urls')),

url(r'^admin/', admin.site.urls),

]

让我们验证 index 视图是否有效，运行以下命令：

$ python3 manage.py runserver

1	$ python3 manage.py runserver

您将在命令行中看到以下输出：

Performing system checks...

System check identified no issues (0 silenced).

You have 13 unapplied migration(s). Your project may not work properly until you apply the migrations for app(s): admin, auth, contenttypes, sessions.
Run 'python manage.py migrate' to apply them.

February 18, 2018 - 22:08:39
Django version 1.11.10, using settings 'mysite.settings'
Starting development server at http://127.0.0.1:8000/
Quit the server with CONTROL-C.

Performing system checks...

System check identified no issues (0 silenced).

You have 13 unapplied migration(s). Your project may not work properly until you apply the migrations for app(s): admin, auth, contenttypes, sessions.

Run 'python manage.py migrate' to apply them.

February 18, 2018 - 22:08:39

Django version 1.11.10, using settings 'mysite.settings'

Starting development server at http://127.0.0.1:8000/

Quit the server with CONTROL-C.

注意：忽略有关未应用数据库迁移的警告，我们将不会在本教程中使用数据库。

在浏览器中转到 http://127.0.0.1:8000/chat/，您应该看到文本 “What chat room would you like to enter?” 以及一个用于输入房间名字的文本输入框。

输入 “lobby” 作为房间名称，然后按 enter 键。你应该被重定向到 http://127.0.0.1:8000/chat/lobby/ 的房间视图，但我们还没有写的房间视图，所以你会得到一个 “页面找不到” 错误页面。

然后可以转到运行 runserver 命令的终端，按下 Control+C 以停止服务器。

4. 集成 Channels 库

到目前为止，我们刚刚创建了一个常规的 Django 应用程序；我们根本就没有使用 Channels 库。现在是时候集成 Channels 库了。

让我们从创建 Channels 的根路由配置文件开始，Channels 路由配置类似于 Django URLconf，它会告诉 Channels 当收到由 Channels 服务器发过来的 HTTP 请求时，应该执行什么代码。

我们将从一个空的路由配置文件开始，创建文件 mysite/routing.py，并写入以下代码：

# mysite/routing.py
from channels.routing import ProtocolTypeRouter

application = ProtocolTypeRouter({
    # (http->django views is added by default)
})

# mysite/routing.py

from channels.routing import ProtocolTypeRouter

application = ProtocolTypeRouter({

# (http->django views is added by default)

})

现在，将 Channels 库添加到已安装的应用程序列表中。编辑 mysite/settings.py 文件并将 ‘channels’ 添加到 INSTALLED_APPS 设置。它看起来像这样：

# 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',
]

# 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',

]

您同样需要在根路由配置中指向 Channels，再次编辑 mysite/settings.py 文件，并将以下内容添加到底部：

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

# mysite/settings.py

# Channels

ASGI_APPLICATION = 'mysite.routing.application'

现在已安装的应用程序中有 Channels，它将控制 runserver 命令，用 Channels 开发服务器替换标准的 Django 开发服务器。

PS：Channels 开发服务器将与需要重载或替换 runserver 命令的任何其他第三方应用程序冲突。whitenoise 中的 whitenoise.runserver_nostatic是一个冲突的例子。为了解决这些问题，请尝试将 Channels 移动到您的 INSTALLED_APPS 的顶部，或者完全删除与其发生冲突的应用程序。

让我们确保 Channels 开发服务器工作正常。运行以下命令：

$ python3 manage.py runserver

1	$ python3 manage.py runserver

您将在命令行中看到以下输出：

....
Django version 1.11.10, using settings 'mysite.settings'
Starting ASGI/Channels development server at http://127.0.0.1:8000/
Quit the server with CONTROL-C.
2018-02-18 22:16:23,729 - INFO - server - HTTP/2 support not enabled (install the http2 and tls Twisted extras)
2018-02-18 22:16:23,730 - INFO - server - Configuring endpoint tcp:port=8000:interface=127.0.0.1
2018-02-18 22:16:23,731 - INFO - server - Listening on TCP address 127.0.0.1:8000

....

Django version 1.11.10, using settings 'mysite.settings'

Starting ASGI/Channels development server at http://127.0.0.1:8000/

Quit the server with CONTROL-C.

2018-02-18 22:16:23,729 - INFO - server - HTTP/2 support not enabled (install the http2 and tls Twisted extras)

2018-02-18 22:16:23,730 - INFO - server - Configuring endpoint tcp:port=8000:interface=127.0.0.1

2018-02-18 22:16:23,731 - INFO - server - Listening on TCP address 127.0.0.1:8000

留意从 Starting ASGI/Channels development server at http://127.0.0.1:8000/ 开始的内容，这表明 Channels 开发服务器已接管了 Django 开发服务器。

在浏览器中转到 http://127.0.0.1:8000/chat/，您仍然应该看到我们以前创建的 index 页面。

然后可以转到运行 runserver 命令的终端，按下 Control+C 以停止服务器。

三、实现聊天服务器

本教程在教程1的基础上开始。我们会让房间页面工作，这样你可以和你自己或者其他人在同一个房间里聊天。

1. 添加房间视图

现在，我们将创建第二个视图，即一个允许您查看在特定聊天室中发布消息的房间视图。

创建新的文件 chat/templates/chat/room.html，您的应用程序目录现在应该看起来像：

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

chat/

__init__.py

templates/

chat/

index.html

room.html

urls.py

views.py

在 chat/templates/chat/room.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"/>
</body>
<script>
    var roomName = {{ room_name_json }};

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

    chatSocket.onmessage = function(e) {
        var data = JSON.parse(e.data);
        var message = data['message'];
        document.querySelector('#chat-log').value += (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) {
        var messageInputDom = document.querySelector('#chat-message-input');
        var message = messageInputDom.value;
        chatSocket.send(JSON.stringify({
            'message': message
        }));

        messageInputDom.value = '';
    };
</script>
</html>

<!DOCTYPE html>

<html>

<head>

</head>

<body>

</body>

var roomName = {{ room_name_json }};

var chatSocket = new WebSocket(

'ws://' + window.location.host +

'/ws/chat/' + roomName + '/');

chatSocket.onmessage = function(e) {

var data = JSON.parse(e.data);

var message = data['message'];

document.querySelector('#chat-log').value += (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) {

var messageInputDom = document.querySelector('#chat-message-input');

var message = messageInputDom.value;

chatSocket.send(JSON.stringify({

'message': message

}));

messageInputDom.value = '';

};

</script>

</html>

在 chat/views.py 中为房间视图创建视图函数，添加导入 mark_safe 和 json 模块，并添加房间视图的视图函数：

# chat/views.py
from django.shortcuts import render
from django.utils.safestring import mark_safe
import json

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

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

# chat/views.py

from django.shortcuts import render

from django.utils.safestring import mark_safe

import json

def index(request):

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

def room(request, room_name):

return render(request, 'chat/room.html', {

'room_name_json': mark_safe(json.dumps(room_name))

})

在 chat/urls.py 中创建房间视图的路由：

# chat/urls.py
from django.conf.urls import url

from . import views

urlpatterns = [
    url(r'^$', views.index, name='index'),
    url(r'^(?P<room_name>[^/]+)/$', views.room, name='room'),
]

# chat/urls.py

from django.conf.urls import url

from . import views

urlpatterns = [

url(r'^$', views.index, name='index'),

url(r'^(?P<room_name>[^/]+)/$', views.room, name='room'),

]

启动 Channels 开发服务器：

$ python3 manage.py runserver

1	$ python3 manage.py runserver

在浏览器中转到 http://127.0.0.1:8000/chat/ 并查看 index 页面。

输入 “lobby” 作为房间名称，然后按 enter 键。您将会重定向到 http://127.0.0.1:8000/chat/lobby/，该页面现在显示一个空的聊天日志。

键入消息 “hello”，然后按 enter 键。什么也没有发生，尤其是消息并不会出现在聊天日志中，为什么？

房间视图试图打开一个 WebSocket 连接到 URL ws://127.0.0.1:8000/ws/chat/lobby/，但我们还没有创建一个接受 WebSocket 连接的 consumer。如果打开浏览器的 JavaScript 控制台，您应该会看到如下所示的错误：

WebSocket connection to 'ws://127.0.0.1:8000/ws/chat/lobby/' failed: Unexpected response code: 500

1	WebSocket connection to 'ws://127.0.0.1:8000/ws/chat/lobby/' failed: Unexpected response code: 500

2. 编写第一个 consumer

当 Django 接受 HTTP 请求时，它会根据根 URLconf 以查找视图函数，然后调用视图函数来处理请求。同样，当 Channels 接受 WebSocket 连接时，它会根据根路由配置以查找对应的 consumer，然后调用 consumer 上的各种函数来处理来自这个连接的事件。

我们将编写一个简单的 consumer，它会在路径 /ws/chat/ROOM_NAME/ 接收 WebSocket 连接，然后把接收任意的消息回送给同一个 WebSocket 连接。

提醒：

使用常见的路径前缀 (如/ws) 来区分 WebSocket 连接与普通 HTTP 连接是很好的做法，因为它将使在某些配置中部署 Channels 更容易。

特别是大型网站，它们很有可能配置像 nginx 这样的生产级别 HTTP 服务器，根据路径将请求发送到生产级别的 WSGI 服务器，例如用于处理普通 HTTP 请求的 Gunicorn + Django，或生产级别的 ASGI 服务器，例如用于处理 WebSocket 请求的 Daphne + Channels。

请注意，对于较小的站点，您可以使用更简单的部署策略，其中 Daphne 服务器处理所有的请求 — HTTP 和 WebSocket — 而不是单独的 WSGI 服务器。在这种部署配置策略中，不需要使用 /ws/ 这样的通用路径前缀。

创建新文件 chat/consumers.py，您的应用程序目录现在应该看起来像：

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

chat/

__init__.py

consumers.py

templates/

chat/

index.html

room.html

urls.py

views.py

在 chat/consumers.py 中写入以下代码：

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

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
        }))

# chat/consumers.py

from channels.generic.websocket import WebsocketConsumer

import json

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

}))

这是一个同步 WebSocket consumer，它接受所有连接，接收来自其客户端的消息，并将这些消息回送到同一客户端。现在，它不向同一个房间的其他客户端广播消息。

提醒：Channels 还支持编写异步 consumers 以提高性能。但是，任何异步 consumers 都必须小心，避免直接执行阻塞操作，例如访问 Django 的 model。有关编写异步 consumers 的详细信息，请参阅 Consumers。

我们需要为 chat 应用程序创建一个路由配置，它有一个通往 consumer 的路由。创建新文件 chat/routing.py。您的应用程序目录现在应该看起来像：

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

chat/

__init__.py

consumers.py

routing.py

templates/

chat/

index.html

room.html

urls.py

views.py

在 chat/routing.py 中输入以下代码：

# chat/routing.py
from django.conf.urls import url

from . import consumers

websocket_urlpatterns = [
    url(r'^ws/chat/(?P<room_name>[^/]+)/$', consumers.ChatConsumer),
]

# chat/routing.py

from django.conf.urls import url

from . import consumers

websocket_urlpatterns = [

url(r'^ws/chat/(?P<room_name>[^/]+)/$', consumers.ChatConsumer),

]

下一步是将根路由指向 chat.routing 模块。在 mysite/routing.py 中导入 AuthMiddlewareStack、URLRouter 和 chat.routing；并在 ProtocolTypeRouter 列表中插入一个 “websocket” 键，格式如下：

# mysite/routing.py
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import chat.routing

application = ProtocolTypeRouter({
    # (http->django views is added by default)
    'websocket': AuthMiddlewareStack(
        URLRouter(
            chat.routing.websocket_urlpatterns
        )
    ),
})

# mysite/routing.py

from channels.auth import AuthMiddlewareStack

from channels.routing import ProtocolTypeRouter, URLRouter

import chat.routing

application = ProtocolTypeRouter({

# (http->django views is added by default)

'websocket': AuthMiddlewareStack(

URLRouter(

chat.routing.websocket_urlpatterns

)

})

这个根路由配置指定当与 Channels 开发服务器建立连接的时候，ProtocolTypeRouter 将首先检查连接的类型。如果是 WebSocket 连接 (ws://或 wss://)，则连接会交给 AuthMiddlewareStack。

AuthMiddlewareStack 将使用对当前经过身份验证的用户的引用来填充连接的 scope，类似于 Django 的 AuthenticationMiddleware 用当前经过身份验证的用户填充视图函数的请求对象（Scopes 将在本教程后面讨论），然后连接将被给到 URLRouter。

根据提供的 url 模式，URLRouter 将检查连接的 HTTP 路径，以将其路由指定到到特定的 consumer。

让我们验证 consumer 的 /ws/chat/ROOM_NAME/ 路径是否工作，启动 Channels 开发服务器：

$ python3 manage.py runserver

1	$ python3 manage.py runserver

转到 http://127.0.0.1:8000/chat/lobby/ 中的房间页面，该页现在显示一个空的聊天日志。

输入消息 “hello”，然后按 enter 键。您现在应该看到 “hello” 在聊天日志中显示。

但是，如果您打开第二个浏览器选项卡输入 http://127.0.0.1:8000/chat/lobby/ 进入同一房间页面上并输入消息，则消息并不会出现在第一个选项卡中。为了做到这一点，我们需要有多个相同 ChatConsumer 实例才能互相交谈。Channels 提供了一种 channel layer 抽象，使 consumers 之间能够进行这种通信。

然后可以转到运行 runserver 命令的终端，按下 Control+C 以停止服务器。

3. 启用 channel layer

channel layer 是一种通信系统，它允许多个 consumer 实例互相交谈，以及与 Django 的其他部分进行通信。

channel layer 提供以下抽象层：

channel 是可以发送消息的邮箱，每个 channel 都有一个名称。任何有名称的 channel 都可以向其他 channel 发送消息。

group 是一组相关的 channels，group 具有名称。任何具有名字的 group 都可以按名称向 group 中添加/删除 channel，也可以向 group 中的所有 channel 发送消息。无法列举特定 group 中的 channel。

每个 consumer 实例都有一个自动生成的唯一的 channel 名称，因此可以通过 channel layer 进行通信。

在我们的聊天应用程序中，我们希望在同一房间中有多个 ChatConsumer 的实例相互通信。要做到这一点，我们将为每个 ChatConsumer 添加它的 channel 到一个 group 中，其名称是基于房间的名称。这将允许 ChatConsumers 将消息传输到同一个房间中的所有其他 ChatConsumers。

我们将使用一个 channel layer，并使用 Redis 作为其后端存储。要在端口6379上启动 Redis 服务器，请运行以下命令：

$ docker run -p 6379:6379 -d redis:2.8

1	$ docker run -p 6379:6379 -d redis:2.8

然后，我们需要安装 channels_redis，以便 Channels 知道如何调用 redis。运行以下命令：

$ pip3 install channels_redis

1	$ pip3 install channels_redis

在使用 channel layer 之前，必须对其进行配置。编辑 mysite/settings.py 文件并将 CHANNEL_LAYERS 设置添加到底部。它应该看起来像：

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

# mysite/settings.py

# Channels

ASGI_APPLICATION = 'mysite.routing.application'

CHANNEL_LAYERS = {

'default': {

'BACKEND': 'channels_redis.core.RedisChannelLayer',

'CONFIG': {

"hosts": [('127.0.0.1', 6379)],

}

提醒：可以配置多个 channel layer。然而，大多数项目只使用一个 “默认” 的 channel layer。

让我们确保 channel layer 可以与 Redis 通信。打开 Django 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'}

$ 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 退出 Django shell。

现在我们有了一个 channel layer，让我们在 ChatConsumer 中使用它。将以下代码放在 chat/consumers.py 中，替换旧代码：

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

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
        }))

# chat/consumers.py

from asgiref.sync import async_to_sync

from channels.generic.websocket import WebsocketConsumer

import json

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

}))

当用户发布消息时，JavaScript 函数将通过 WebSocket 将消息传输到 ChatConsumer。ChatConsumer 将接收该消息并将其转发到与房间名称对应的 group。在同一 group 中的每个 ChatConsumer (并因此在同一个房间中) 将接收来自该 group 的消息，通过 WebSocket 将其转发并返回到 JavaScript，它将会追加到聊天日志中。

新的 ChatConsumer 代码中有几个部分需要进一步解释：

self.scope[‘url_route’][‘kwargs’][‘room_name’]

从给 consumer 打开 WebSocket 连接的 chat/routes.py 中的 URL 路由中获取 “room_name” 参数。

每个 consumer 都有一个 scope，其中包含有关其连接的信息，特别是来自 URL 路由和当前经过身份验证的用户 (如果有的话) 中的任何位置或关键字参数。

self.room_group_name = ‘chat_%s’ % self.room_name

直接从用户指定的房间名称构造一个 Channels group 名称，无需任何引用或转义。

组名可能只包含字母、数字、连字符和句点。因此，此示例代码将在具有其他字符的房间名称上发生失败。

async_to_sync(self.channel_layer.group_add)(…)

加入一个 group。

async_to_sync(…) wrapper 是必需的，因为 ChatConsumer 是同步 WebsocketConsumer，但它调用的是异步 channel layer 方法（所有 channel layer 方法都是异步的）。

group 名称仅限于 ASCII 字母、连字符和句点。由于此代码直接从房间名称构造 group 名称，因此如果房间名称中包含的其他无效的字符，代码运行则会失败。

self.accept()

接收 WebSocket 连接。

如果你在 connect() 方法中不调用 accept()，则连接将被拒绝并关闭。例如，您可能希望拒绝连接，因为请求的用户未被授权执行请求的操作。

如果你选择接收连接，建议 accept() 作为在 connect() 方法中的最后一个操作。

async_to_sync(self.channel_layer.group_discard)(…)

离开一个 group。

async_to_sync(self.channel_layer.group_send)

将 event 发送到一个 group。event 具有一个特殊的键 ‘type’ 对应接收 event 的 consumers 调用的方法的名称。

让我们验证新 consumer 的 /ws/chat/ROOM_NAME/ 路径是否工作。要启动 Channels 开发服务器，请运行以下命令：

$ python3 manage.py runserver

1	$ python3 manage.py runserver

打开浏览器选项卡到 http://127.0.0.1:8000/chat/lobby/ 的房间页面。打开另一个浏览器选项卡到同一个房间页面。

在第二个浏览器选项卡中，输入消息 “hello”，然后按 enter 键。在第二个浏览器选项卡和第一个浏览器选项卡中，您现在应该看到 “hello” 在聊天日志中显示。

您现在有一个基本的功能齐全的聊天服务器！

四、将聊天服务器重写为异步方式

本教程在教程2的基础上开始。我们将重写 consumer 代码使其变成是异步的而不是同步的，以提高其性能。

1. 将 consumer 改写为异步

我们编写的 ChatConsumer 当前是同步的，同步的 consumers 很方便，因为它们可以调用常规的同步 I/O 函数，例如访问 Django models 而不用编写特殊的代码。但是，异步的 consumers 可以提供更高级别的性能，因为它们在处理请求时不需要创建其他线程。

ChatConsumer 只使用 async-native 库 (Channels 和 channel layer)，特别是它不访问同步的 Django models。因此，它可以被改写为异步的而不会变得复杂化。

提醒：即使 ChatConsumer 访问 Django models 或其他同步的代码，它仍然可以将其重写为异步的。像 asgiref.sync.sync_to_async 和 channels.db.database_sync_to_async 这样的实用工具可以用来从异步 consumer 那里调用同步的代码。但是，性能增益将小于仅使用 async-native 库的方式。

让我们重写 ChatConsumer 使其变为异步的，在 chat/consumers.py 中输入以下代码：

# chat/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
import json
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

from channels.generic.websocket import AsyncWebsocketConsumer

import json

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

}))

这些用于 ChatConsumer 的新代码与原始代码非常相似，它们具有以下差异：

现在 ChatConsumer 继承自 AsyncWebsocketConsumer 而不是 WebsocketConsumer。
所有方法都是 async def，而不仅仅是 def。
await 被用于调用执行 I/O 的异步函数。
在 channel layer 上调用方法时，不再需要 async_to_sync。

让我们验证 consumer 的 /ws/chat/ROOM_NAME/ 路径是否仍然有效。启动 Channels 开发服务器，运行以下命令：

$ python3 manage.py runserver

1	$ python3 manage.py runserver

打开浏览器选项卡到 http://127.0.0.1:8000/chat/lobby/ 的房间页面。打开另一个浏览器选项卡到同一个房间页面。

现在，您的聊天服务器是完全异步的了！

五、自动化测试

本教程在教程3的基础上开始。我们已经建立了一个简单的聊天服务器，现在我们将为它创建一些自动化测试。

1. 测试视图

为了确保聊天服务器能够继续工作，我们将编写一些测试。

我们将编写一套端到端的测试，使用 Selenium 来控制 Chrome web 浏览器。这些测试将确保：

当一个聊天信息被发布，然后它能被大家在同一房间看到
当一个聊天信息被发布，那么它在不同的房间是不会被别人看到的

如果您尚未拥有 Chrome 浏览器，请安装它。

安装 chromedriver，根据自己的系统下载对应二进制版本进行安装，需翻墙。

安装 Selenium，运行以下命令：

$ pip3 install selenium

1	$ pip3 install selenium

创建新的文件 chat/tests.py。您的应用程序目录现在应该看起来像：

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

chat/

__init__.py

consumers.py

routing.py

templates/

chat/

index.html

room.html

tests.py

urls.py

views.py

在 chat/tests.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) > 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')

# 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) > 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')

我们的测试套件扩展了 ChannelsLiveServerTestCase，而不是 Django 常用来进行端到端测试的套件 (StaticLiveServerTestCase 或 LiveServerTestCase)。这样，Channels 路由配置里面的 URLs（如 /ws/room/ROOM_NAME/ ）将会在套件里面工作。

要运行测试，请运行以下命令：

$ python3 manage.py test chat.tests

1	$ 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'...

Creating test database for alias 'default'...

System check identified no issues (0 silenced).

----------------------------------------------------------------------

Ran 2 tests in 5.014s

Destroying test database for alias 'default'...

你现在有一个经过测试的聊天服务器了!

接下来应该做什么呢？

祝贺！您已经完全实现了一个聊天服务器，通过在异步样式中编写它来高性能，并编写了自动测试以确保它不会中断。

这是教程的结尾。现在，你应该清楚地知道如何启动一个使用了 Channels 的你自己的应用程序和做其他的操作。当您需要学习新的技巧时，请回到文档的其余部分。

<补充>

如果想使用Django Channels实现一个实时信息打印到前端，简单可以使用Redis发布订阅功能。

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.stdout.decode()
        # }))

        rc = redis.Redis(host='10.10.0.1',
                         port=6379)
        p = rc.pubsub()
        p.subscribe('chat')
        for item in p.listen():
            if item['type'] == 'message' or item['type'] == 'pmessage':
                self.send(text_data=json.dumps({'message': bytes.decode(item['data'])}))