在线聊天——WebSocket 简介与实际实现

Posted 2022-03-03 胖虎是只mao

在 html5 之前，浏览器和服务器的通信都是通过 HTTP 协议进行的，那么引入 WebSocket 有什么用呢？想像一下，有一个网页正在用文字直播某一场足球比赛的比分，在没有 WebSocket 时，客户端也就是浏览器实现比分的动态更新通常通过下面两种方式：AJAX 和长轮询。

这两种技术有什么区别呢？看下面的例子（S 代表服务器，C 代表客户端）：

Ajax 方式：

C：告诉我最新的比分
S：比分还没更新

C：（5 秒后）告诉我最新的比分
S：比分还没更新

C：（5 秒后）告诉我最新的比分
S：比分还没更新
...

C：（5 秒后）告诉我最新的比分
S：2-1
(C 收到比分，显示比分)

C：（5 秒后）告诉我最新的比分
...（n 次后）
S：3-1
(C 收到比分，显示比分)

长轮询：

C：告诉我最新的比分
S：比分还没更新，等更新了告诉你
...

S：（5 分钟后）最新比分：2-1
(C 收到比分，显示比分)

C：告诉我最新的比分
S：比分还没更新，等更新了告诉你
...

S：（3 分钟后）最新比分：3-1
(C 收到比分，显示比分)C：告诉我最新的比分
S：比分还没更新，等更新了告诉你
...

S：（5 分钟后）最新比分：2-1
(C 收到比分，显示比分)

C：告诉我最新的比分
S：比分还没更新，等更新了告诉你
...

S：（3 分钟后）最新比分：3-1
(C 收到比分，显示比分)

Ajax 方式比较简单，就是让浏览器隔一段时间发送一个 Ajax 请求，服务器收到请求后立即返回，但返回的不一定是有效数据。客户端一直发送请求，直到获取到有效比分，然后更新页面。而使用长轮询方式，发送一个请求后，服务器不会立即返回，而是阻塞在那，直到比分有更新才返回。

这两种方式有以下几个共同点：

• 请求都是由客户端发起
• 请求都是采用 HTTP 协议

这也是 HTTP 协议一个不足的地方，永远是客户端去联系服务器，服务器不能主动联系客户端。

不同点：

• Ajax 请求，客户端会一直发请求，直到获得响应。
• 长轮询方式，客户端发送请求后，会阻塞在那，直至获得响应。

WebSocket 协议的出现弥补了这个不足。有了 Websocket ，服务器可以主动联系客户端，给客户端发送消息。前提是客户端与服务器要先建立连接，因为 WebSocket 是基于 TCP 协议的，它的握手过程使用 HTTP 协议。

使用 WebSocket 的通信方式：

C：我需要建立 Websocket 连接。需要的服务: ... Websocket 版本: ...
S：ok，已经确认并建立连接
C：比分有更新告诉我
S：ok

S：最新比分：2-1
(C 收到比分，显示比分)
S：最新比分：3-1
(C 收到比分，显示比分)

使用 WebSocket 只需要建立一次连接，客户端不需要发送请求就可以接收到最新的比分。

二、实际实现

在实现后端代码之前先安装几个第三方库：

sudo pip3 install flask-sockets gunicorn redis

后端对 WebSocket 的支持使用 Flask-Sockets 库实现，该库支持为 WebSocket 注册一个蓝图。

创建蓝图

现在我们在 /handlers 目录下面新建 ws.py 文件，并创建同名蓝图：

from flask import Blueprint

ws = Blueprint('ws', __name__, url_prefix='/ws')

在 /handlers/init.py 文件中引入，并创建一个列表：

from .front import front
from .course import course
from .live import live
from .admin import admin
from .ws import ws


bp_list = [front, course, live, admin, ws]

然后在 simpledu/app.py 文件中注册：

from simpledu.handlers import bp_list,ws

def register_blueprints(app):
    """注册蓝图
    """

    for bp in bp_list:
        app.register_blueprint(bp)

创建 Sockets 实例并注册蓝图

接下来，在 /app.py 文件中注册套接字对象，也就是创建 Sockets 类的实例：

from flask_sockets import Sockets        # 新增代码


def register_blueprints(app):
    """注册蓝图
    """

    for bp in bp_list:
        app.register_blueprint(bp)

    sockets = Sockets(app)                    # 新增代码
    sockets.register_blueprint(ws)    # 新增代码

为了避免 Flask 主服务阻塞以提升聊天服务器的性能，服务器端的实现基于 gevent ，它是一个支持异步 I/O 的第三方库。

引入 gevent 后，我们需要考虑一个问题，如何同步消息？我们这里使用 Redis 的 Pub/Sub 发布订阅系统来做一个简单的消息队列。订阅者订阅频道，发布者发布消息到频道，频道负责收发消息队列。

终端执行如下命令，启动 Redis 服务：

sudo service redis-server start

