Django Channels 入门

作者: 爱吃鱼de大猫 | 来源:发表于2018-11-03 11:33 被阅读49次

原文：https://realpython.com/getting-started-with-django-channels/

本文中，我们将使用 Django Channels来构建一个实时应用程序：当客户端上线或下线时，实时更新用户列表数据。使用 WebSockets (通过 Django Channels) 技术进行客户端和服务器之间的通信，当有客户端上线，服务器会向所有连接的客户端发送一个广播，并自动更新客户端屏幕显示而不用刷新页面。

理解本文需要的知识储备：

Django 开发经验
WebSocket 概念

项目任务：

为 Django 项目添加 WebSocket 的支持（通过 Django Channels）
Django 使用 Redis，建立简单的连接
实现基本的用户身份验证
使用 Django 信号（Django Signals）机制来操作用户上下线的动作

将要用到的工具包：

Python (v3.6.0)
Django (v1.10.5)
Django Channels (v1.0.3)
Redis (v3.2.8)

开始

首先创建一个新的虚拟环境来隔离我们项目的依赖包的安装

$ mkdir django-example-channels 
$ cd django-example-channels 
$ python3.6 -m venv env 
$  source env/bin/activate 
(env)$

安装 Django, Django Channels, and ASGI Redis，创建一个新的 Django 项目和 app

（ENV）$ PIP安装django的== 1个 .10.5 通道== 1 .0.2 asgi_redis == 1 .0.0
（ENV）$ django-admin.py startproject命令example_channels
（ENV）$  CD example_channels
（ENV）$蟒manage.py的startApp example
（env）$ python manage.py migrate

下载安装 Redis

启动 Redis 服务默认使用 6379 端口，Django 将使用该端口连接 Redis 服务。

更新项目配置文件 settings.py 中的 INSTALLED_APPS 项

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'channels',
    'example',
]

配置 CHANNEL_LAYERS：

CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'asgi_redis.RedisChannelLayer',
        'CONFIG': {
            'hosts': [('localhost', 6379)],
        },
        'ROUTING': 'example_channels.routing.channel_routing',
    }
}

WebSocket 101

通常 Django 使用HTTP在客户机和服务器之间通信：

客户端发送 HTTP 请求
Django 解析请求，提取 URL 然后将它匹配到一个视图函数进行处理
视图处理请求并返回 HTTP 响应

与HTTP不同的是WebSocket协议允许双向通信，这意味着服务器可以将数据推送到客户端，而无需用户请求。HTTP中只有客户端请求然后得到响应，而WebSocket协议中，服务器可以同时与多个客户端进行通信，下面我们将要演示的使用ws://前缀，而不是http://

有什么不清楚的请自行查阅相关CHANNEL文档

Consumers and Groups

新建一个文件 example_channels/example/consumers.py，创建首个 consumer，它负责处理客户端和服务器的基础连接。

from channels import Group


def ws_connect(message):
    Group('users').add(message.reply_channel)


def ws_disconnect(message):
    Group('users').discard(message.reply_channel)

Consumer 对应到Django的视图，任何连接到服务器的客户端用户将被添加到“users”群组，可以接收到服务器发送的信息。当客户端离线时，该用户通道（channel）将会被移除出群组中，用户无法接收到信息。

接下来，进行路由的设置，它的工作方式与Django URL配置几乎相同，将以下代码添加到 example_channels/routing.py 这个新文件中：

from channels.routing import route
from example.consumers import ws_connect, ws_disconnect


channel_routing = [
    route('websocket.connect', ws_connect),
    route('websocket.disconnect', ws_disconnect),
]

上面我们通过定义一个 channel_routing 替换 urlpatterns ，用 route() 替换掉 url()。将我们的 consumer 处理函数匹配到 WebSockets。

模板

编写可以进行 WebSockets的Html 代码，构建项目模板文件夹 example_channels/example/templates/example，新建：

a _base.html：

<!doctype html>
<html lang="en">
<head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, user-scalable=no, initial-scale=1.0, maximum-scale=1.0, minimum-scale=1.0">
  <meta http-equiv="X-UA-Compatible" content="ie=edge">
  <link href="//maxcdn.bootstrapcdn.com/bootstrap/3.3.7/css/bootstrap.min.css" rel="stylesheet">
  <title>Example Channels</title>
</head>
<body>
  <div class="container">
    <br>
    {% block content %}{% endblock content %}
  </div>
  <script src="//code.jquery.com/jquery-3.1.1.min.js"></script>
  {% block script %}{% endblock script %}
