使用Streamlit与Asyncio实现实时数据拉取的优化问题咨询

问题分析与优化方向 #

1. 循环阻塞页面渲染：原代码的死循环会卡住主线程，导致后续组件无法加载。需要改用Streamlit原生的刷新机制实现非阻塞更新。
2. 异步函数实现不规范：原异步函数在while True循环内用return会直接终止协程，不符合持续生成数据的逻辑，需通过状态管理传递数据。

优化后完整代码 #

import asyncio
import streamlit as st
import numpy as np

st.set_page_config(layout="wide")

# 初始化会话状态，存储实时数据
if "real_time_data" not in st.session_state:
    st.session_state.real_time_data = np.zeros(5)

async def data_generator():
    # 持续生成/拉取实时数据，这里用随机数模拟业务逻辑
    while True:
        st.session_state.real_time_data = np.random.randn(5, 1)
        await asyncio.sleep(5)  # 按需求设置5秒间隔

def update_dashboard():
    # 按需更新仪表盘内容，后续可扩展更多图表/组件
    st.write(st.session_state.real_time_data[0])

# 确保异步任务仅启动一次
if "task_started" not in st.session_state:
    asyncio.create_task(data_generator())
    st.session_state.task_started = True

# 渲染仪表盘和其他页面元素，无阻塞
update_dashboard()
st.write("TESTING!")  # 现在可放在任意位置正常显示

# 定时刷新页面，实现仪表盘实时更新
st.rerun(interval=5)

核心优化说明 #

• 非阻塞页面更新：使用st.rerun(interval=5)替代死循环，页面每隔5秒自动刷新，不会阻塞后续组件渲染，方便后续扩展仪表盘元素。
• 规范异步数据处理：异步函数不再返回数据，而是直接更新Streamlit会话状态st.session_state，通过状态管理实现数据在组件间的共享，符合Streamlit的设计逻辑。
• 避免重复异步任务：用会话状态标记task_started，确保异步数据生成任务仅在页面首次加载时启动一次，防止资源浪费。

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