如何使用Tornado创建一个简单的Python WebSocket服务器

本文概述

龙卷风和WebSockets
WebSockets在行动
在Nginx后面跑
下一步是什么？

随着实时Web应用程序的普及, WebSockets已成为其实现中的关键技术。必须不断按下重新加载按钮以从服务器接收更新的日子已经一去不复返了。想要提供实时更新的Web应用程序不再需要轮询服务器以查找更改-而是服务器在发生更改时将更改向下推送。强大的Web框架已开始支持WebSocket。例如, Ruby on Rails 5进一步完善了它, 并增加了对动作电缆的支持。
在Python世界中, 存在许多流行的Web框架。 Django之类的框架几乎提供了构建Web应用程序所需的所有内容, 而缺少的任何内容都可以通过Django可用的数千个插件之一来弥补。但是, 由于Python或其大多数Web框架的工作方式, 处理长期存在的连接很快会成为噩梦。线程模型和全局解释器锁通常被认为是Python的致命弱点。
但是所有这些已经开始改变。借助Python 3的某些新功能以及适用于Python的框架(例如Tornado), 处理长期存在的连接不再是一个挑战。 Tornado用Python提供了Web服务器功能, 在处理长期连接方面特别有用。
在本文中, 我们将研究如何使用Tornado在Python中构建简单的WebSocket服务器。该演示应用程序将允许我们上传制表符分隔值(TSV)文件, 对其进行解析并将其内容提供给唯一的URL。
龙卷风和WebSockets Tornado是一个异步网络库, 专门处理事件驱动的网络。由于服务器自然可以同时容纳数以万计的打开连接, 因此服务器可以利用此优势并在单个节点中处理大量WebSocket连接。 WebSocket是一种协议, 可通过单个TCP连接提供全双工通信通道。由于它是开放式套接字, 因此该技术使Web连接成为有状态的, 并有助于与服务器之间的实时数据传输。服务器保持客户端的状态, 使基于WebSockets的实时聊天应用程序或Web游戏的实现变得容易。
WebSocket设计为在Web浏览器和服务器中实现, 并且目前在所有主要的Web浏览器中都受支持。连接一次打开, 并且消息可以在连接关闭之前来回传播多次。
安装龙卷风相当简单。它在PyPI中列出, 可以使用pip或easy_install进行安装：

pip install tornado

Tornado带有自己的WebSockets实现。就本文而言, 这几乎是我们所需要的。
WebSockets在行动使用WebSocket的优点之一是其有状态属性。这改变了我们通常认为客户端-服务器通信的方式。这种情况的一种特殊用例是, 要求服务器执行缓慢的长时间过程并将结果逐渐流回客户端。
在我们的示例应用程序中, 用户将能够通过WebSocket上传文件。在连接的整个生命周期中, 服务器将在内存中保留已解析的文件。根据请求, 服务器可以将文件的一部分发送回前端。此外, 该文件将通过URL提供, 然后可供多个用户查看。如果将另一个文件上传到相同的URL, 则查看该文件的每个人都可以立即看到新文件。
对于前端, 我们将使用AngularJS。该框架和库将使我们能够轻松处理文件上传和分页。但是, 对于与WebSockets相关的所有内容, 我们将使用标准的JavaScript函数。
这个简单的应用程序将分为三个单独的文件：

parser.py：实现了带有请求处理程序的Tornado服务器
templates / index.html：前端HTML模板
static / parser.js：对于我们的前端JavaScript

打开一个WebSocket
在前端, 可以通过实例化WebSocket对象来建立WebSocket连接：

new WebSocket(WEBSOCKET_URL);

这是我们在页面加载时必须要做的事情。实例化WebSocket对象后, 必须附加处理程序以处理三个重要事件：

打开：建立连接时触发
消息：从服务器收到消息时触发
关闭：关闭连接时触发

$scope.init = function() { $scope.ws = new WebSocket('ws://' + location.host + '/parser/ws'); $scope.ws.binaryType = 'arraybuffer'; $scope.ws.onopen = function() { console.log('Connected.') }; $scope.ws.onmessage = function(evt) { $scope.$apply(function () { message = JSON.parse(evt.data); $scope.currentPage = parseInt(message['page_no']); $scope.totalRows = parseInt(message['total_number']); $scope.rows = message['data']; }); }; $scope.ws.onclose = function() { console.log('Connection is closed...'); }; }$scope.init();

由于这些事件处理程序不会自动触发AngularJS的$ scope生命周期, 因此需要将处理程序函数的内容包装在$ apply中。如果你有兴趣, 可以使用AngularJS特定的软件包, 这些软件包使在AngularJS应用程序中集成WebSocket更容易。
值得一提的是, 断开的WebSocket连接不会自动重新建立, 并且需要应用程序在触发close事件处理程序时尝试重新连接。这有点超出本文的范围。
选择要上传的文件
由于我们正在使用AngularJS构建单页应用程序, 因此尝试使用陈旧的方法提交包含文件的表单将不起作用。为了简化操作, 我们将使用Danial Farid的ng-file-upload库。使用它, 我们需要做的就是允许用户上传文件, 是使用特定的AngularJS指令在我们的前端模板中添加一个按钮：

< button class="btn btn-default" type="file" ngf-select="uploadFile($file, $invalidFiles)" accept=".tsv" ngf-max-size="10MB"> Select File< /button>

在许多方面, 该库使我们可以设置可接受的文件扩展名和大小。就像任何< input type =” file” > 元素一样, 单击此按钮将打开标准文件选择器。
上载档案
当你要传输二进制数据时, 可以在数组缓冲区和Blob之间进行选择。如果仅仅是图像数据之类的原始数据, 请选择blob并在服务器中正确处理。数组缓冲区用于固定长度的二进制缓冲区, 并且文本文件(例如TSV)可以字节串的格式传输。此代码段显示了如何以阵列缓冲区格式上载文件。