</body>
</html>

user_list.html：

{% extends 'example/_base.html' %}

{% block content %}{% endblock content %}

{% block script %}
  <script>
    var socket = new WebSocket('ws://' + window.location.host + '/users/');

    socket.onopen = function open() {
      console.log('WebSockets connection created.');
    };

    if (socket.readyState == WebSocket.OPEN) {
      socket.onopen();
    }
  </script>
{% endblock script %}

现在，当我们的客户端成功使用 WebSocket 建立和服务器的连接时，我们可以在后台的命令行看到相应信息。

视图

在example_channels/example/views.py文件中，创建支持Django视图的模板渲染的代码：

from django.shortcuts import render


def user_list(request):
    return render(request, 'example/user_list.html')

将URL添加到 example_channels/example/urls.py中：

from django.conf.urls import url
from example.views import user_list


urlpatterns = [
    url(r'^$', user_list, name='user_list'),
]

将 example_channels/example_channels/urls.py中的地址，更新到项目的 URL 中：

from django.conf.urls import include, url
from django.contrib import admin

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

测试

启动项目进行测试：

(env)$ python manage.py runserver

您也可以在两个不同的终端上运行 pythonManage.py runserver-noWorker 和 pythonManage.py runWorker，以作为两个独立的进程测试接口服务器和工作服务器。两种方法都有效！

现在你访问 http://localhost:8000/ 在后台的命令行终端应该可以看到类似以下的信息：

[2018/02/19 23:24:57] HTTP GET / 200 [0.02, 127.0.0.1:52757]

[2018/02/19 23:24:58] WebSocket HANDSHAKING /users/ [127.0.0.1:52789]

[2018/02/19 23:25:03] WebSocket DISCONNECT /users/ [127.0.0.1:52789]

用户身份验证

接下来我们需要做的就是处理用户的身份验证，我们目标是用户登录到系统后，能够看到本组中其他成员的列表。首先构建用户创建账号和登录的方式，新建一个简单的登录页面，用户可以通过账号和密码进行登录。

example_channels/example/templates/example/log_in.html：

{% extends 'example/_base.html' %}

{% block content %}
  <form action="{% url 'example:log_in' %}" method="post">
    {% csrf_token %}
    {% for field in form %}
      <div>
        {{ field.label_tag }}
        {{ field }}
      </div>
    {% endfor %}
    <button type="submit">Log in</button>
  </form>
  <p>Don't have an account? <a href="{% url 'example:sign_up' %}">Sign up!</a></p>
{% endblock content %}

接下来更新 example_channels/example/views.py：

from django.contrib.auth import login, logout
from django.contrib.auth.forms import AuthenticationForm
from django.core.urlresolvers import reverse
from django.shortcuts import render, redirect


def user_list(request):
    return render(request, 'example/user_list.html')


def log_in(request):
    form = AuthenticationForm()
    if request.method == 'POST':
        form = AuthenticationForm(data=request.POST)
        if form.is_valid():
            login(request, form.get_user())
            return redirect(reverse('example:user_list'))
        else:
            print(form.errors)
    return render(request, 'example/log_in.html', {'form': form})


def log_out(request):
    logout(request)
    return redirect(reverse('example:log_in'))

Django 本身自带通用身份验证表单功能，我们可以用它来提供用户的登录验证。表单检验用户的账号和密码是否匹配，验证通过后返回一个 User对象。用户登录后将重定向到项目的主页。用户也应该可以进行注销的操作，所以我们继续创建一个注销视图，用户注销后将转回登录页面。

更新 example_channels/example/urls.py：

from django.conf.urls import url
from example.views import log_in, log_out, user_list


urlpatterns = [
    url(r'^log_in/$', log_in, name='log_in'),
    url(r'^log_out/$', log_out, name='log_out'),
    url(r'^$', user_list, name='user_list')
]

我们还需要一个注册页面来提供新用户注册，example_channels/example/templates/example/sign_up.html：

{% extends 'example/_base.html' %}

{% block content %}
  <form action="{% url 'example:sign_up' %}" method="post">
    {% csrf_token %}
    {% for field in form %}
      <div>
        {{ field.label_tag }}
        {{ field }}
      </div>
    {% endfor %}
    <button type="submit">Sign up</button>
    <p>Already have an account? <a href="{% url 'example:log_in' %}">Log in!</a></p>
  </form>
{% endblock content %}

登录和注册页面类似并相互链接。然后在视图中加入函数：

