Streamlit 和 Dash 都是 Python 中用于快速构建数据可视化 Web 应用尤其是动态数据看板的开源框架但设计理念、使用场景和开发体验有显著差异✅Streamlit定位面向数据科学家/分析师强调“写脚本即应用”无需前端知识。动态更新通过st.experimental_rerun()、st.session_state、st.cache_datav1.32配合st.cache_resource实现轻量级状态管理与数据刷新支持st.autorefresh()第三方组件或结合st.empty()time.sleep()轮询适合低频实时。优势开发极快、代码简洁纯 Python、内置交互组件slider, button, file_uploader 等、天然支持 Markdown/图表库Matplotlib/Plotly/Altair。局限默认无原生 WebSocket高并发/复杂状态管理需谨慎不适合大型企业级 SPA。✅Dash由 Plotly 开发定位面向工程师/产品化需求基于 React 构建声明式回调callback驱动 UI 更新。动态更新原生支持dcc.Interval组件实现定时轮询可集成 Flask-SocketIO 或自定义后端 API 实现 WebSocket 推送dash.long_callback支持异步/后台任务。优势组件生态丰富dash-bootstrap-components、dash-ag-grid、强类型回调、服务端状态可控、易于部署与权限扩展。局限学习曲线略高需理解 callback 输入/输出依赖、代码量相对多。选型建议快速原型、内部汇报看板、小团队数据监控 → 优先 Streamlit多用户权限、高频实时如秒级股票行情、复杂交互逻辑、需与现有 Flask/Django 集成 → 优选 Dash。# Streamlit 动态示例每3秒刷新随机折线图importstreamlitasstimportpandasaspdimportnumpyasnpimporttime st.title( 动态数据看板Streamlit)placeholderst.empty()foriinrange(100):datapd.DataFrame({t:np.arange(i1),value:np.cumsum(np.random.randn(i1).astype(float))})withplaceholder.container():st.line_chart(data.set_index(t),height300)st.text(f更新时间:{pd.Timestamp.now().strftime(%H:%M:%S)})time.sleep(3)# Dash 动态示例核心结构需完整 app.py 运行fromdashimportDash,dcc,html,Input,Output,callbackimportplotly.expressaspximportpandasaspdimportnumpyasnp appDash(__name__)app.layouthtml.Div([html.H1( 动态数据看板Dash),dcc.Graph(idlive-graph),dcc.Interval(idinterval-component,interval3000,n_intervals0)# 3秒触发一次])callback(Output(live-graph,figure),Input(interval-component,n_intervals))defupdate_graph(n):dfpd.DataFrame({t:list(range(n1)),value:np.cumsum(np.random.randn(n1))})returnpx.line(df,xt,yvalue,titlef更新次数:{n})# app.run_server(debugTrue)Streamlit原生不支持 WebSocket 服务端推送即无法像 Dash Flask-SocketIO 或 FastAPI 那样主动向客户端“推”数据其架构是请求-响应式stateless HTTPUI 更新依赖客户端主动刷新或服务端重载st.rerun()。但可通过以下工程化方案实现“类 WebSocket”的实时体验尤其适用于 MQTT、Kafka、数据库变更等外部事件驱动场景✅ 推荐方案后端事件监听 客户端轮询 状态共享推荐生产使用# requirements.txtstreamlit1.35.0paho-mqtt2.0.0redis5.0.5# 或使用 st.cache_resource threading.Event轻量▶ 步骤 1独立后台线程监听 MQTT避免阻塞 Streamlit 主线程# mqtt_listener.py —— 在 Streamlit 启动时运行一次importthreadingimportpaho.mqtt.clientasmqttimportjsonfromstreamlit.runtime.scriptrunnerimportadd_script_run_ctxfromstreamlitimportcache_resource# 全局共享状态线程安全latest_data{timestamp:0,value:0}data_lockthreading.Lock()defon_message(client,userdata,msg):globallatest_datatry:payloadjson.loads(msg.payload.decode())withdata_lock:latest_data{timestamp:time.time(),value:payload.get(value,0)}exceptExceptionase:print(fMQTT parse error:{e})cache_resourcedefstart_mqtt_listener():clientmqtt.Client()client.on_messageon_message client.connect(broker.hivemq.com,1883,60)# 公共测试 Brokerclient.subscribe(streamlit/realtime)# 启动为守护线程Streamlit 退出时自动结束threadthreading.Thread(targetclient.loop_forever,daemonTrue)add_script_run_ctx(thread)# 关联到当前脚本上下文thread.start()returnclient▶ 步骤 2Streamlit 页面轮询 智能刷新避免全页重载# app.pyimportstreamlitasstimporttimefrommqtt_listenerimportlatest_data,data_lock st.title( MQTT 实时看板类 WebSocket)# 使用 st.empty() 局部更新不触发整页 rerunchart_placeholderst.empty()status_placeholderst.empty()# 持续轮询客户端 JS 不感知服务端控制whileTrue:withdata_lock:data_copylatest_data.copy()# 用 Plotly 动态图支持增量更新dfpd.DataFrame([data_copy])figpx.scatter(df,xtimestamp,yvalue,titleMQTT 实时数据点)chart_placeholder.plotly_chart(fig,use_container_widthTrue)status_placeholder.text(f✅ 最新接收时间:{time.strftime(%H:%M:%S,time.localtime(data_copy[timestamp]))})time.sleep(1)# 轮询间隔可动态调整✅ 优势无前端 JS、兼容所有部署环境Cloud、Docker、线程安全、低延迟~1s⚠️ 注意while True会占用一个线程建议配合st.stop()或st.session_state控制启停。✅ 进阶方案Streamlit FastAPI分离前后端—— 真正的 WebSocket若需毫秒级推送如高频交易监控推荐将 Streamlit 作为纯前端静态页面用FastAPI 提供 WebSocket API再通过st.components.v1.html注入自定义 JS 订阅# fastapi_backend.pyfromfastapiimportFastAPI,WebSocket,WebSocketDisconnectimportasyncio appFastAPI()classConnectionManager:def__init__(self):self.active_connections[]asyncdefconnect(self,ws):awaitws.accept();self.active_connections.append(ws)defdisconnect(self,ws):self.active_connections.remove(ws)asyncdefbroadcast(self,message):forwsinself.active_connections:awaitws.send_text(message)managerConnectionManager()app.websocket(/ws)asyncdefwebsocket_endpoint(websocket:WebSocket):awaitmanager.connect(websocket)try:whileTrue:# 模拟接收 MQTT 消息实际中此处桥接 paho-mqttawaitasyncio.sleep(1)awaitmanager.broadcast(json.dumps({value:round(np.random.randn(),3)}))exceptWebSocketDisconnect:manager.disconnect(websocket)# streamlit_app.py —— 前端仅负责展示importstreamlitasstimportjson st.markdown( script const ws new WebSocket(ws://localhost:8000/ws); let chartData []; ws.onmessage (event) { const d JSON.parse(event.data); chartData.push({t: new Date().getTime(), v: d.value}); if (chartData.length 100) chartData.shift(); // 使用 Chart.js / Plotly.js 渲染此处省略 }; /script ,unsafe_allow_htmlTrue)❌ 不推荐方案已淘汰/不稳定st.experimental_rerun()配合全局变量 → 状态丢失、竞态严重streamlit-autorefresh第三方库 → 仅 HTTP 轮询非真正推送修改 Streamlit 源码注入 WebSocket → 维护成本高违反设计哲学 总结Streamlit “实时”的本质是「服务端事件捕获 客户端可控轮询」方案延迟复杂度是否需额外服务适用场景后台线程 st.empty()轮询~0.5–2s★★☆否内部监控、IoT 传感器秒级Streamlit FastAPI WebSocket100ms★★★★是FastAPI金融行情、多人协作白板外部消息队列 Webhook 触发st.rerun()3s★★是如云函数低频告警通知如异常检测关键提示Streamlit 的设计哲学是「简单优先」真正的双向 WebSocket 属于基础设施层职责。与其强行改造 Streamlit不如用它做快速 UI把实时性交给专业后端FastAPI/Node.js通过 REST/WebSocket 解耦——这才是云原生时代的最佳实践。