实时通信库之Socket.IO
Socket.IO
概述
Socket.IO是一个基于WebSocket的实时通信库，它提供了简单易用的API，使得在Web应用程序中添加实时通信功能变得非常容易。
Socket.IO支持多种传输方式，包括WebSocket、Ajax轮询、Ajax长轮询、JSONP等，可以根据环境和浏览器能力自动选择最佳的传输方式。
官网： https://socket.io/
特点
实时性：Socket.IO使用WebSocket协议进行通信，可以做到实时通信，无需刷新页面即可接收新数据
多平台支持：Socket.IO可以在Web、iOS、Android等平台上使用，可以实现跨平台的实时通信
自适应传输：Socket.IO支持多种传输方式，可以根据环境和浏览器能力自动选择最佳的传输方式，从而获得更好的性能和稳定性
支持房间功能：Socket.IO支持将客户端分组到不同的房间中，可以实现向指定房间内的客户端发送消息的功能
兼容性：Socket.IO支持所有主流浏览器和操作系统，包括Internet Explorer、Chrome、Firefox、Safari等
可靠性：Socket.IO具有自动重连和心跳检测等机制，可以保证连接的稳定性和可靠性
易用性：Socket.IO提供了简单易用的API，让开发者可以很方便地在Web应用程序中添加实时通信功能
总之，Socket.IO是一个强大而灵活的实时通信框架，可以帮助开发者快速搭建实时应用程序，并轻松处理复杂的实时通信场景。
Python-Socket.IO的基本使用
在Python中使用Socket.IO，可以借助python-socketio库，它是Socket.IO的客户端和服务器的Python实现。
文档：https://python-socketio.readthedocs.io
安装库
# 用于处理 WebSocket 通信的 Socket.IO 客户端和服务器端库
pip install python-socketio
# 一个快速的并发网络库，用于构建高性能的网络应用程序
pip install eventlet
运行方式
1.WSGI服务器运行
使用多进程多线程模式的WSGI服务器运行，如uWSGI、gunicorn
import socketio
# create a Socket.IO servers
sio = socketio.Server()
# 打包成WSGI应用，可以使用WSGI服务器托管运行
app = socketio.WSGIApp(sio) # Flask Django
2.作为框架中的一部分
作为框架应用中的一部分，例如：Flask、Django 应用。使用uWSGI、或gunicorn服务器运行
from wsgi import app
import socketio
sio = socketio.Server()
app = socketio.WSGIApp(sio, app)
3.使用协程的方式运行 ，这里推荐方式
因为服务器与客户端进行即时通信时，会尽可能的使用长连接，所以若服务器采用多进程或多线程方式运行，受限于服务器能创建的进程或线程数，能够支持的并发连接客户端不会很高，也就是服务器性能有限。采用协程方式运行服务器，可以提升即时通信服务器的性能。
import eventlet
eventlet.monkey_patch()
import socketio
import eventlet.wsgi
# 指明在evenlet模式下
sio = socketio.Server(async_mode='eventlet')
app = socketio.Middleware(sio)
eventlet.wsgi.server(eventlet.listen(('', 8000)), app)
服务器端实现
导入socketio、eventlet模块，然后创建一个SocketIO实例sio
定义三个事件处理函数，分别用于处理连接、断开连接和消息事件。在消息事件处理函数中，向所有客户端广播接收到的消息
最后，通过eventlet启动一个WSGI服务器，并将其绑定到8000端口上
import eventlet
import socketio
# 创建SocketIO实例
# cors_allowed_origins:跨域配置
# async_mode：指明在evenlet模式下
sio = socketio.Server(cors_allowed_origins="*", async_mode='eventlet')
# 定义事件处理函数
@sio.on('connect')
def on_connect(sid, environ):
print("客户端进行连接：", sid, environ)
"""
与客户端建立好连接后被执行
sid: string sid是socketio为当前连接客户端生成的识别id
environ: dict 在连接握手时客户端发送的握手数据(HTTP报文解析之后的字典)
"""
@sio.on('disconnect')
def on_disconnect(sid):
print("disconnect", sid)
"""
与客户端断开连接后被执行
sid: string sid是断开连接的客户端id
"""
# 自定义事件，事件的定义由前后端约定
@sio.on('my_event')
def my_event(sid, data):
print("my_event", sid, data)
"""
自定义事件消息的处理方法
sid: string sid是发送此事件消息的客户端id
data: data是客户端发送的消息数据
"""
sio.emit('message', data) # 向所有客户端广播消息
# 启动
app = socketio.Middleware(sio)
# 通过eventlet启动了一个WSGI服务器，并将其绑定到8000端口上
eventlet.wsgi.server(eventlet.listen(('localhost', 8000)), app)
客户端实现
引入Socket.IO库，并通过io.connect()方法连接到服务器。
在连接成功后，输出连接成功的socket id。
监听message事件，接收服务器发送过来的消息。
在发送消息时，调用了emit()方法，并将消息内容作为参数传递
<!DOCTYPE html>
<html>
<head>
<title>Socket.IO Test</title>
<script src="https://cdn.socket.io/4.0.1/socket.io.min.js"></script>
<script>
var socket = io.connect('http://localhost:8000');
socket.on('connect', function() {
console.log('Connected:', socket.id);
});
socket.on('disconnect', function() {
console.log('Disconnected');
});
socket.on('message', function(data) {
console.log('Received message:', data);
});
function sendMessage() {
var message = document.getElementById('message').value;
socket.emit('my_event', message);
}
</script>
</head>
<body>
<input type="text" id="message">
<button onclick="sendMessage()">Send</button>
</body>
</html>
运行程序
>python server.py
(12212) wsgi starting up on http://127.0.0.1:8000
(12212) accepted ('127.0.0.1', 10356)
测试
在浏览器中打开客户端页面，点击“Send”按钮发送消息，可以在服务器端看到收到的消息，并且该消息会自动被广播到所有客户端。
Firecamp插件
Firecamp是一款强大的API开发和测试工具，它支持WebSocket和Socket.IO协议，并提供了很多方便的功能来快速测试和调试WebSocket应用程
可以在Chrome浏览器中安装Firecamp扩展程序，地址：https://chrome.google.com/webstore/detail/firecamp-a-multi-protocol/eajaahbjpnhghjcdaclbkeamlkepinbl?hl=zh-CN
事件处理方法
在使用Socket.IO时，可以通过事件处理函数来处理客户端和服务端之间的各种事件。
编写事件处理方法，可以接收指定的事件消息数据，并在处理方法中对消息数据进行处理。
以下是一些Socket.IO常见事件处理函数
connect事件
当客户端与服务端成功建立连接时触发。
@socketio.on('connect')
def connect():
print('Connected:', request.sid)
disconnect事件
当客户端与服务端断开连接时触发。
@socketio.on('disconnect')
def disconnect():
print('Disconnected:', request.sid)
自定义事件
除了预定义的事件之外，还可以定义自己的事件，并在客户端和服务端之间进行通信。
在服务端中，可以通过装饰器@socketio.on()来定义事件处理函数，并在其中调用emit()方法向客户端发送消息。
@socketio.on('my_event')
def handle_my_custom_event(data):
emit('my_response', {'data': data['message']})
在客户端中，通过socket.emit()方法触发自定义事件，并在回调函数中处理服务端返回的数据。
socket.emit('my_event', { message: 'hello world!' });
socket.on('my_response', function(data) {
console.log('Received data:', data);
});
broadcast事件
如果想将消息广播到所有已连接的客户端，可以使用broadcast参数。在服务端中，可以将broadcast=True作为参数传递给emit()方法，这样消息会被广播至所有已连接的客户端。
@socketio.on('message')
def handle_message(message):
emit('message', message, broadcast=True)
room事件
如果想将消息发送到指定房间内的所有客户端，可以使用join_room()方法将客户端加入指定的房间，并在emit()方法中增加room参数指定目标房间。
@socketio.on('join')
def on_join(data):
username = data['username']
room = data['room']
join_room(room)
send(username + ' has joined the room.', room=room)
@socketio.on('message')
def handle_message(message):
room = message.get('room')
if room:
emit('message', message, room=room)
else:
emit('message', message, broadcast=True)
客户端中，使用socket.emit()方法将用户加入指定房间，并在回调函数中接收服务端返回的消息。
socket.emit('join', { username: 'Alice', room: 'chatroom' });
socket.on('message', function(data) {
console.log('Received data:', data);
});
发送事件消息
在Socket.IO中，可以通过emit()方法向客户端或服务端发送消息。emit()方法可以接受多个参数，其中第一个参数是消息名称，后面的参数则为具体的消息内容。
发送消息到所有客户端
定义一个message事件处理函数，当客户端发送message事件时，服务端会将该消息广播到所有连接到该应用的客户端。
@socketio.on('message')
def handle_message(message):
emit('message', message, broadcast=True)
使用socket.emit()方法向服务端发送message事件，并传递一条消息。在回调函数中，使用console.log()方法打印服务端返回的消息。
socket.emit('message', 'Hello world!');
socket.on('message', function(data) {
console.log('Received data:', data);
});
发送消息到指定客户端
定义一个private_message事件处理函数，当客户端发送private_message事件时，服务端会将该消息发送给指定房间内的客户端。
@socketio.on('private_message')
def handle_private_message(data):
recipient= data['recipient']
message = data['message']
emit('message', message, room=recipient)
使用socket.emit()方法向服务端发送private_message事件，并传递一个包含收件人和消息内容的对象。在回调函数中，使用console.log()方法打印服务端返回的消息。
socket.emit('private_message', { recipient: 'user123', message: 'Hi there!' });
socket.on('message', function(data) {
console.log('Received data:', data);
});
发送消息到一组客户端
在Socket.IO中，可以使用room参数将一条消息发送到指定的房间。通过将客户端加入指定的房间，可以实现向该房间内的所有客户端发送消息的功能。
room相关方法
1.将连接的客户端添加到一个room
sio.enter_room(sid, room_name)
2.将客户端从一个room中移除
sio.leave_room(sid, room_name)
3.查询sid客户端所在的所有房间
sio.rooms(sid)
4.使用send发送message事件消息
sio.send({'data': 'abc'})
sio.send({'data': 'abc'}, room=user_sid)
使用示例
定义三个事件处理函数。
当用户加入房间时，服务端会将该用户加入指定的房间，并向该房间内的所有客户端发送一条加入房间的通知消息
当用户离开房间时，服务端会将该用户从房间中移除，并向该房间内的所有客户端发送一条离开房间的通知消息
当用户发送消息时，服务端会将该消息发送给指定房间内的所有客户端
@socketio.on('join_room')
def on_join(data):
room = data['room']
join_room(room)
emit('message', f'User {data["username"]} has joined the room {room}.', room=room)
@socketio.on('leave_room')
def on_leave(data):
room = data['room']
leave_room(room)
emit('message', f'User {data["username"]} has left the room {room}.', room=room)
@socketio.on('send_message')
def send_message(data):
room = data['room']
message = data['message']
emit('message', f'{data["username"]}: {message}', room=room)
# 在群组发消息时跳过指定客户端
# emit('message', data, room=room, skip_sid=sid)
在客户端中，使用socket.emit()方法向服务端发送join_room、send_message和leave_room事件，并传递与事件相关的数据。在回调函数中，使用console.log()方法打印服务端返回的消息。
// 加入房间
socket.emit('join_room', { username: 'Alice', room: 'chatroom' });
// 发送消息
socket.emit('send_message', { username: 'Alice', room: 'chatroom', message: 'Hello world!' });
socket.on('message', function(data) {
console.log('Received data:', data);
});
// 离开房间
socket.emit('leave_room', { username: 'Alice', room: 'chatroom' });
综合简单使用
创建Socket IO实例
创建server.py文件，创建Socket IO实例
import socketio
# 创建协程服务器 运行socketio
# cors_allowed_origins:跨域配置
# async_mode：指明在evenlet模式下
sio = socketio.Server(cors_allowed_origins="*", async_mode='eventlet')
app = socketio.Middleware(sio)
消息事件处理
from werkzeug.wrappers import Request
from server import sio
@sio.on('connect')
def on_connect(sid, environ):
print("客户端建立连接: ", sid, environ)
# 加入房间
sio.enter_room(sid, "roomName")
# 响应消息
msg_data = {"msg": "新人加入房间", "code": 200}
# 向房间发送消息
sio.emit('message', msg_data, room='roomName')
# sio.send(msg_data, room=room)
# socketio会将连接请求中携带的http数据都解析之后存到environ中
request = Request(environ)
# 借助werkzeug解读字典，获取数据,如获取token
token = request.args.get('token')
@sio.on('disconnect')
def on_disconnect(sid):
print("客户端断开连接: ", sid)
# 查询sid所在的房间
rooms = sio.rooms(sid)
print(rooms)
# 将当前客户端 清理其房间
for room in rooms:
sio.leave_room(sid, room)
@sio.on('message')
def on_message(sid, data):
# 查询sid所在的房间
rooms = sio.rooms(sid)
print(rooms)
# 客户端发送的内容
msg = data.get('msg')
# 回复发送者消息
resp_msg = {"msg": msg, "code": 200}
# 向房间内发送消息
# sio.emit('message', resp_msg, room="roomName")
sio.send(resp_msg, room="roomName")
封装启动入口
创建main.py文件，作为SocketIO的启动入口
import eventlet
eventlet.monkey_patch() # 协程库的函数替换
import eventlet.wsgi
from server import app
# 必须导入,加载注册事件监听
SERVER_ADDRESS = ('127.0.0.1', 8000)
listen_sock = eventlet.listen(SERVER_ADDRESS)
eventlet.wsgi.server(listen_sock, app)
结合消息中间件
结合消息中间件，可以使得Socket.IO应用程序更加健壮和可扩展。通过将Socket.IO与消息中间件（如RabbitMQ、Kafka等）集成，可以实现以下功能：
消息持久化
当Socket.IO服务宕机或重启时，未处理的消息不会丢失，因为它们已经被存储在消息中间件中
消息队列
将来自多个客户端的请求放入消息队列中，可以实现负载均衡和流量控制等功能，避免了Socket.IO服务因过多请求而崩溃
分布式处理
通过将Socket.IO服务分布到多个节点上，并使用消息中间件进行数据同步，可以实现水平扩展和高可用性
使用Redis
mgr = socketio.RedisManager('redis://')
sio = socketio.Server(client_manager=mgr)
使用RabbitMQ
pip install kombu
import socketio
# 为socketio服务器添加manager对象，帮助从rabbitmq取推送任务
RABBITMQ = 'amqp://python:pwd@localhost:5672/username'
mgr = socketio.KombuManager(RABBITMQ)
sio = socketio.Server(async_mode='eventlet', client_manager=mgr)
app = socketio.Middleware(sio)