def sign_up(request):
    form = UserCreationForm()
    if request.method == 'POST':
        form = UserCreationForm(data=request.POST)
        if form.is_valid():
            form.save()
            return redirect(reverse('example:log_in'))
        else:
            print(form.errors)
    return render(request, 'example/sign_up.html', {'form': form})

同样我们使用自带的表单来提供用户注册处理，注册成功后将定向到登录页面。要记得在代码中导入表单模块：

from  django.contrib.auth.forms  import  AuthenticationForm,  UserCreationForm

再次更新 example_channels/example/urls.py：

from django.conf.urls import url
from example.views import log_in, log_out, sign_up, user_list


urlpatterns = [
    url(r'^log_in/$', log_in, name='log_in'),
    url(r'^log_out/$', log_out, name='log_out'),
    url(r'^sign_up/$', sign_up, name='sign_up'),
    url(r'^$', user_list, name='user_list')
]

到此，我们重新打开浏览器访问 http://localhost:8000/sign_up/ ，填好注册信息就创建我们第一个注册用户。（默认用户是michael，密码 johnson123）Sign_up视图将我们重定向到log_in视图，我们可以对新创建的用户进行身份验证。登录后，我们可以测试新的身份验证视图。然后使用“注册”表单创建几个新用户，为下一节做准备。

登录提醒

我们已经构建了基本的登录验证功能，但还没有完成用户列表的显示，还要实现当用户登录下线时服务器自动更新这个列表。接下来，我们将更新消费者函数，以便当用户登录或退出时发送通知消息。该消息包括用户名和连接状态信息。

example_channels/example/consumers.py：

import json
from channels import Group
from channels.auth import channel_session_user, channel_session_user_from_http


@channel_session_user_from_http
def ws_connect(message):
    Group('users').add(message.reply_channel)
    Group('users').send({
        'text': json.dumps({
            'username': message.user.username,
            'is_logged_in': True
        })
    })


@channel_session_user
def ws_disconnect(message):
    Group('users').send({
        'text': json.dumps({
            'username': message.user.username,
            'is_logged_in': False
        })
    })
    Group('users').discard(message.reply_channel)

example_channels/example/templates/example/user_list.html：

{% extends 'example/_base.html' %}

{% block content %}
  <a href="{% url 'example:log_out' %}">Log out</a>
  <br>
  <ul>
    {% for user in users %}
      <!-- NOTE: We escape HTML to prevent XSS attacks. -->
      <li data-username="{{ user.username|escape }}">
        {{ user.username|escape }}: {{ user.status|default:'Offline' }}
      </li>
    {% endfor %}
  </ul>
{% endblock content %}

{% block script %}
  <script>
    var socket = new WebSocket('ws://' + window.location.host + '/users/');

    socket.onopen = function open() {
      console.log('WebSockets connection created.');
    };

    socket.onmessage = function message(event) {
      var data = JSON.parse(event.data);
      // NOTE: We escape JavaScript to prevent XSS attacks.
      var username = encodeURI(data['username']);
      var user = $('li').filter(function () {
        return $(this).data('username') == username;
      });

      if (data['is_logged_in']) {
        user.html(username + ': Online');
      }
      else {
        user.html(username + ': Offline');
      }
    };

    if (socket.readyState == WebSocket.OPEN) {
      socket.onopen();
    }
  </script>
{% endblock script %}

在主页上，我们扩展用户列表用来显示用户数据，将每个用户名存储为一个数据属性，方便在DOM中搜索到该数据项。向WebSocket添加一个事件监听器，用来处理服务器的消息。当收到消息时，解析JSON数据，定位到该用户的<li>元素，更新该用户状态。

Django不会跟踪用户是否登录，因此我们还要创建一个简单的模型来实现这个功能。创建一个LoggedInUser模型，与用户模型进行一对一的连接。

example_channels/example/models.py：

from django.conf import settings
from django.db import models


class LoggedInUser(models.Model):
    user = models.OneToOneField(
        settings.AUTH_USER_MODEL, related_name='logged_in_user')

当用户登录时应用将创建一个 LoggedInUser 实例，当用户退出时这个实例将被删除。更新我们的数据库：

(env)$ python manage.py makemigrations
(env)$ python manage.py migrate

更新用户列表视图函数，提供检索需要渲染的用户列表：

example_channels/example/views.py

from django.contrib.auth import get_user_model, login, logout
from django.contrib.auth.decorators import login_required
from django.contrib.auth.forms import AuthenticationForm, UserCreationForm
from django.core.urlresolvers import reverse
from django.shortcuts import render, redirect


