Python利用Socket.IO实现消息实时推送

即时通讯简介

即时通讯（Instant Messaging）是一种基于互联网的即时交流消息的业务

• 类型：

• 在线push：适用于web页面和app，自己构建的IM服务器(socket.io框架，大佬可以自己封装socket)

• 离线push：适用于app，成本高，大厂可以自己用服务器做业务，小厂可以用第三方IM服务商：网易云信，融云，环信

• 传统的推送实现

  • 轮询：客户端定时发出HTTP请求，查询服务器有没有新消息，效率低下，消耗资源
  • Comet：基于长连接，长轮询，等等服务器推送数据。消耗资源

WebSocket协议

• 简介：在单个TCP连接上进行全双工通信的协议，使用ws或者wss统一的资源标识符，例如：ws://example.com/wsapi
  WebSocket默认使用80端口，在TLS上默认使用443端口。
• 优点：
  较小的控制开销，数据包大量减少；更强的实时性，服务器可以随时主动给客户端发送数据，
  保持连接状态，更好的二进制支持
  支持扩展，没有同源限制，可以发送文本，也可发送二进制数据…
• 协议特点：
  WebSocket是独立的、建立在TCP之上的协议，只需一次握手
• 一个栗子：
  客户端发送请求：

GET / HTTP/1.1
Upgrade: websocket
Connection: Upgrade
Host: example.com
Origin: http://example.com
Sec-WebSocket-Key: sN9cRrP/n9NdMgdcy2VJFQ==
Sec-WebSocket-Version: 13

服务器响应：

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: fFBooB7FAkLlXgRSz0BT3v4hq5s=
Sec-WebSocket-Location: ws://example.com/

• Connection必须设置Upgrade，表示客户端希望连接升级
• Upgrade字段必须设置Websocket，表示希望升级到Websocket协议。
• Origin字段是可选的，通常用来表示在浏览器中发起此Websocket连接所在的页面，类似于Referer。但是，与Referer不同的是，Origin只包含了协议和主机名称。
• Sec-WebSocket-Key是随机的字符串，服务器端会用这些数据来构造出一个SHA-1的信息摘要。把“Sec-WebSocket-Key”加上一个特殊字符串“258EAFA5-E914-47DA-95CA-C5AB0DC85B11”，然后计算SHA-1摘要，之后进行BASE-64编码，将结果做为“Sec-WebSocket-Accept”头的值，返回给客户端。如此操作，可以尽量避免普通HTTP请求被误认为Websocket协议。
• 其他一些定义在HTTP协议中的字段，如Cookie等，也可以在Websocket中使用。

Socket.IO框架

• 简介：Socket.IO 本是一个面向实时 web 应用的 JavaScript 库，现在已成为拥有众多语言支持的Web即时通讯应用的框架。
  Socket.IO 不等价于 WebSocket，WebSocket只是Socket.IO实现即时通讯的其中一种技术依赖

• 优点：Socket.IO 会自动选择合适双向通信协议，仅仅需要程序员对套接字的概念有所了解。

• 缺点：Socket.io要求客户端与服务器端均须使用该框架。

• 使用方法

• 1.创建服务器

# 安装
pip install python-socketio# 使用协程的方式运行socketio服务器
import evenlet
eventlet.monkey_patch()import socketio
import eventlet.wsgi# 打包称WSGI应用，使用WSGI服务器托管运行
sio = socketio.Server(async_mode='eventlet')  # 指明启动模式
app = socketio.Middleware(sio)
eventlet.wsgi.server(eventlet.listen(('', 8000)), app)

• 2.事件处理
  简介：不同于HTTP服务的编写方式，SocketIO服务编写不再以请求Request和响应Response来处理，而是对收发的数据以消息（message）来对待，收发的不同类别的消息数据又以事件（event）来区分。

"""
定义事件处理方法:connect 为特殊事件，当客户端连接后自动执行disconnect 为特殊事件，当客户端断开连接后自动执行connect、disconnect与自定义事件处理方法的函数传入参数不同
"""@sio.on('connect')
def on_connect(sid, environ):"""与客户端建立好连接后被执行:param sid: string sid是socketio为当前连接客户端生成的识别id:param environ: dict 在连接握手时客户端发送的握手数据(HTTP报文解析之后的字典)"""pass@sio.on('disconnect')
def on_disconnect(sid):"""与客户端断开连接后被执行:param sid: string sid是断开连接的客户端id"""pass# 以字符串的形式表示一个自定义事件，事件的定义由前后端约定
@sio.on('my custom event')  
def my_custom_event(sid, data):"""自定义事件消息的处理方法:param sid: string sid是发送此事件消息的客户端id:param data: data是客户端发送的消息数据"""pass"""
发送事件消息
"""# 群发
sio.emit('my event', {'data': 'foobar'})# 指定用户发送
sio.emit('my event', {'data': 'foobar'}, room=user_sid)# 给一组用户发送，提供room参数给用户分组
# 当客户端连接后，socketio会自动将客户端添加到以此客户端sid为名的room中
@sio.on('chat')
def begin_chat(sid):sio.enter_room(sid, 'chat_users')# 将客户端从一个room中移除
@sio.on('exit_chat')
def exit_chat(sid):sio.leave_room(sid, 'chat_users')# 查询sid客户端所在的所有房间
sio.rooms(sid)# 给一组用户发送消息的示例
@sio.on('my message')
def message(sid, data):sio.emit('my reply', data, room='chat_users')# 发消息时跳过指定客户端
@sio.on('my message')
def message(sid, data):sio.emit('my reply', data, room='chat_users', skip_sid=sid)# 对于'message'事件，可以使用send方法
sio.send({'data': 'foobar'})
sio.send({'data': 'foobar'}, room=user_sid)