后端首先实现了一个 Chatroom 类，主要用于管理客户端与频道的连接以及向所有客户端发送消息，然后用 gevent 异步启动聊天室。接着使用 Flask-Sockets 提供的方式实现一个接口（视图函数），用于将客户端加入到 Redis 的聊天室频道，然后接收客户端发来的消息并把消息发布到 Redis 的聊天室频道。

将以下代码写入 /handlers/ws.py 文件：

import redis
import gevent
from flask import Blueprint


ws = Blueprint('ws', __name__, url_prefix='/ws')

# 此对象是一个 Redis 客户端，它既可以发布消息到频道，也可以订阅频道
redis = redis.from_url('redis://127.0.0.1:6379')


class Chatroom:

    def __init__(self):
        self.clients = []               # 用户列表
        self.pubsub = redis.pubsub()    # 初始化发布订阅系统
        self.pubsub.subscribe('chat')   # 订阅 chat 频道

    def register(self, client):
        """注册用户，把用户添加到用户列表里

        :para client: geventwebsocket.websocket.WebSocket 类的实例
        """
        self.clients.append(client)

    def send(self, client, data):
        """发送数据给浏览器

        :para client: geventwebsocket.websocket.WebSocket 类的实例
        :para data: 要发送的消息，二进制字典字符串 b'"user": xxx'
        """
        # 调用 client 的 send 方法给浏览器发送消息
        # 如果出现异常，表示连接已关闭，将客户端移除
        try:
            client.send(data.decode())
        except:
            self.clients.remove(client)

    def run(self):
        # 下面一行代码中发布订阅系统对象 self.pubsub 的 listen 方法处于阻塞状态
        # 这里之所以使用 Redis 的发布订阅系统
        # 就是因为它能阻塞监听，并且消息队列功能保证消息按照先进先出的次序移动
        # 当用户在浏览器页面输入信息并点击「发言」按钮后
        # 浏览器调用 inbox 对象向服务器发送数据
        # 服务器的视图函数调用 geventwebsocket.websocket.WebSocket 类的
        # 实例的 receive 方法接收数据
        # 然后 redis 客户端调用 publish 方法向 chat 频道发送此数据
        # 此处收到消息，进入 for 循环
        for message in self.pubsub.listen():
            if message['type'] == 'message':
                # 要发送给浏览器的数据，二进制字典字符串 b'"user": xxx'
                data = message.get('data')
                # 向用户列表中的全部用户发送数据
                # 也就是 geventwebsocket.websocket.WebSocket 实例向浏览器发送数据
                for client in self.clients:
                    # 发送消息需要一小段时间，这里使用 gevent 异步发送
                    gevent.spawn(self.send, client, data)

    def start(self):
        # 因为 self.run 方法会阻塞运行，这里使用异步执行
        gevent.spawn(self.run)


chat = Chatroom()   # 初始化聊天室对象
chat.start()        # 异步启动聊天室

有同学可能会感到奇怪，Chatroom 类中已经定义了 run 函数，为什么又要定义一个 start 函数？在注释里面已经写了，它是用来异步执行 run 函数的。在真实的公司项目中，像这种 WebSocket 服务通常是独立出来作为一个单独的服务进行管理，或者直接使用专门做这些服务的公司的产品。在我们这个小项目中， WebSocket 服务是嵌入到 Web 服务内部的，如果不异步执行 run 函数，就会导致 Web 服务被阻塞，页面请求得不到响应。

接着上面的代码，继续在 /handlers/ws.py 中添加处理函数。我们的聊天功能实际上是为每个客户端建立了一个 WebSocket 连接，客户端其实就是浏览器。将如下代码写入其中：

@ws.route('/send')
def inbox(ws):
    # 注册用户，也就是把 ws 放到聊天室的 clients 列表里
    chat.register(ws)
    # 使用 Flask-Sockets，ws 连接对象会被自动注入到路由处理函数，该处理函数用来处理前端发过来的消息。
    # 注意下面的 while 循环，里面的 receive() 函数实际是在阻塞运行的，直到前端发送消息过来。
    # 消息会被放入到 chat 频道，也就是我们的消息队列中，这样一直循环，直到 websocket 连接关闭。
    while not ws.closed:
        message = ws.receive()
        if message:
            # 发送消息到 chat 频道
            redis.publish('chat', message)

Flask-Sockets 本身代码极为简单，创建了一个套接字类，将其实例作为与应用对象同等效力的对象，并将应用对象的 wsgi_app 属性改成 SocketMiddleware 类的实例。

其余的工作都是由依赖库完成的。创建遵循 WebSocket 协议的、充当服务器的对象 ws 是由 GeventWebSocket 库来完成的；处理前端发来的即时消息，是由 Redis 服务器的 PUB/SUB 系统的消息队列来做的；处理程序阻塞，实现异步执行代码的工作是由 Gevent 来完成的；最后多进程异步启动应用程序是由 Gunicorn 来完成的。