Django+Plotly/Dash结合Celery/Redis实现异步绘图更新方案咨询

核心逻辑是将耗时计算完全剥离到Celery Worker执行，Dash端只负责任务触发、状态轮询和最终结果渲染，全程不阻塞用户请求，从根源上避开30秒请求超时问题。

第一步：编写Celery异步任务 #

不要直接在任务中返回Plotly Figure对象，Celery默认使用JSON序列化，原生Figure对象无法跨进程传输，只需要返回构建图表所需的结构化数据即可。
在worker目录下新建tasks.py写入任务逻辑：

# worker/tasks.py
import numpy as np
from celery import shared_task

@shared_task
def generate_scatter_data(seed):
    # 原scatter_plot函数中所有耗时的采集、计算、归一化逻辑全部迁移到这里
    np.random.seed(seed=seed)
    x = np.arange(1,1000,1).tolist() # 转为普通Python列表保证可序列化
    y = np.random.rand(1000).tolist()
    return {"x": x, "y": y}

提前在Django配置文件中完成Celery对接Redis的基础参数配置，确保Celery Worker进程可以正常启动、任务可以正常收发。

第二步：新增任务状态查询接口（可选） #

如果需要前后端分离的逻辑，可以新增两个接口分别负责任务触发和状态查询；如果全部逻辑走Dash服务端回调，这一步可以跳过，直接在回调里操作Celery即可。
修改views.py：

# worker/views.py
from django.shortcuts import render
from django.http import JsonResponse
from celery.result import AsyncResult
from .tasks import generate_scatter_data

def scatter_plot_dispatch(request):
    return render(request, 'worker/scatter_plot.html')

# 触发计算任务
def start_plot_task(request, seed):
    task = generate_scatter_data.delay(seed)
    return JsonResponse({"task_id": task.id})

# 查询任务执行状态和结果
def get_task_result(request, task_id):
    result = AsyncResult(task_id)
    resp = {
        "status": result.status,
        "data": None
    }
    if result.ready():
        resp["data"] = result.get()
    return JsonResponse(resp)

在urls.py中注册新接口路由：

# worker/urls.py
from django.urls import path
from . import views
from .dash_apps.test_apps import plot

urlpatterns = [
    path('scatterplot/', views.scatter_plot_dispatch, name="worker.scatter_plot"),
    path('api/start-task/<int:seed>/', views.start_plot_task, name="start_plot_task"),
    path('api/get-task-result/<str:task_id>/', views.get_task_result, name="get_task_result"),
]

第三步：改造Dash应用逻辑 #

在原有布局中加入任务ID存储组件、轮询触发器和状态提示区，把原来的同步回调拆成「触发任务+轮询状态+渲染结果」三个环节：用户选完种子后立刻返回加载提示，每隔1秒轮询一次任务状态，任务完成后自动渲染图表、停止轮询。
修改plot.py：

# worker/dash_apps/test_apps/plot.py
from dash import dcc, html, no_update
from dash.dependencies import Input, Output, State
from django_plotly_dash import DjangoDash
import plotly.graph_objects as go
from celery.result import AsyncResult
from .tasks import generate_scatter_data

external_stylesheets = ['https://codepen.io/chriddyp/pen/bWLwgP.css']
app = DjangoDash('ScatterPlot', external_stylesheets=external_stylesheets)

seeds = [11, 42, 97, 1001]

app.layout = lambda: html.Div([
    html.Div([
        html.Div([
            dcc.Dropdown(
                id='seed',
                options=[{'label': i, 'value': i} for i in seeds],
                placeholder='Select the seed...'
            )
        ],
        style={'width': '30%', 'display': 'inline-block'})
    ]), 
    # 存储当前任务ID的隐藏组件
    dcc.Store(id='current-task-id', data=None),
    # 轮询触发器，默认不启动，间隔1秒
    dcc.Interval(id='task-poll-interval', interval=1000, disabled=True),
    # 状态提示区
    html.Div(id='status-message', style={"margin": "20px 0", "color": "#666"}),
    dcc.Graph(id='Scatter-plot'),
])

# 回调1：用户选择种子后触发异步任务，显示加载提示
@app.callback(
    Output('current-task-id', 'data'),
    Output('task-poll-interval', 'disabled'),
    Output('status-message', 'children'),
    Output('Scatter-plot', 'figure'),
    Input('seed', 'value'),
    prevent_initial_call=True
)
def trigger_task(seed):
    if not seed:
        return no_update, True, "", go.Figure()
    # 直接提交Celery任务
    task = generate_scatter_data.delay(seed)
    # 立即返回提示，清空旧图表，启动轮询
    return (
        task.id,
        False,
        "图表生成预计耗时最长1分钟，请稍候",
        go.Figure()
    )

# 回调2：轮询任务状态，完成后更新图表
@app.callback(
    Output('Scatter-plot', 'figure', allow_duplicate=True),
    Output('status-message', 'children', allow_duplicate=True),
    Output('task-poll-interval', 'disabled', allow_duplicate=True),
    Input('task-poll-interval', 'n_intervals'),
    State('current-task-id', 'data'),
    prevent_initial_call=True
)
def poll_task_status(n, task_id):
    if not task_id:
        return no_update, no_update, True
    result = AsyncResult(task_id)
    if result.status == "PENDING":
        return no_update, "图表生成预计耗时最长1分钟，请稍候", False
    elif result.status == "SUCCESS":
        data = result.get()
        fig = go.Figure(data=go.Scatter(x=data["x"], y=data["y"], mode='markers'))
        return fig, "图表生成完成", True
    else:
        return no_update, "图表生成失败，请重新选择种子", True

模板文件scatter_plot.html不需要做任何修改，原有渲染逻辑可以直接复用。

Heroku部署注意事项 #

• 需要单独开一个worker dyno运行Celery进程，启动命令配置为celery -A djcel worker --loglevel=info
• 直接使用Heroku Redis插件提供的连接字符串，从环境变量读取配置即可，不要硬编码Redis地址
• 如果实际任务耗时普遍超过30秒，可以把轮询间隔调整为2-3秒，减少无效请求
• 相同参数的查询结果可以加Redis缓存，避免重复计算浪费资源

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