Gradio 应用并发压力测试实战基于 load_test 脚本的流式负载验证方案【免费下载链接】gradioBuild and share delightful machine learning apps, all in Python. Star to support our work!项目地址: https://gitcode.com/GitHub_Trending/gr/gradio并发能力是机器学习 Web 应用上生产环境的硬指标。本指南围绕 Gradio 仓库中scripts/load_test目录的完整压测工具链展开讲解如何用 nginx 搭建类生产环境、用三种不同实现纯 Gradio、裸 FastAPI、带 Worker 队列的 FastAPI统一以 500 tokens/100 tokens/s 的流式负载做横向对比最终在 Notebook 中通过并发线程模拟多客户端量化 Gradio 在并发场景下的延迟与吞吐表现。读完本文你将掌握一套可直接复制运行的 Gradio 流式应用压测方法与结果解读思路。一、压测目标与测试资产总览scripts/load_test/README.md 明确了该目录的用途对 Gradio 应用进行负载测试load test验证其同时处理多连接multiple connections的能力。测试的核心场景是流式输出——这是 Chatbot、生成式 AI 应用最常见的实时交互形态。整个目录共包含 6 个文件构成一套完整的环境搭建 → 被测对象 → 并发施压 → 结果对比闭环文件角色说明nginx.conf环境配置生产环境反向代理配置启用 WebSocket 升级与 SSE 直通gradio_app.py被测对象 A纯 Gradio 流式 Chatbot 应用兼容 Gradio 3.x 与 4.xsimple.py被测对象 B裸 FastAPI提供原生 WebSocket 与 SSE 端点不含 Gradioworkers.py被测对象 CFastAPI 队列 Worker 线程模拟 Gradio 的队列实现但剥离其全部额外开销load.ipynb施压工具Notebook对上述应用执行 5/25/100/250 路并发压测README.md说明文档目录用途、环境搭建与被测对象的设计意图三层被测对象的设计思路非常清晰simple.py 测的是裸协议上限workers.py 测的是Gradio 式队列架构下限gradio_app.py 测的是真实 Gradio 开销三者同负载、同速率才能把 Gradio 的框架级损耗从网络和协议层面剥离出来单独评估。二、搭建类生产压测环境nginx 反向代理压测要贴近生产就不能让请求直接打到 Gradio 的开发服务器上。README 给出的建议是For a proper production environment load test, you can run gradio behind an nginx config.操作步骤如下在运行 Gradio 的机器上安装 nginx将仓库中的 nginx.conf 复制到/etc/nginx/conf.d/*.conf将压测施压端跑 Notebook 的机器与 Gradio 服务器分离部署以把网络延迟的影响计入测试结果。关键配置项逐条解析如下对应 nginx.confserver { listen 80; location / { proxy_pass http://127.0.0.1:7860/; # Change this if your Gradio app will be running on a different port proxy_buffering off; proxy_redirect off; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; } }listen 80nginx 监听 80 端口对外提供服务proxy_pass http://127.0.0.1:7860/默认转发到本机 7860 端口Gradio/FastAPI 默认端口如果你的应用改用了其他端口必须同步修改这里proxy_buffering off流式压测的关键。关闭代理缓冲让 SSE 的分块数据一到即转发否则 nginx 会攒满缓冲区才下发破坏流式语义、拉高首字延迟proxy_http_version 1.1上游使用 HTTP/1.1这是 WebSocket 升级的前提HTTP/1.0 不支持 Upgrade 头proxy_set_header Upgrade $http_upgrade与Connection upgrade显式透传 WebSocket 升级握手所需的两个头保证/queue/joinGradio 3.x 的 WebSocket 队列端点与/ws裸 WebSocket 端点能正常穿越代理Host $host保留原始 Host避免后端基于 Host 的路由或校验失效。这套配置同时覆盖了 HTTPSSE 流与 WebSocket 两条链路是压测能测到真实网络效果的前提。三、被测对象一纯 Gradio 流式 Chatbot 应用gradio_app.pygradio_app.py 是一个刻意精简的 Gradio 应用一个 Chatbot 一个 Textbox 输入框 一个耗时显示框函数以生成器yield方式流式输出 500 个 token每个 token 间隔 10ms等效 100 tokens/sfrom time import sleep import gradio as gr version, _, _ gr.__version__.split(.) with gr.Blocks() as demo: chatbot gr.Chatbot() text gr.Textbox() time gr.Number(labelTime to Complete) def respond(text): output [Lorem] * 500 for i in range(len(output) 1): yield [[text, .join(output[:i])]] sleep(0.01) if version 3: text.submit(respond, text, chatbot) else: text.submit(respond, text, chatbot, concurrency_limitNone) if __name__ __main__: if version 3: demo.queue(concurrency_count250).launch(max_threads250) else: demo.launch(max_threads250)几个值得注意的设计点版本自适应通过gr.__version__.split(.)读取主版本号3.x 走text.submit(...)默认路径4.x 显式传入concurrency_limitNone。README 特别注明该应用compatible with gradio 3.x as well即同一份代码可分别部署在 3.x 与 4.x 上做跨版本并发能力对比。concurrency_limitNone的含义在 Gradio 4.x 中事件默认并发上限由Blocks.queue()的default_concurrency_limit决定默认值为 1即同一事件同一时刻只跑一个实例。设置为None表示不限制该事件的并发实例数这正是压测想要的——否则 250 路并发请求会被串行排队测不出服务器的真实吞吐上限。对应实现见 gradio/events.py 中对concurrency_limit的文档说明Can be set to None to mean no concurrency_limit。max_threads2504.x 中demo.launch(max_threads250)将 FastAPI 的线程池上限放开到 2503.x 则通过demo.queue(concurrency_count250).launch(max_threads250)设置队列并发数为 250。两者都是为了支撑 250 路并发压测的容量。从源码层面看Gradio 4.x 的流式响应走的是 gradio/routes.py 中定义的/queue/joinrouter.post(/queue/join)与/queue/datarouter.get(/queue/data)两个端点客户端先 POST/queue/join提交PredictBody含session_hash、fn_index、data服务端把任务推入blocks._queue并返回event_id随后客户端 GET/queue/data?session_hash...建立 SSE 长连接逐条接收data: ...格式的流式事件直到收到含close_stream的结束消息。这就是 load.ipynb 中 Gradio 4.x 压测脚本所模拟的完整交互协议。四、被测对象二裸 FastAPI 流式端点simple.pysimple.py 不依赖 Gradio只用一个最小 FastAPI 应用同时暴露SSE 与 WebSocket两个端点各自以同样节奏流式产出 500 条消息每条 10ms100 条/simport asyncio from fastapi import FastAPI, WebSocket from fastapi.responses import StreamingResponse app FastAPI() async def number_generator(): for number in range(1, 501): message Lorem * (number 1) yield fdata: {message}\n\n await asyncio.sleep(0.01) app.get(/sse) async def sse(): return StreamingResponse(number_generator(), media_typetext/event-stream) app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() for number in range(1, 501): message Lorem * (number 1) await websocket.send_text(message) await asyncio.sleep(0.01) await websocket.close() import uvicorn uvicorn.run(app, host0.0.0.0, port7860)要点SSE 端点用 FastAPI 的StreamingResponsemedia_typetext/event-stream实现产出标准data: ...\n\n帧WebSocket 端点accept()后直接send_text500 条消息后关闭无任何队列与调度逻辑端口同样固定为 7860与 Gradio 默认端口一致便于在同一 nginx 配置下切换被测对象。README 明确了它的定位对比原生 WebSocket / SSE 流与带 Gradio 开销的流之间的性能差距即测量协议层的基准线。五、被测对象三带队列与 Worker 的 FastAPI 实现workers.pyworkers.py 是整个压测体系中最接近 Gradio 内部机制的参照实现它用 asyncio 队列 1000 个 worker 槽位复刻了 Gradio 队列调度的大致形态但剥掉了 Gradio 的事件校验、会话管理、消息序列化等全部额外开销。核心架构对应 workers.pyEvent数据结构dataclass保存session_id、输入data、输出outputsasyncio.Queue、modesse/ws、websocket引用与completed标志全局队列与槽位queue: list[Event]是待处理事件队列active_jobs: list[Event | None] [None] * 1000是 1000 个并发 worker 槽位调度协程queue_process每 50ms 轮询一次若队列非空且存在空闲槽位就从队列头部取出事件塞进槽位并用asyncio.create_task后台执行process_event对应 workers.py 的run_coro_in_background与queue_processSSE 双端点模式POST /sse/send提交任务并返回session_idGET /sse/listen?session_id...轮询active_jobs找到对应事件再通过StreamingResponse从outputs队列中逐条消费并转为 SSE 帧事件完成后向队列写入None哨兵结束流WebSocket 模式/ws端点accept()后先receive_text()拿到输入创建事件入队然后循环sleep(1)等待completed置位后返回。这个实现模拟了 Gradio 队列的提交任务 → 排队 → worker 执行 → 流式回传四段式生命周期且刻意把 worker 槽位放大到 1000使压测瓶颈集中在调度逻辑而非槽位数量。README 的定位是对比Gradio 式队列 流式与真正 Gradio的性能差距——两者相减即可近似得到 Gradio 框架层会话管理、事件校验、消息编解码等的净开销。六、并发施压load.ipynb 压测脚本逐段拆解load.ipynb 是施压端核心做法是用ThreadPoolExecutor开 N 个线程并发执行request()统计平均完成耗时与平均消息数。只需把URL变量指向被测应用所在地址即可README 说明它支持对 Gradio 3.x、4.x 以及app.py即 simple.py/workers.py 类应用分别压测。6.1 通用并发框架from concurrent.futures import ThreadPoolExecutor def run_in_parallel(func, n): if not callable(func) or not isinstance(n, int) or n 1: raise ValueError(Invalid function or number of repetitions) def task_wrapper(): return func() with ThreadPoolExecutor(max_workersn) as executor: futures [executor.submit(task_wrapper) for _ in range(n)] results [future.result() for future in futures] return resultsrun_in_parallel(func, n)用最多n个线程同时执行n次func返回所有结果。每个request()返回三元组(duration, message_count, output)随后用sum(...)/len(...)求平均耗时与平均消息数。6.2 Gradio 4.x 压测HTTP SSE 协议Gradio 4.x 的队列走 HTTP 接口request()模拟完整的提交 订阅两段式流程对应 load.ipynb 中 Gradio 4 小节def request(): start_time time.time() session_hash uuid.uuid4().hex payload {data: [test], fn_index: 0, session_hash: session_hash} url fhttp://{URL}/ resp requests.post(f{url}queue/join, jsonpayload, timeout5) assert resp.status_code 200 message_count 0 output with requests.get( f{url}queue/data?session_hash{session_hash}, streamTrue ) as response: response.raise_for_status() for line in response.iter_lines(): if line: decoded_line line.decode(utf-8) if decoded_line.startswith(data:): data decoded_line.replace(data: , ) if close_stream in data: break output data message_count 1 end_time time.time() duration end_time - start_time return (duration, message_count, json.loads(output)[output][data])要点每个会话生成唯一session_hashfn_index0指向 Blocks 中第一个注册的事件即text.submitPOST /queue/join提交任务对应 gradio/routes.py 中router.post(/queue/join)的实现服务端入队后返回{event_id: ...}GET /queue/data?session_hash...以streamTrue方式读取 SSE 流逐行解析data:前缀的消息遇到包含close_stream的消息即视为流结束并 break——这正是 gradio/routes.py 的/queue/data端点返回的 SSE 帧格式data: {...}\n\n最终返回(总耗时, 收到的消息数, 最终输出)消息数应稳定在 500 左右实际样本中约 500~535 条含生成器 501 次 yield 与心跳帧。Notebook 中的示例运行结果Gradio 4.x100 路并发为avg_duration ≈ 83.28s, avg_msg ≈ 535.58250 路并发约95.03s / 526.68。这些是仓库内记录的真实样本可用于理解数量级不同机器配置下数值会有显著差异不应视为通用基准。6.3 Gradio 3.x 压测WebSocket 协议Gradio 3.x 的队列走 WebSocketrequest()使用websocket-client库连接ws://URL/queue/join按协议循环收消息ws websocket.create_connection(f{url}queue/join) while True: message ws.recv() message_count 1 message json.loads(message) msg message[msg] if msg send_hash: ws.send(json.dumps({session_hash: session_hash, fn_index: 0})) if msg send_hash: ws.send(json.dumps({ data: [test], event_data: None, fn_index: 0, session_hash: session_hash, })) if msg process_completed: output message[output][data] break握手协议为连接后先收到send_hash回传session_hash与fn_index随后提交实际data最终以process_completed消息作为结束信号。仓库记录样本100 路并发约18.25s / 506.0250 路并发约46.62s / 506.4。6.4 裸 SSE 与裸 WebSocket 压测对应 simple.py 的request()更简单SSErequests.get(f{url}sse, streamTrue)逐行读data:帧message_count 500即停止WebSocketwebsocket.create_connection(f{url}ws)后循环recv()收满 500 条即break并关闭连接。两者都不需要 session、不需要 JSON 信封直接消费原始数据流。6.5 带 Worker 队列的 SSE / WebSocket 压测对应 workers.pySSE 侧需要两步先POST /sse/send提交{data: test}拿到session_id再GET /sse/listen?session_id...流式读取 500 条 SSE 帧WebSocket 侧则先ws.send(test)发送输入再循环recv()收满 500 条。仓库记录样本SSE w/ Workers 在 100 路并发约5.66s / 500.0、250 路约11.79s / 500.0WebSocket w/ Workers 在 100 路约16.15s / 500.0、250 路约40.52s / 500.0。6.6 压测执行清单启动被测应用python gradio_app.py或python simple.py/python workers.py按第二节配置好 nginx验证curl http://服务器IP/可达打开 load.ipynb将URL变量改为被测服务器地址依次执行单元格分别在 1/5/25/100/250 路并发下记录avg_duration与avg_msg更换被测对象重复 1–4横向对比四组数据。七、结果解读与对比方法论整个压测体系提供了四层纵向对比维度这是 README 设计的精髓对比组合说明能分离出的成本裸 SSE vs 裸 WebSocketsimple.py协议层基准网络、协议与 uvicorn 本身的成本simple.py vs workers.py是否引入队列调度队列 worker 调度逻辑的成本workers.py vs gradio_app.py是否引入 Gradio会话管理、校验、编解码等框架净开销Gradio 3.x vs 4.x同应用跨版本版本间队列/传输架构差异解读数据时的几个关键提醒同负载才可比所有被测对象统一为500 条消息、每条 10ms、100 tokens/s的流式节奏因此耗时差异可归因于实现层而非负载差异消息数与心跳帧Gradio 4.x 的 SSE 流中消息数略高于 500样本约 504~535是心跳帧与生成器额外 yield 造成的属于正常现象判断流是否完整应看是否收到close_stream而不是死板地对 500并发数上限与配置强相关concurrency_limitNone、max_threads250、3.x 的concurrency_count250都是为了让 250 路并发不被应用层主动限流若自行调整这些参数压测曲线会随之改变网络延迟是特性不是误差README 明确要求施压端与被测端分机部署压测耗时包含真实网络往返更贴近生产样本数据的适用前提Notebook 中记录的数值来自仓库作者当时的特定机器与网络环境本文仅用于说明输出格式与数量级不应被引用为任何通用性能结论。八、从压测脚本反推 Gradio 的并发模型压测脚本与源码互相印证可以勾勒出 Gradio 4.x 队列的并发工作模型依据 gradio/routes.py 与 gradio/events.py入队POST /queue/join将携带session_hash的请求推入blocks._queue返回event_id队列停止时返回 503队列满时同样以 503 拒绝对应queue_join_helper中的错误映射执行worker 从队列取出事件执行事件级并发受concurrency_limit约束同concurrency_id的事件共享该组最低的并发上限回传GET /queue/data?session_hash...建立 SSE 长连接queue_data_helper后台起一个心跳任务周期性写入HeartbeatMessage同时持续从pending_messages_per_session消费生成器产出的事件逐条序列化为data: {...}\n\n帧下发结束生成器耗尽后下发含close_stream的完成消息若客户端断开服务端调用clean_events清理会话并取消心跳任务对应sse_stream中的request.is_disconnected()分支。这套机制解释了压测中并发越高、平均完成耗时越陡增的现象高并发下队列调度、线程池切换、SSE 长连接心跳与客户端逐行解析都会叠加延迟。因此在生产环境为 Gradio 流式应用做容量规划时应依据本文方法在自己机器上跑出并发-耗时曲线再结合concurrency_limit、max_threads、queue()参数与 nginx 缓冲策略proxy_buffering off做针对性调优。九、快速上手速查# 1. 安装被测应用依赖gradio_app.py 需要 gradiosimple/workers 需要 fastapi/uvicorn pip install gradio fastapi uvicorn websocket-client requests # 2. 启动被测应用任选其一 python scripts/load_test/gradio_app.py python scripts/load_test/simple.py python scripts/load_test/workers.py # 3. 配置 nginx复制仓库配置到 conf.d 并重启 cp scripts/load_test/nginx.conf /etc/nginx/conf.d/loadtest.conf systemctl reload nginx # 4. 打开 scripts/load_test/load.ipynb设置 URL 为被测服务器地址依次运行单元格压测完成后用avg_duration平均完成耗时与avg_msg平均消息数两个指标按第七节的四层对比框架整理成表即可得到一份可量化、可复现的 Gradio 并发能力报告。这套工具链随 Gradio 仓库持续演进具体参数与端点行为以当前仓库内 gradio/routes.py、gradio/events.py 的实现为准。【免费下载链接】gradioBuild and share delightful machine learning apps, all in Python. Star to support our work!项目地址: https://gitcode.com/GitHub_Trending/gr/gradio创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Gradio 应用并发压力测试实战：基于 load_test 脚本的流式负载验证方案
Gradio 应用并发压力测试实战基于 load_test 脚本的流式负载验证方案【免费下载链接】gradioBuild and share delightful machine learning apps, all in Python. Star to support our work!项目地址: https://gitcode.com/GitHub_Trending/gr/gradio并发能力是机器学习 Web 应用上生产环境的硬指标。本指南围绕 Gradio 仓库中scripts/load_test目录的完整压测工具链展开讲解如何用 nginx 搭建类生产环境、用三种不同实现纯 Gradio、裸 FastAPI、带 Worker 队列的 FastAPI统一以 500 tokens/100 tokens/s 的流式负载做横向对比最终在 Notebook 中通过并发线程模拟多客户端量化 Gradio 在并发场景下的延迟与吞吐表现。读完本文你将掌握一套可直接复制运行的 Gradio 流式应用压测方法与结果解读思路。一、压测目标与测试资产总览scripts/load_test/README.md 明确了该目录的用途对 Gradio 应用进行负载测试load test验证其同时处理多连接multiple connections的能力。测试的核心场景是流式输出——这是 Chatbot、生成式 AI 应用最常见的实时交互形态。整个目录共包含 6 个文件构成一套完整的环境搭建 → 被测对象 → 并发施压 → 结果对比闭环文件角色说明nginx.conf环境配置生产环境反向代理配置启用 WebSocket 升级与 SSE 直通gradio_app.py被测对象 A纯 Gradio 流式 Chatbot 应用兼容 Gradio 3.x 与 4.xsimple.py被测对象 B裸 FastAPI提供原生 WebSocket 与 SSE 端点不含 Gradioworkers.py被测对象 CFastAPI 队列 Worker 线程模拟 Gradio 的队列实现但剥离其全部额外开销load.ipynb施压工具Notebook对上述应用执行 5/25/100/250 路并发压测README.md说明文档目录用途、环境搭建与被测对象的设计意图三层被测对象的设计思路非常清晰simple.py 测的是裸协议上限workers.py 测的是Gradio 式队列架构下限gradio_app.py 测的是真实 Gradio 开销三者同负载、同速率才能把 Gradio 的框架级损耗从网络和协议层面剥离出来单独评估。二、搭建类生产压测环境nginx 反向代理压测要贴近生产就不能让请求直接打到 Gradio 的开发服务器上。README 给出的建议是For a proper production environment load test, you can run gradio behind an nginx config.操作步骤如下在运行 Gradio 的机器上安装 nginx将仓库中的 nginx.conf 复制到/etc/nginx/conf.d/*.conf将压测施压端跑 Notebook 的机器与 Gradio 服务器分离部署以把网络延迟的影响计入测试结果。关键配置项逐条解析如下对应 nginx.confserver { listen 80; location / { proxy_pass http://127.0.0.1:7860/; # Change this if your Gradio app will be running on a different port proxy_buffering off; proxy_redirect off; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_set_header Host $host; } }listen 80nginx 监听 80 端口对外提供服务proxy_pass http://127.0.0.1:7860/默认转发到本机 7860 端口Gradio/FastAPI 默认端口如果你的应用改用了其他端口必须同步修改这里proxy_buffering off流式压测的关键。关闭代理缓冲让 SSE 的分块数据一到即转发否则 nginx 会攒满缓冲区才下发破坏流式语义、拉高首字延迟proxy_http_version 1.1上游使用 HTTP/1.1这是 WebSocket 升级的前提HTTP/1.0 不支持 Upgrade 头proxy_set_header Upgrade $http_upgrade与Connection upgrade显式透传 WebSocket 升级握手所需的两个头保证/queue/joinGradio 3.x 的 WebSocket 队列端点与/ws裸 WebSocket 端点能正常穿越代理Host $host保留原始 Host避免后端基于 Host 的路由或校验失效。这套配置同时覆盖了 HTTPSSE 流与 WebSocket 两条链路是压测能测到真实网络效果的前提。三、被测对象一纯 Gradio 流式 Chatbot 应用gradio_app.pygradio_app.py 是一个刻意精简的 Gradio 应用一个 Chatbot 一个 Textbox 输入框 一个耗时显示框函数以生成器yield方式流式输出 500 个 token每个 token 间隔 10ms等效 100 tokens/sfrom time import sleep import gradio as gr version, _, _ gr.__version__.split(.) with gr.Blocks() as demo: chatbot gr.Chatbot() text gr.Textbox() time gr.Number(labelTime to Complete) def respond(text): output [Lorem] * 500 for i in range(len(output) 1): yield [[text, .join(output[:i])]] sleep(0.01) if version 3: text.submit(respond, text, chatbot) else: text.submit(respond, text, chatbot, concurrency_limitNone) if __name__ __main__: if version 3: demo.queue(concurrency_count250).launch(max_threads250) else: demo.launch(max_threads250)几个值得注意的设计点版本自适应通过gr.__version__.split(.)读取主版本号3.x 走text.submit(...)默认路径4.x 显式传入concurrency_limitNone。README 特别注明该应用compatible with gradio 3.x as well即同一份代码可分别部署在 3.x 与 4.x 上做跨版本并发能力对比。concurrency_limitNone的含义在 Gradio 4.x 中事件默认并发上限由Blocks.queue()的default_concurrency_limit决定默认值为 1即同一事件同一时刻只跑一个实例。设置为None表示不限制该事件的并发实例数这正是压测想要的——否则 250 路并发请求会被串行排队测不出服务器的真实吞吐上限。对应实现见 gradio/events.py 中对concurrency_limit的文档说明Can be set to None to mean no concurrency_limit。max_threads2504.x 中demo.launch(max_threads250)将 FastAPI 的线程池上限放开到 2503.x 则通过demo.queue(concurrency_count250).launch(max_threads250)设置队列并发数为 250。两者都是为了支撑 250 路并发压测的容量。从源码层面看Gradio 4.x 的流式响应走的是 gradio/routes.py 中定义的/queue/joinrouter.post(/queue/join)与/queue/datarouter.get(/queue/data)两个端点客户端先 POST/queue/join提交PredictBody含session_hash、fn_index、data服务端把任务推入blocks._queue并返回event_id随后客户端 GET/queue/data?session_hash...建立 SSE 长连接逐条接收data: ...格式的流式事件直到收到含close_stream的结束消息。这就是 load.ipynb 中 Gradio 4.x 压测脚本所模拟的完整交互协议。四、被测对象二裸 FastAPI 流式端点simple.pysimple.py 不依赖 Gradio只用一个最小 FastAPI 应用同时暴露SSE 与 WebSocket两个端点各自以同样节奏流式产出 500 条消息每条 10ms100 条/simport asyncio from fastapi import FastAPI, WebSocket from fastapi.responses import StreamingResponse app FastAPI() async def number_generator(): for number in range(1, 501): message Lorem * (number 1) yield fdata: {message}\n\n await asyncio.sleep(0.01) app.get(/sse) async def sse(): return StreamingResponse(number_generator(), media_typetext/event-stream) app.websocket(/ws) async def websocket_endpoint(websocket: WebSocket): await websocket.accept() for number in range(1, 501): message Lorem * (number 1) await websocket.send_text(message) await asyncio.sleep(0.01) await websocket.close() import uvicorn uvicorn.run(app, host0.0.0.0, port7860)要点SSE 端点用 FastAPI 的StreamingResponsemedia_typetext/event-stream实现产出标准data: ...\n\n帧WebSocket 端点accept()后直接send_text500 条消息后关闭无任何队列与调度逻辑端口同样固定为 7860与 Gradio 默认端口一致便于在同一 nginx 配置下切换被测对象。README 明确了它的定位对比原生 WebSocket / SSE 流与带 Gradio 开销的流之间的性能差距即测量协议层的基准线。五、被测对象三带队列与 Worker 的 FastAPI 实现workers.pyworkers.py 是整个压测体系中最接近 Gradio 内部机制的参照实现它用 asyncio 队列 1000 个 worker 槽位复刻了 Gradio 队列调度的大致形态但剥掉了 Gradio 的事件校验、会话管理、消息序列化等全部额外开销。核心架构对应 workers.pyEvent数据结构dataclass保存session_id、输入data、输出outputsasyncio.Queue、modesse/ws、websocket引用与completed标志全局队列与槽位queue: list[Event]是待处理事件队列active_jobs: list[Event | None] [None] * 1000是 1000 个并发 worker 槽位调度协程queue_process每 50ms 轮询一次若队列非空且存在空闲槽位就从队列头部取出事件塞进槽位并用asyncio.create_task后台执行process_event对应 workers.py 的run_coro_in_background与queue_processSSE 双端点模式POST /sse/send提交任务并返回session_idGET /sse/listen?session_id...轮询active_jobs找到对应事件再通过StreamingResponse从outputs队列中逐条消费并转为 SSE 帧事件完成后向队列写入None哨兵结束流WebSocket 模式/ws端点accept()后先receive_text()拿到输入创建事件入队然后循环sleep(1)等待completed置位后返回。这个实现模拟了 Gradio 队列的提交任务 → 排队 → worker 执行 → 流式回传四段式生命周期且刻意把 worker 槽位放大到 1000使压测瓶颈集中在调度逻辑而非槽位数量。README 的定位是对比Gradio 式队列 流式与真正 Gradio的性能差距——两者相减即可近似得到 Gradio 框架层会话管理、事件校验、消息编解码等的净开销。六、并发施压load.ipynb 压测脚本逐段拆解load.ipynb 是施压端核心做法是用ThreadPoolExecutor开 N 个线程并发执行request()统计平均完成耗时与平均消息数。只需把URL变量指向被测应用所在地址即可README 说明它支持对 Gradio 3.x、4.x 以及app.py即 simple.py/workers.py 类应用分别压测。6.1 通用并发框架from concurrent.futures import ThreadPoolExecutor def run_in_parallel(func, n): if not callable(func) or not isinstance(n, int) or n 1: raise ValueError(Invalid function or number of repetitions) def task_wrapper(): return func() with ThreadPoolExecutor(max_workersn) as executor: futures [executor.submit(task_wrapper) for _ in range(n)] results [future.result() for future in futures] return resultsrun_in_parallel(func, n)用最多n个线程同时执行n次func返回所有结果。每个request()返回三元组(duration, message_count, output)随后用sum(...)/len(...)求平均耗时与平均消息数。6.2 Gradio 4.x 压测HTTP SSE 协议Gradio 4.x 的队列走 HTTP 接口request()模拟完整的提交 订阅两段式流程对应 load.ipynb 中 Gradio 4 小节def request(): start_time time.time() session_hash uuid.uuid4().hex payload {data: [test], fn_index: 0, session_hash: session_hash} url fhttp://{URL}/ resp requests.post(f{url}queue/join, jsonpayload, timeout5) assert resp.status_code 200 message_count 0 output with requests.get( f{url}queue/data?session_hash{session_hash}, streamTrue ) as response: response.raise_for_status() for line in response.iter_lines(): if line: decoded_line line.decode(utf-8) if decoded_line.startswith(data:): data decoded_line.replace(data: , ) if close_stream in data: break output data message_count 1 end_time time.time() duration end_time - start_time return (duration, message_count, json.loads(output)[output][data])要点每个会话生成唯一session_hashfn_index0指向 Blocks 中第一个注册的事件即text.submitPOST /queue/join提交任务对应 gradio/routes.py 中router.post(/queue/join)的实现服务端入队后返回{event_id: ...}GET /queue/data?session_hash...以streamTrue方式读取 SSE 流逐行解析data:前缀的消息遇到包含close_stream的消息即视为流结束并 break——这正是 gradio/routes.py 的/queue/data端点返回的 SSE 帧格式data: {...}\n\n最终返回(总耗时, 收到的消息数, 最终输出)消息数应稳定在 500 左右实际样本中约 500~535 条含生成器 501 次 yield 与心跳帧。Notebook 中的示例运行结果Gradio 4.x100 路并发为avg_duration ≈ 83.28s, avg_msg ≈ 535.58250 路并发约95.03s / 526.68。这些是仓库内记录的真实样本可用于理解数量级不同机器配置下数值会有显著差异不应视为通用基准。6.3 Gradio 3.x 压测WebSocket 协议Gradio 3.x 的队列走 WebSocketrequest()使用websocket-client库连接ws://URL/queue/join按协议循环收消息ws websocket.create_connection(f{url}queue/join) while True: message ws.recv() message_count 1 message json.loads(message) msg message[msg] if msg send_hash: ws.send(json.dumps({session_hash: session_hash, fn_index: 0})) if msg send_hash: ws.send(json.dumps({ data: [test], event_data: None, fn_index: 0, session_hash: session_hash, })) if msg process_completed: output message[output][data] break握手协议为连接后先收到send_hash回传session_hash与fn_index随后提交实际data最终以process_completed消息作为结束信号。仓库记录样本100 路并发约18.25s / 506.0250 路并发约46.62s / 506.4。6.4 裸 SSE 与裸 WebSocket 压测对应 simple.py 的request()更简单SSErequests.get(f{url}sse, streamTrue)逐行读data:帧message_count 500即停止WebSocketwebsocket.create_connection(f{url}ws)后循环recv()收满 500 条即break并关闭连接。两者都不需要 session、不需要 JSON 信封直接消费原始数据流。6.5 带 Worker 队列的 SSE / WebSocket 压测对应 workers.pySSE 侧需要两步先POST /sse/send提交{data: test}拿到session_id再GET /sse/listen?session_id...流式读取 500 条 SSE 帧WebSocket 侧则先ws.send(test)发送输入再循环recv()收满 500 条。仓库记录样本SSE w/ Workers 在 100 路并发约5.66s / 500.0、250 路约11.79s / 500.0WebSocket w/ Workers 在 100 路约16.15s / 500.0、250 路约40.52s / 500.0。6.6 压测执行清单启动被测应用python gradio_app.py或python simple.py/python workers.py按第二节配置好 nginx验证curl http://服务器IP/可达打开 load.ipynb将URL变量改为被测服务器地址依次执行单元格分别在 1/5/25/100/250 路并发下记录avg_duration与avg_msg更换被测对象重复 1–4横向对比四组数据。七、结果解读与对比方法论整个压测体系提供了四层纵向对比维度这是 README 设计的精髓对比组合说明能分离出的成本裸 SSE vs 裸 WebSocketsimple.py协议层基准网络、协议与 uvicorn 本身的成本simple.py vs workers.py是否引入队列调度队列 worker 调度逻辑的成本workers.py vs gradio_app.py是否引入 Gradio会话管理、校验、编解码等框架净开销Gradio 3.x vs 4.x同应用跨版本版本间队列/传输架构差异解读数据时的几个关键提醒同负载才可比所有被测对象统一为500 条消息、每条 10ms、100 tokens/s的流式节奏因此耗时差异可归因于实现层而非负载差异消息数与心跳帧Gradio 4.x 的 SSE 流中消息数略高于 500样本约 504~535是心跳帧与生成器额外 yield 造成的属于正常现象判断流是否完整应看是否收到close_stream而不是死板地对 500并发数上限与配置强相关concurrency_limitNone、max_threads250、3.x 的concurrency_count250都是为了让 250 路并发不被应用层主动限流若自行调整这些参数压测曲线会随之改变网络延迟是特性不是误差README 明确要求施压端与被测端分机部署压测耗时包含真实网络往返更贴近生产样本数据的适用前提Notebook 中记录的数值来自仓库作者当时的特定机器与网络环境本文仅用于说明输出格式与数量级不应被引用为任何通用性能结论。八、从压测脚本反推 Gradio 的并发模型压测脚本与源码互相印证可以勾勒出 Gradio 4.x 队列的并发工作模型依据 gradio/routes.py 与 gradio/events.py入队POST /queue/join将携带session_hash的请求推入blocks._queue返回event_id队列停止时返回 503队列满时同样以 503 拒绝对应queue_join_helper中的错误映射执行worker 从队列取出事件执行事件级并发受concurrency_limit约束同concurrency_id的事件共享该组最低的并发上限回传GET /queue/data?session_hash...建立 SSE 长连接queue_data_helper后台起一个心跳任务周期性写入HeartbeatMessage同时持续从pending_messages_per_session消费生成器产出的事件逐条序列化为data: {...}\n\n帧下发结束生成器耗尽后下发含close_stream的完成消息若客户端断开服务端调用clean_events清理会话并取消心跳任务对应sse_stream中的request.is_disconnected()分支。这套机制解释了压测中并发越高、平均完成耗时越陡增的现象高并发下队列调度、线程池切换、SSE 长连接心跳与客户端逐行解析都会叠加延迟。因此在生产环境为 Gradio 流式应用做容量规划时应依据本文方法在自己机器上跑出并发-耗时曲线再结合concurrency_limit、max_threads、queue()参数与 nginx 缓冲策略proxy_buffering off做针对性调优。九、快速上手速查# 1. 安装被测应用依赖gradio_app.py 需要 gradiosimple/workers 需要 fastapi/uvicorn pip install gradio fastapi uvicorn websocket-client requests # 2. 启动被测应用任选其一 python scripts/load_test/gradio_app.py python scripts/load_test/simple.py python scripts/load_test/workers.py # 3. 配置 nginx复制仓库配置到 conf.d 并重启 cp scripts/load_test/nginx.conf /etc/nginx/conf.d/loadtest.conf systemctl reload nginx # 4. 打开 scripts/load_test/load.ipynb设置 URL 为被测服务器地址依次运行单元格压测完成后用avg_duration平均完成耗时与avg_msg平均消息数两个指标按第七节的四层对比框架整理成表即可得到一份可量化、可复现的 Gradio 并发能力报告。这套工具链随 Gradio 仓库持续演进具体参数与端点行为以当前仓库内 gradio/routes.py、gradio/events.py 的实现为准。【免费下载链接】gradioBuild and share delightful machine learning apps, all in Python. Star to support our work!项目地址: https://gitcode.com/GitHub_Trending/gr/gradio创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考