如何改造Dash实时绘图代码为可复用函数实现多图批量更新

• 封装图表实例：将每个实时图表的独立状态（X/Y队列、ID、坐标轴配置等）封装为类，避免全局变量冲突，天然支持多实例复用
• 动态注册回调：每个图表实例初始化时自动注册对应ID的更新回调，不需要硬编码多套callback逻辑
• 对接socket数据源：单独启动后台线程接收socket数据，收到数据后直接调用对应图表实例的方法更新内部队列，回调触发时直接读取队列生成图表即可

import dash
from dash import dcc, html, Input, Output
import plotly.graph_objects as go
from collections import deque
import random
import threading
import time

class RealtimeChart:
    def __init__(self, chart_id, max_data_len=100, interval=1000, y_range=(0,100)):
        self.chart_id = chart_id
        self.interval_id = f"{chart_id}-interval"
        self.max_data_len = max_data_len
        self.y_range = y_range
        # 加锁保证线程安全，避免socket接收线程和Dash回调线程同时操作队列异常
        self.lock = threading.Lock()
        self.X = deque(maxlen=max_data_len)
        self.Y = deque(maxlen=max_data_len)
        # 初始值填充
        self.X.append(1)
        self.Y.append(0)
    
    def get_layout_components(self):
        # 返回当前图表对应的页面组件，直接添加到app的layout列表即可
        return html.Div([
            html.H3(f"实时图表 {self.chart_id}"),
            dcc.Graph(id=self.chart_id, animate=True),
            dcc.Interval(
                id=self.interval_id,
                interval=self.interval,
                n_intervals=0
            )
        ])
    
    def register_callback(self, app):
        # 为当前图表自动注册专属更新回调
        @app.callback(
            Output(self.chart_id, 'figure'),
            Input(self.interval_id, 'n_intervals')
        )
        def update_graph(n):
            with self.lock:
                x_data = list(self.X)
                y_data = list(self.Y)
            # 构造散点折线图
            data = go.Scatter(
                x=x_data,
                y=y_data,
                name='实时数据',
                mode='lines+markers'
            )
            # 动态适配X轴范围
            if max(x_data) < self.max_data_len:
                x_range = [0, self.max_data_len]
            else:
                x_range = [min(x_data), max(x_data)]
            return {
                'data': [data],
                'layout': go.Layout(
                    xaxis=dict(range=x_range),
                    yaxis=dict(range=self.y_range)
                )
            }
    
    def push_new_data(self, y_value):
        # 对外暴露的更新接口，收到socket数据后调用该方法即可更新对应图表数据源
        with self.lock:
            self.X.append(self.X[-1] + 1)
            self.Y.append(y_value)

if __name__ == '__main__':
    app = dash.Dash(__name__)
    # 示例创建2个不同配置的实时图表，可按需新增
    chart1 = RealtimeChart(chart_id="temp_chart", max_data_len=100, interval=1000, y_range=(0,100))
    chart2 = RealtimeChart(chart_id="humidity_chart", max_data_len=80, interval=1000, y_range=(0,200))
    # 注册两个图表的回调
    chart1.register_callback(app)
    chart2.register_callback(app)
    # 把图表组件加到页面布局，这里配置为左右分栏展示
    app.layout = html.Div([
        html.Div(chart1.get_layout_components(), style={"width": "48%", "display": "inline-block", "vertical-align": "top"}),
        html.Div(chart2.get_layout_components(), style={"width": "48%", "display": "inline-block", "vertical-align": "top"})
    ])
    # 模拟socket接收数据的后台线程，替换为你自己的socket接收解析逻辑即可
    def socket_receive_task():
        while True:
            # 这里替换为socket.recv() + 数据解析逻辑，拿到对应指标的数值
            new_temp = random.randint(0,100)
            new_humidity = random.randint(0,200)
            # 推送数据到对应图表
            chart1.push_new_data(new_temp)
            chart2.push_new_data(new_humidity)
            time.sleep(1)
    # 启动后台数据接收线程
    threading.Thread(target=socket_receive_task, daemon=True).start()
    # 启动Dash服务
    app.run_server(debug=True)

• 把示例中socket_receive_task里的随机数生成逻辑替换为你自己的socket数据接收、解析逻辑，解析出对应指标的数值后，调用对应图表实例的push_new_data方法即可完成数据更新
• 需要新增图表时，只要新增一个RealtimeChart实例，注册回调、加到布局中即可，不需要修改其他逻辑
• 如果图表数量多、卡顿，可以把dcc.Graph的animate参数设置为False提升性能

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