跳转至

Tornado整合socketio(一)

最近在做的项目中，需要将手机中的视频流或者音频流发送给服务端，再由服务端转发给浏览器端，起初我使用redis作为中转，将数据发到redis中，再由redis的发布订阅功能，整体架构如下

image-20210519225715299

主要是利用了redis的pub/sub功能，这种方案也没有什么问题，但是整体的性能瓶颈受redis的影响。

最近接触到socketio，发现这种需求可以使用它来实现，但是网上查找了一些资料，在python的使用中，主要还是flask-socketio与原生的应用上，由于目前项目使用Tornado来构建，所以用了几天时间将socketio与Tornado的融合使用。

本教程会分几篇来介绍，主要以下几个章节

  1. socketio介绍与脚手架的搭建
  2. 定义消息处理事件
  3. 命名空间的使用
  4. 消息的发布与响应
  5. room的使用
  6. 前端vue中使用socketio与后端通信

一、socketio 简介

Socket.IO 支持实时、双向和基于事件的通信。它能够在任何平台、浏览器或设备上运行，可靠性和速度同样出色。它兼容websocket，在不支持websocket的设备上，会使用更加低层的长链接协议

Socket.io是一个WebSocket库，包括了客户端的js和服务器端的nodejs，它的目标是构建可以在不同浏览器和移动设备上使用的实时应用。它会自动根据浏览器从WebSocket、AJAX长轮询、Iframe流等等各种方式中选择最佳的方式来实现网络实时应用，非常方便和人性化，而且支持的浏览器最低达IE5.5

socket.io特点

  • 实时分析：将数据推送到客户端，这些客户端会被表示为实时计数器，图表或日志客户。
  • 实时通信和聊天：只需几行代码便可写成一个Socket.IO的”Hello,World”聊天应用。
  • 二进制流传输：从1.0版本开始，Socket.IO支持任何形式的二进制文件传输，例如：图片，视频，音频等。
  • 文档合并：允许多个用户同时编辑一个文档，并且能够看到每个用户做出的修改。

我这个项目主要利用其二进制流的传输。

二、python-socketio 简介

最初socketio的后台使用nodejs，后来又有了java,c++,python等后端的应用。

python的后端库地址 https://github.com/miguelgrinberg/python-socketio

但是注意，不同的版本并不兼容，参考下表

JavaScript Socket.IO version Socket.IO protocol revision Engine.IO protocol revision python-socketio version
0.9.x 1, 2 1, 2 Not supported
1.x and 2.x 3, 4 3 4.x
3.x and 4.x 5 4 5.x

比如python-socketio用的是5.x的，那前端应该使用3.x或者4.x的socket.io 库，不同的版本不兼容。

另外，我演示时使用python版本是3.7.8的，我试过在3.5上的python中安装python-socketio时会安装失败，在安装一个依赖库didict时会失败

我的环境如下

python:3.7.8

python-socketio: 5.3.0

tornado: 6.0.4

windows x64

三、Tornado 的搭建

首先先使用Tornado构建一个简单的基础web应用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
#-*- coding:utf-8 -*-
# author:Yang
# datetime:2021/5/19 23:26
# software: PyCharm

import os
import tornado.httpserver
import tornado.web
import tornado.gen
import tornado.concurrent
import tornado.autoreload
from tornado.platform.asyncio import AsyncIOMainLoop
import asyncio
import traceback


class Index(tornado.web.RequestHandler):
    def get(self):
        self.write("hello world")


class NotFount(tornado.web.RequestHandler):
    def get(self):
        self.write("404 not found")


def make_app(**kwargs):
    settings = dict(
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        static_path=os.path.join(os.path.dirname(__file__), "static"),
        debug=True,
    )
    return tornado.web.Application([
        (r'/', Index),
        (r'.*', NotFount),

    ], **settings, **kwargs)


if __name__ == '__main__':
    try:
        AsyncIOMainLoop().install()
        app = make_app()
        app.listen(8080)
        loop = asyncio.get_event_loop()
        processPoolNum = 2
        try:
            loop.run_forever()
        except KeyboardInterrupt:
            for task in asyncio.Task.all_tasks():
                task.cancel()
            loop.stop()
            loop.run_forever()
        finally:
            loop.close()
    except:
        print(traceback.print_exc())
    finally:
        pass

上面的代码即可以搭建一个基础的Tornado web服务， 之后的代码就在这个基础上做添加。

四、添加python-socketio

python-socketio 分为server端与client端，在server端安装命令为

pip install python-socketio

先初始化socketio.AsyncServer 类的实例

1
2
3
import socketio

sio = socketio.AsyncServer(async_mode='tornado')

以下将python-socketio 简称为socketio

socketio server 有两个版本，一个同步的Server()，一个异步的AsyncServer() ，功能是样的，只是异步的server可以构建在asyncio环境中，由于我的Tornado应用也是使用asyncio，所以这里我使用了异步的server。

之后创建一个路由, 修改make_app函数

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def make_app(**kwargs):
    settings = dict(
        template_path=os.path.join(os.path.dirname(__file__), "templates"),
        static_path=os.path.join(os.path.dirname(__file__), "static"),
        debug=True,
    )
    return tornado.web.Application([
        (r'/', Index),
        (r"/socket.io/", socketio.get_tornado_handler(sio)),
        (r'.*', NotFount),

    ], **settings, **kwargs)