• 3.python客户端

import socketiosio = socketio.Client()@sio.on('connect')
def on_connect():pass@sio.on('event')
def on_event(data):passsio.connect('http://10.211.55.7:8000')
sio.wait()

消息推送案例

• 流程：
  1.A关注B，写入数据库。
  2.Web服务想消息队列中写入对B的通知消息
  3.SocketIO服务从消息队列中取出推送的消息
  4.B上线收到SocketIO推送的关注通知消息

• 中间件的选择

"""
使用Redis或者RabbitMQ作为消息中间件
"""
mg = socketio.RedisManager('redis://0.0.0.0:8570')
sio = socketio.Server(client_manager=mg)
# 或者
pip install kombumg = socketio.KombuManager('amqp://')
sio = socketio.Server(client_manager=mg)

• 所有代码

"""
将IM服务单独设置为一个包
编写socketIO服务器启动代码project/im/main.py
"""
import eventlet
eventlet.monkey_patch()import eventlet.wsgi
import sys
import osBASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, os.path.join(BASE_DIR, 'common'))if len(sys.argv) < 2:print('Usage: python main.py [port]')exit(1)from server import app
import chat
import notifySERVER_ADDRESS = ('', port)
sock = eventlet.listen(SERVER_ADDRESS)
eventlet.wsgi.server(sock, app)"""
在project/im/server.py文件中补充消息队列rabbitmq的配置信息和jwt使用的秘钥
"""
import socketioRABBITMQ = 'amqp://python:rabbitmqpwd@localhost:5672/project_demo'
JWT_SECRET = 'TPmi4aLWRbyVq8zu9v82dWYW17/z+UvRnYTt4P6fAXA'mgr = socketio.KombuManager(RABBITMQ)sio = socketio.Server(async_mode='eventlet', client_manager=mgr)
app = socketio.Middleware(sio)"""
在im目录中新建notify.py
"""from server import sio, JWT_SECRET
from werkzeug.wrappers import Request
from utils.jwt_util import verify_jwtdef check_jwt_token(environ):"""检验jwt token:param environ::return:"""request = Request(environ)token = request.args.get('token')if token:payload = verify_jwt(token, JWT_SECRET)if payload:user_id = payload.get('user_id')return user_idreturn None@sio.on('connect')
def on_connect(sid, environ):"""与客户端建立连接后执行"""# 检验连接客户端的jwt tokenuser_id = check_jwt_token(environ)print('user_id={}'.format(user_id))# 若检验出user_id，将此客户端添加到user_id的room中if user_id:sio.enter_room(sid, str(user_id))@sio.on('disconnect')
def on_disconnect(sid):"""与客户端断开连接时执行"""# 客户端离线时将客户端从所有房间中移除rooms = sio.rooms(sid)for room in rooms:sio.leave_room(sid, room)"""
在project/app/src/user/following.py 添加用户关注的业务接口
"""
class FollowingListResource(Resource):"""关注用户"""method_decorators = {'post': [login_required],'get': [login_required],}class FollowingListResource(Resource):"""关注用户"""method_decorators = {'post': [login_required],'get': [login_required],}def get(self):"""获取粉丝列表"""passdef post(self):"""关注用户"""json_parser = RequestParser()json_parser.add_argument('target', type=parser.user_id, required=True, location='json')args = json_parser.parse_args()target = args.targetif target == g.user_id:return {'message': 'User cannot follow self.'}, 400ret = 1# 记录到数据库try:follow = Relation(user_id=g.user_id, target_user_id=target, relation=Relation.RELATION.FOLLOW)db.session.add(follow)db.session.commit()except IntegrityError:db.session.rollback()ret = Relation.query.filter(Relation.user_id == g.user_id,Relation.target_user_id == target,Relation.relation != Relation.RELATION.FOLLOW)\.update({'relation': Relation.RELATION.FOLLOW})db.session.commit()if ret > 0:timestamp = time.time()cache_user.UserFollowingCache(g.user_id).update(target, timestamp)cache_user.UserFollowersCache(target).update(g.user_id, timestamp)cache_statistic.UserFollowingsCountStorage.incr(g.user_id)cache_statistic.UserFollowersCountStorage.incr(target)cache_user.UserRelationshipCache(g.user_id).clear()# 发送关注通知_user = cache_user.UserProfileCache(g.user_id).get()_data = {'user_id': g.user_id,'user_name': _user['name'],'user_photo': _user['photo'],'timestamp': int(time.time())}# 通过socketio提供的kombu管理对象 向rabbitmq中写入数据，记录需要由socketio服务器向客户端推送消息的任务current_app.sio_mgr.emit('following notify', data=_data, room=str(target))return {'target': target}, 201