User = get_user_model()


@login_required(login_url='/log_in/')
def user_list(request):
    """
    NOTE: This is fine for demonstration purposes, but this should be
    refactored before we deploy this app to production.
    Imagine how 100,000 users logging in and out of our app would affect
    the performance of this code!
    """
    users = User.objects.select_related('logged_in_user')
    for user in users:
        user.status = 'Online' if hasattr(user, 'logged_in_user') else 'Offline'
    return render(request, 'example/user_list.html', {'users': users})


def log_in(request):
    form = AuthenticationForm()
    if request.method == 'POST':
        form = AuthenticationForm(data=request.POST)
        if form.is_valid():
            login(request, form.get_user())
            return redirect(reverse('example:user_list'))
        else:
            print(form.errors)
    return render(request, 'example/log_in.html', {'form': form})


@login_required(login_url='/log_in/')
def log_out(request):
    logout(request)
    return redirect(reverse('example:log_in'))


def sign_up(request):
    form = UserCreationForm()
    if request.method == 'POST':
        form = UserCreationForm(data=request.POST)
        if form.is_valid():
            form.save()
            return redirect(reverse('example:log_in'))
        else:
            print(form.errors)
    return render(request, 'example/sign_up.html', {'form': form})

如果该用户有对应的 LoggedInUser 则标记为在线，否则标记为离线。添加了一个@login_required装饰器，用来限制仅仅对注册用户的访问。

添加以下导入包

from django.contrib.auth import get_user_model, login, logout
from django.contrib.auth.decorators import login_required

现在，用户可以登录，注销，这些会触发服务器向客户端发送消息。

但是当用户第一次登录时，我们无法知道哪些用户已经登录。用户只在其他用户的状态更改时才看到更新。这就是LoggedInUser发挥作用的地方，但我们需要一种方法，在用户登录时创建LoggedInUser实例，然后在该用户注销时将其删除。

但我们现在还没有办法知道当用户第一个登录时是哪一位。只用当其他用户登录状态更新是我们才能看到。

Django库包含一个称为Signals的特性，当发生某些操作时，它会广播通知。应用程序可以监听这些通知，然后对它们采取行动。我们可以利用两个有用的内置信号(user_login和user_logout)来处理LoggedInUser行为。

example_channels/example/signals.py****：

from django.contrib.auth import user_logged_in, user_logged_out
from django.dispatch import receiver
from example.models import LoggedInUser


@receiver(user_logged_in)
def on_user_login(sender, **kwargs):
    LoggedInUser.objects.get_or_create(user=kwargs.get('user'))


@receiver(user_logged_out)
def on_user_logout(sender, **kwargs):
    LoggedInUser.objects.filter(user=kwargs.get('user')).delete()

example_channels/example/apps.py

from django.apps import AppConfig


class ExampleConfig(AppConfig):
    name = 'example'

    def ready(self):
        import example.signals

example_channels/example/init.py

default_app_config  =  'example.apps.ExampleConfig'

完整性检查

到此，代码部分已经完成，我们用多个用户账号连接到服务器来测试一下我们的应用。启动 Django 服务，登录系统，访问项目主页。我们应该能够看到所有的用户列表，此时用户的状态都是“离线”。打开新的浏览器匿名窗口，用另一个账号登录，这时各个窗口的用户列表会自动更新到“在线”状态。你可以通过不同的浏览器、设备来测试登录登出。

查看客户端浏览器上的开发人员控制台和终端中的服务器活动，你可以观察到：当用户登录时，WebSocket 连接被创建，当用户注销时，WebSocket 连接被销毁。

总结

本文我们讨论了：

Django Channels
WebSockets
用户身份验证
Django 信号
部分前端开发技术

重要的是 Django Channels 扩展了 Django 框架的传统功能，通过 WebSockets 我们可以将消息从服务器直接发送到客户端。这个功能可以让我们进一步做出很多有意思的东西，比如聊天室、多人在线游戏、能够实时通信的协作应用。一般的应用使用 WebSockets ，在服务器完成任务后向客户端发送状态更新来代替传统的定期轮询服务器，从而得到性能改进。

本文只是简单介绍了 Django Channels 的基本使用，感兴趣的童鞋可以阅读 Django Channels 项目的文档，看看你还可以用它来实现什么有趣的东东。

Django Channels 入门

开始

WebSocket 101

Consumers and Groups

模板

视图

测试

用户身份验证

登录提醒

完整性检查

总结

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

好玩的 Python