flask websocket服务搭建，flask-sock 和 flask-socketio

flask websocket服务搭建，flask-sock 和 flask-socketio

首先说下socket、socketio 和 websockets 之间的区别

socket是网络上运行的两个程序之间的双向通信链路的一个端点。这是一个非常低级的东西，其他统统都是在 TCP 套接字之上实现的。 WebSocket 是 Web 的标准通信协议。它允许在客户端和服务器之间创建全双工通信通道。 Socket.IO 是一种基于 HTTP 和 WebSocket 构建的通信协议，提供自动重新连接、基于变乱的通知等额外功能。 Flask-SocketIO 是 Socket.IO 服务器端协议的实现，作为 Flask 扩展。
两个主流框架flask-sock 和 flask-socketio
flask-sock是原生的websocket协议，flask-socketio是socketio协议，必要根据你自己的需求选择合适的框架，假如用socketio那么客户端建议也用socketio库举行连接
websocket收发的都是原生的数据包
而socketio在原生的数据包上还加了协议数据好比状态码和监听变乱等
特性WebSocketSocket.IO连接协议基于WebSocket协议封装的WebSocket协议，提供更多功能数据包结构包含Opcode, Payload Length, Mask, Payload Data包含Event name和Data功能界说标准的双向通信，没有内建的重连或变乱支持支持变乱、重连、自动重试、定名空间等功能发送和吸收方式通过raw WebSocket send和onmessage直接发送和吸收封装的数据通过emit和on方法，使用变乱名来识别支持的传输方式只支持WebSocket支持多个传输方式（WebSocket, Polling等）

安装两个库

1. pip install flask-socketio flask-sock

复制代码

flask-sock例子

服务端example

1. from flask import Flask, render_template
2. from flask_sock import Sock
4. app = Flask(__name__)
5. app.config['SOCK_SERVER_OPTIONS'] = {'ping_interval': 25}
7. sock = Sock(app)
10. @app.route('/')
11. def index():
12.     return render_template('index.html')
15. @sock.route('/echo')
16. def echo(ws):
17.     while True:
18.         data = ws.receive()
19.         if data == 'close':
20.             break
21.         ws.send(data)
24. if __name__ == '__main__':
25.     app.run()

复制代码

客户端example

1. from flask import Flask, render_template
2. from flask_sock import Sock
4. app = Flask(__name__)
5. app.config['SOCK_SERVER_OPTIONS'] = {'ping_interval': 25}
7. sock = Sock(app)
10. @app.route('/')
11. def index():
12.     return render_template('index.html')
15. @sock.route('/echo')
16. def echo(ws):
17.     while True:
18.         data = ws.receive()
19.         if data == 'close':
20.             break
21.         ws.send(data)
24. if __name__ == '__main__':
25.     app.run()

复制代码

flask-socketio例子

服务端example

1. from flask import Flask, render_template
2. from flask_socketio import SocketIO, emit
4. app = Flask(__name__)
5. app.config['SECRET_KEY'] = 'your_secret_key'  # 设置一个密钥用于会话、Cookie等
7. socketio = SocketIO(app)
9. @app.route('/')
10. def index():
11.     return render_template('index.html')
13. @socketio.on('message')
14. def handle_message(message):
15.     print('Received message: ' + message)
16.     emit('message', message)  # Echo the received message back
18. @socketio.on('close')
19. def handle_close():
20.     print('Closing connection')
22. if __name__ == '__main__':
23.     socketio.run(app)

复制代码

客户端example

1. import time
2. import socketio
4. # 创建 SocketIO 客户端实例
5. sio = socketio.Client()
7. @sio.event
8. def connect():
9.     print("Connection established")
11. @sio.event
12. def message(data):
13.     print(f"Message received: {data}")
15. @sio.event
16. def disconnect():
17.     print("Disconnected from server")
19. # 连接到服务器
20. sio.connect('http://localhost:5000')
22. # 发送消息
23. sio.send("Hello, Socket.IO!")
25. # 等待一些时间以接收消息
26. time.sleep(1)
28. # 发送关闭消息并断开连接
29. sio.disconnect()

复制代码

假如要用websocket格式来测试这个socketio接口

1. import websockets
2. import asyncio
3. import json
4. import time
5. import random
7. async def test_socketio():
8.     # Socket.IO 在 WebSocket URL 上添加特定的路径和参数
9.     uri = "ws://localhost:5000/socket.io/?EIO=4&transport=websocket"
10.    
11.     async with websockets.connect(uri) as ws:
12.         print("Connected to server")
13.         
14.         # 发送 Socket.IO 握手消息
15.         await ws.send("40")  # Socket.IO v4 握手消息
16.         
17.         # 发送消息
18.         message = {
19.             "type": "message",
20.             "data": "Hello, Socket.IO!"
21.         }
22.         
23.         # Socket.IO 消息格式: "42" + JSON数组 [event, data]
24.         socketio_msg = f'42{json.dumps(["message", message])}'
25.         await ws.send(socketio_msg)
26.         
27.         # 接收消息
28.         try:
29.             while True:
30.                 response = await asyncio.wait_for(ws.recv(), timeout=2)
31.                 print(f"Received: {response}")
32.                
33.                 # 如果是 PING 消息（2），回应 PONG（3）
34.                 if response.startswith("2"):
35.                     await ws.send("3")
36.                     continue
37.                
38.                 # 解析常规消息
39.                 if response.startswith("42"):
40.                     try:
41.                         data = json.loads(response[2:])
42.                         print(f"Parsed message: {data}")
43.                     except json.JSONDecodeError:
44.                         print("Failed to parse message")
45.         
46.         except asyncio.TimeoutError:
47.             print("No more messages")
48.         
49.         # 发送关闭消息
50.         await ws.send("41")  # Socket.IO 关闭消息
52. if __name__ == "__main__":
53.     asyncio.run(test_socketio())

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。