$scope.uploadFile = function(file, errFiles) { ws = $scope.ws; $scope.f = file; $scope.errFile = errFiles & & errFiles[0]; if (file) { reader = new FileReader(); rawData = http://www.srcmini.com/new ArrayBuffer(); reader.onload = function(evt) { rawData = evt.target.result; ws.send(rawData); }reader.readAsArrayBuffer(file); } }

ng-file-upload指令提供了uploadFile函数。在这里, 你可以使用FileReader将文件转换为数组缓冲区, 然后通过WebSocket发送该文件。
请注意, 通过将大型文件读入数组缓冲区来通过WebSocket发送大文件可能不是上载的最佳方法, 因为它会很快占用大量内存, 从而导致不良的体验。
在服务器上接收文件

文章图片
龙卷风使用4位操作码确定消息类型, 并为二进制数据返回str, 为文本返??回unicode。

if opcode == 0x1: # UTF-8 data self._message_bytes_in += len(data) try: decoded = data.decode("utf-8") except UnicodeDecodeError: self._abort() return self._run_callback(self.handler.on_message, decoded) elif opcode == 0x2: # Binary data self._message_bytes_in += len(data) self._run_callback(self.handler.on_message, data)

在Tornado Web服务器中, 以str类型接收数组缓冲区。
在此示例中, 我们期望的内容类型是TSV, 因此将文件解析并转换为字典。当然, 在实际应用中, 有更明智的方式来处理任意上载。

def make_message(self, page_no=1): page_size = 100 return { "page_no": page_no, "total_number": len(self.rows), "data": self.rows[page_size * (page_no - 1):page_size * page_no] }def on_message(self, message): if isinstance(message, str): self.rows = [csv.reader([line], delimiter="\t").next() for line in (x.strip() for x in message.splitlines()) if line] self.write_message(self.make_message())

请求页面
由于我们的目标是在小页面的大块中显示上载的TSV数据, 因此我们需要一种请求特定页面的方法。为简单起见, 我们将简单地使用相同的WebSocket连接将页码发送到我们的服务器。

$scope.pageChanged = function() { ws = $scope.ws; ws.send($scope.currentPage); }

服务器将以unicode形式收到此消息：

def on_message(self, message): if isinstance(message, unicode): page_no = int(message) self.write_message(self.make_message(page_no))

尝试使用来自Tornado WebSocket服务器的命令进行响应将自动以JSON格式对其进行编码。因此, 只发送包含100行内容的字典是完全可以的。
与他人共享访问权限
为了能够与多个用户共享对同一上传的访问权限, 我们需要能够唯一标识上传。每当用户通过WebSocket连接到服务器时, 都会生成一个随机UUID并将其分配给他们的连接。

def open(self, doc_uuid=None): if doc_uuid is None: self.uuid = str(uuid.uuid4())

uuid.uuid4()会生成一个随机UUID, 而str()会将UUID转换为标准格式的十六进制数字字符串。
如果另一个具有UUID的用户连接到服务器, 则将文件处理程序的相应实例添加到以UUID为键的字典中, 并在关闭连接时将其删除。

@classmethod @tornado.gen.coroutine def add_clients(cls, doc_uuid, client): with (yield lock.acquire()): if doc_uuid in cls.clients: clients_with_uuid = FileHandler.clients[doc_uuid] clients_with_uuid.append(client) else: FileHandler.clients[doc_uuid] = [client]@classmethod @tornado.gen.coroutine def remove_clients(cls, doc_uuid, client): with (yield lock.acquire()): if doc_uuid in cls.clients: clients_with_uuid = FileHandler.clients[doc_uuid] clients_with_uuid.remove(client)if len(clients_with_uuid) == 0: del cls.clients[doc_uuid]

同时添加或删除客户端时, 客户端字典可能会引发KeyError。由于Tornado是异步网络库, 因此它提供了用于同步的锁定机制。使用协程的简单锁适合处理客户字典的这种情况。
如果有任何用户上传文件或在页面之间移动, 则所有具有相同UUID的用户都将查看同一页面。

@classmethod def send_messages(cls, doc_uuid): clients_with_uuid = cls.clients[doc_uuid] message = cls.make_message(doc_uuid)for client in clients_with_uuid: try: client.write_message(message) except: logging.error("Error sending message", exc_info=True)

在Nginx后面跑实现WebSockets非常简单, 但是在生产环境中使用WebSockets时需要考虑一些棘手的事情。 Tornado是一台Web服务器, 因此可以直接获取用户的请求, 但是出于多种原因, 将其部署在Nginx后面可能是一个更好的选择。但是, 要通过Nginx使用WebSocket, 需要花费更多的精力：

http { upstream parser { server 127.0.0.1:8080; }server { location ^~ /parser/ws { proxy_pass http://parser; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection "upgrade"; } } }

【如何使用Tornado创建一个简单的Python WebSocket服务器】这两个proxy_set_header指令使Nginx将必需的标头传递给升级到WebSocket的连接所必需的后端服务器。
下一步是什么？在本文中, 我们实现了一个简单的Python Web应用程序, 该应用程序使用WebSockets维护服务器与每个客户端之间的持久连接。使用像Tornado这样的现代异步网络框架, 在Python中同时保持数以万计的开放连接是完全可行的。
尽管此演示应用程序的某些实现方面可以用其他方法完成, 但我希望它仍有助于在https://www.srcmini.com/tornado框架中演示WebSockets的用法。演示应用程序的源代码可在GitHub上获得。
相关：Python多线程教程：并发和并行