注意这里只能添加/socket.io/ 的路由，这里先这么写，之后在介绍client端面时再说明为什么。

还有一点要注意的，这个/socket.io/ 要定义在.* 路由之前，否则也会命中NotFount。

五、定义事件处理函数

当我们使用websocket时，我们会定义几个常用的方法，如connect 为成功连接上以后调用，disconnect 为断开连接时的操作。

对于服务端，同样我们也可以定义这些函数,我们称为事件(event)， 定义事件有两种方式

  1. 使用sio.event装饰器
1
2
3
@sio.event
def my_event(sid, data):
    print("my_event get an message {} from {}".format(data, sid))
  1. 使用sio.on方法
1
2
3
@sio.on("my event")
def event_special(sid, data):
    print('special event get an message {} from {}'.format(data, sid))

两种方法定义的事件相同，使用第一个种方法，函数名即为事件名，这里就为my_event, 由于函数名不能有特殊的字符有空格，但是第二种方法，可以将事件名定义在装饰器参数中，如这里的my event 事件， 它中间有个空格，所以看需求，需要定义事件的名字中有特殊字符的需要使用on方法。

connect 和disconnect 事件是特殊的事件，在客户端进行连接和断开连接时自动调用

1
2
3
4
5
6
7
@sio.event
def connect(sid, environ, auth):
    print('connect ', sid)

@sio.event
def disconnect(sid):
    print('disconnect ', sid)

注意在自定义事件时，需要两个参数，一个是sid, 这个是客户端标识，一个是data, 这个是客户端发送过来的数据。

六、使用客户端进行连接

客户端的安装与服务端有所不同，使用pip install "python-socketio[client]" 安装同步版本

使用pip install "python-socketio[asyncio_client]" 安装异步版本，我这里使用同步版本的客户端，如果你需要在asyncio中使用则需要安装异步版本

1
2
3
4
5
import socketio

sio = socketio.Client()

sio.connect('http://localhost:8080')

三行代码即可进行连接web服务，并与之建立长连接，这里有一点要注意，我们在server端是创建了一个路由，

1
(r"/socket.io/", socketio.get_tornado_handler(sio))

但是这里连接的时候却不能将/socket.io/ 添加到connect的url中，只能写到根url

查看connect函数定义

1
2
3
def connect(self, url, headers={}, auth=None, transports=None,
                namespaces=None, socketio_path='socket.io', wait=True,
                wait_timeout=1):

这里有一个socketio_path 参数，也正是这个参数，所以在定义server端的时候，要加上那么一条路由，当然这个参数也可以自己定义，这里为了简单就不自定义了，只要知道这里是可以自己定义的。

调用client，观察server端的输出，当有客户端连接到服务端以后，服务端会自动触发connect事件，这里就执行了自定义的函数

1
2
3
@sio.event
def connect(sid, environ, auth):
    print('connect ', sid)

打印输出

connect VUke1-fDf8dYFAyEAAAB, 其中后面的为客户端的sid， 当关闭客户端时，又会触发disconnect事件，输出disconnect VUke1-fDf8dYFAyEAAAB。

七、定义客户端事件

其实对于这种长连接的方式，客户端与服务端的界线已经有些模糊了，服务端也可以向客户端发送数据请求，这里的客户端也就相关于服务端。

我们在客户端定义一个connect事件, 用于连接上服务端以后执行的函数

1
2
3
@sio.event
def connect():
    print("I'm connected!")

这里再次连接服务端，当连接成功以后，就会打印出 I'm connected!

八、客户端发送自定义事件

我们在服务端定义了两个事件 my_event 和my event, 那么客户端如何发送这两个事件呢？

客户端可以使用emit 函数

1
2
sio.connect('http://localhost:8080')
sio.emit("my event", {"data": "hello server"})

这行代码即可向服务端发送my event 事件，再将观察服务端的输出，

触发了my event 函数

1
2
3
@sio.on("my event")
def event_special(sid, data):
    print('special event get an message {} from {}'.format(data, sid))

得到输出：

1
special event get an message {'data': 'hello server'} from kh-Ui9gCLG9b5jD4AAAF

九、服务端向客户端发送事件

客户端可以向服务端发送事件，服务端也可以向客户端发送事件，我们先在客户端定义一个事件

1
2
3
@sio.on("client event")
def client_event(data):
    print("get server message:{}".format(data))

修改服务端my event事件代码

1
2
3
4
@sio.on("my event")
async def event_special(sid, data):
    print('special event get an message {} from {}'.format(data, sid))
    await sio.emit("client event", "hello {}".format(sid))

当服务端接收到my event事件以后，再向客户端发送一个client event事件，注意这里由于服务端使用的是异步的,所以这里要将函数改为async def, emit函数也需要使用await 。

再将调用客户端进行连接，这时客户端会得到如下输出

1
2
I'm connected!
get server message:hello tNp1Yb7OONZn2FwxAAAB

I'm connected! 是触发了connect事件，get server message:hello tNp1Yb7OONZn2FwxAAAB 是触发了client event 事件。

十、自动重连

当我们把服务端关掉以后，此时客户端的脚本并没有退出，当再将启动服务端的时候，客户端可以自动连接上，这个也是该库的方便之处，如果要自己写长连接的话，还要考虑重连问题。

到此，已经掌握了socketio与Tornado的最基本的使用，可以相互发送消息，之后的文章里会介绍更加详细命名空间，与web前端的交互操作。

参考

python-socketio官方文档