多进程服务器转发客户端消息,python-socketio 多进程部署
实现原理
当使用多进程的时候。多个socketio服务器通过消息队列来沟通之间的客户端sid。若发现该sid在自己的连接中。就由该进程处理发送给其下面连接的客户端
详细的可以看这里
socket.io要实现多进程以及广播，房间等功能，势必需要接入一个redis之类的消息队列，进而socket.io的emit会调用对应队列管理器pubsub_manager的emit方法，比如用redis做消息队列则最终调用 redis_manager中的_publish()
方法通过redis的订阅发布功能将消息推送到flask_socketio频道。另一方面，每个进程在初始化时都订阅了 flask_socketio频道，而且都有一个协程(或线程)在监听频道中是否有消息，一旦有消息，就会调用pubsub_manager._handle_emit()方法对本机对应的socket发送对应的消息，最终是通过socket.io服务器的_emit_internal()方法实现对本机中room为sid的所有socket发送消息的，如果room为None，则就是广播，即对所有连接到本机的所有客户端推送消息。
作者：ssjhust
链接：https://juejin.cn/post/6844903632819716110
来源：掘金
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
服务端实现
设定client_manger即可
mgr = socketio.AsyncRedisManager('redis://127.0.0.1:6379/2')
sio = socketio.AsyncServer(async_mode='asgi',client_manager=mgr)
#!/usr/bin/env python
import asyncio
import uvicorn
import socketio
# 添加一行clientManger
mgr = socketio.AsyncRedisManager('redis://127.0.0.1:6379/2')
sio = socketio.AsyncServer(async_mode='asgi',client_manager=mgr)
app = socketio.ASGIApp(sio, static_files={
'/': 'app.html',
})
@sio.on('my_event')
async def test_message(sid, message):
await sio.emit('my_response', {'data': message['data']}, room=sid)
@sio.on('connect')
async def test_connect(sid, environ):
print(sid)
await sio.emit('my_response', {'data': 'Connected', 'count': sid}, room=sid)
@sio.on('disconnect')
def test_disconnect(sid):
print('Client disconnected')
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=5000)
实验效果
部署两个app 分别监听5000和50001 表示负载均衡后面的两个服务器
然后用a客户端连接5000。用b客户端向a发消息。如果a接收成功证明服务器之间进行了通讯转发
app1.py
#!/usr/bin/env python
import asyncio
import uvicorn
import socketio
# 添加一行clientManger
mgr = socketio.AsyncRedisManager('redis://127.0.0.1:6379/2')
sio = socketio.AsyncServer(async_mode='asgi',client_manager=mgr)
app = socketio.ASGIApp(sio, static_files={
'/': 'app.html',
})
background_task_started = False
@sio.on('my_event')
async def test_message(sid, message):
print(message)
@sio.on('disconnect request')
async def disconnect_request(sid):
await sio.disconnect(sid)
@sio.on('connect')
async def test_connect(sid, environ):
print(sid)
await sio.emit('my_response', {'data': 'Connected', 'count': sid}, room=sid)
@sio.on('disconnect')
def test_disconnect(sid):
print('Client disconnected')
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=5000)
app2.py
只需要修改端口号
if __name__ == '__main__':
uvicorn.run(app, host='127.0.0.1', port=5001)
接收消息的客户端1 连接app1服务器
import socketio
sio = socketio.Client()
@sio.event
def connect():
print('connection established')
@sio.on("my_event")
def my_message(data):
print('message received with ', data)
@sio.event
def disconnect():
print('disconnected from server')
sio.connect('http://localhost:5000')
sio.wait()
发送消息的客户端 连接app2
import socketio
import json
sio = socketio.Client()
@sio.event
def connect():
print('connection established')
@sio.on("my_event")
def my_message(data):
print('message received with ', data)
@sio.event
def disconnect():
print('disconnected from server')
sio.connect('http://localhost:5001')
data = json.dumps({"message":"拜托了，另一个我","sid":"d84b2e8cbcde479f813627b6ab8e4a47"},ensure_ascii=False)
sio.emit("my_event",data)
sio.wait()
注意发送者需要提供sid 才能夸进程发送成功