如何用python-socketio创建多客户端压测Flask-Socketio服务器并排错

先修复当前代码错误 #

服务器端（app.py） #

当前代码未处理自定义命名空间，导致客户端连接/{thread}命名空间时报错。以下是两种修复方式：

from flask import Flask
from flask_socketio import SocketIO, Namespace

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, cors_allowed_origins="*")

# 方式1：使用默认命名空间（推荐，避免动态命名空间复杂度）
@socketio.on('message')
def handle_message(message):
    print(f"Hello: {message['Hello']}")

# 方式2：若需使用动态命名空间，注册通用命名空间类
# class DynamicNamespace(Namespace):
#     def on_message(self, message):
#         print(f"Hello: {message['Hello']} (namespace: {self.namespace})")
#
# # 提前注册需要的命名空间
# for ns in range(5):
#     socketio.on_namespace(DynamicNamespace(f'/{ns}'))

if __name__ == '__main__':
    # 替换默认Werkzeug为eventlet，提升并发支持
    socketio.run(app, debug=True, async_mode='eventlet')

客户端（test_client.py） #

问题点：全局客户端实例复用导致连接冲突、服务器地址拼接错误、变量未定义、命名空间使用不当。修正后代码：

import socketio
from multiprocessing.pool import ThreadPool

def f(thread_num):
    # 每个线程创建独立的客户端实例
    sio = socketio.Client()
    # 连接基础服务器地址，指定默认命名空间
    sio.connect('http://127.0.0.1:5000', namespaces=['/'])
    # 若使用动态命名空间，需对应服务器端注册的路径
    # sio.connect('http://127.0.0.1:5000', namespaces=[f'/{thread_num}'])
    sio.emit('message', {'Hello': thread_num})
    sio.disconnect()

threads = 5
t = ThreadPool(threads)
t.map(f, range(threads))

10000客户端场景的优化实现 #

服务器端：替换为生产级并发服务器 #

Werkzeug开发服务器仅支持低并发，无法承载10000连接，必须更换为：

• eventlet（Flask-SocketIO官方推荐，原生支持WebSocket）
• gevent（需额外安装gevent-websocket）

安装依赖：

pip install eventlet flask-socketio

生产环境服务器启动代码：

if __name__ == '__main__':
    socketio.run(
        app,
        debug=False,          # 生产环境关闭debug
        async_mode='eventlet',
        host='0.0.0.0',
        port=5000,
        workers=4             # 多进程提升并发能力
    )

客户端：优化并发控制与资源占用 #

1. 异步客户端（推荐）：基于aiohttp的异步客户端比线程更节省系统资源，适合大规模并发

import asyncio
import socketio

async def connect_and_send(thread_num):
    sio = socketio.AsyncClient()
    await sio.connect('http://127.0.0.1:5000')
    await sio.emit('message', {'Hello': thread_num})
    await sio.disconnect()

async def main():
    tasks = [connect_and_send(i) for i in range(10000)]
    # 控制并发数，避免系统资源耗尽
    semaphore = asyncio.Semaphore(100)
    async def bounded_task(task):
        async with semaphore:
            await task
    await asyncio.gather(*[bounded_task(t) for t in tasks])

if __name__ == '__main__':
    asyncio.run(main())

2. 系统资源调整：Linux默认文件描述符限制为1024，无法支持10000连接，需临时调整：

ulimit -n 65535

或永久修改/etc/security/limits.conf文件。

3. 分批启动客户端：一次性启动10000客户端会瞬间耗尽资源，可按批次逐步启动，每批次控制在100-200个。

参考文档 #

• Flask-SocketIO官方文档：包含服务器部署、并发模型、命名空间等核心内容
• Python-SocketIO客户端文档：覆盖同步/异步客户端使用、连接配置等细节

内容的提问来源于stack exchange，提问作者user15033700