Python Dash设置background=True后全局静态变量无法更新的解决方案咨询

我用Python Dash编写了一段代码，目标是通过点击按钮修改类静态变量Global.test的值，同时给回调设置background=True启用后台运行属性。但测试发现，设置background=True后，Global.test的值无法成功更新：update_clicks函数中将Global.test设为5，但点击测试按钮触发btn_test回调时，返回的Global.test值仍为None；移除background=True则能正常返回5。请问如何在保留background=True属性的同时实现Global.test值的修改？

原代码如下：

import os
import dash
from dash import DiskcacheManager, CeleryManager, Input, Output, html
from Global import *

class Global():
    test = None

if 'REDIS_URL' in os.environ:
    # 若环境变量中设置了REDIS_URL，则使用Redis和Celery
    from celery import Celery
    celery_app = Celery(__name__, broker=os.environ['REDIS_URL'], backend=os.environ['REDIS_URL'])
    background_callback_manager = CeleryManager(celery_app)

else:
    # 本地开发时为非生产应用使用Diskcache
    import diskcache
    cache = diskcache.Cache("./cache")
    background_callback_manager = DiskcacheManager(cache)

app = dash.Dash(__name__, background_callback_manager=background_callback_manager)

app.layout = html.Div(
    [
        html.Div([html.P(id="paragraph_id", children=["Button not clicked"])])],
        html.Button(id="button_id", children="Run Job!"),
        html.Button(id="button_test", children="Test Job!"),
        html.Div('test', id='div_test')
    ]
)

@app.callback(
    output=Output("paragraph_id", "children"),
    inputs=Input("button_id", "n_clicks"),
    background=True,
    running=[
        (Output("button_id", "disabled"), True, False),
    ],
    prevent_initial_call=True 
)
def update_clicks(n_clicks):
    Global.test = 5
    print('Run : ' + str(Global.test))    
    return [f"Clicked {n_clicks} times"]


@app.callback(
    Output('div_test', 'children'),
    Input('button_test', 'n_clicks'),
    prevent_initial_call=True 
)
def btn_test(n):    
    print('test : ' + str(Global.test))
    return Global.test

if __name__ == "__main__":
    app.run_server(debug=True)

问题原因 #

• 后台回调（无论使用Celery还是Diskcache）运行在独立的进程/线程中，与主Dash应用进程不属于同一个内存空间。
• 你在update_clicks中修改的Global.test只是后台进程里的变量副本，主进程中的Global.test完全没有变化，因此btn_test（运行在主进程）读取的还是初始值None。
• 普通回调与主进程共享内存空间，所以修改全局变量能直接生效。

解决方案 #

方案1：利用后台回调管理器的缓存存储状态 #

通过Dash提供的background_callback_manager自带的缓存（Redis或Diskcache）来存储共享状态，不同进程都能访问缓存中的数据。

修改后的代码：

import os
import dash
from dash import DiskcacheManager, CeleryManager, Input, Output, html

class Global():
    test = None

if 'REDIS_URL' in os.environ:
    from celery import Celery
    celery_app = Celery(__name__, broker=os.environ['REDIS_URL'], backend=os.environ['REDIS_URL'])
    background_callback_manager = CeleryManager(celery_app)

else:
    import diskcache
    cache = diskcache.Cache("./cache")
    background_callback_manager = DiskcacheManager(cache)

app = dash.Dash(__name__, background_callback_manager=background_callback_manager)

app.layout = html.Div(
    [
        html.Div([html.P(id="paragraph_id", children=["Button not clicked"])]),
        html.Button(id="button_id", children="Run Job!"),
        html.Button(id="button_test", children="Test Job!"),
        html.Div('test', id='div_test')
    ]
)

@app.callback(
    output=Output("paragraph_id", "children"),
    inputs=Input("button_id", "n_clicks"),
    background=True,
    running=[
        (Output("button_id", "disabled"), True, False),
    ],
    prevent_initial_call=True 
)
def update_clicks(n_clicks):
    # 将值存入缓存，使用固定key标识
    background_callback_manager.cache.set('global_test', 5)
    print('Run : ' + str(background_callback_manager.cache.get('global_test')))    
    return [f"Clicked {n_clicks} times"]


@app.callback(
    Output('div_test', 'children'),
    Input('button_test', 'n_clicks'),
    prevent_initial_call=True 
)
def btn_test(n):    
    # 从缓存读取值，默认返回None
    test_val = background_callback_manager.cache.get('global_test') or None
    print('test : ' + str(test_val))
    return test_val

if __name__ == "__main__":
    app.run_server(debug=True)

方案2：使用Dash内置的dcc.Store组件存储状态 #

如果不需要跨多个进程共享状态（比如本地开发场景），可以用dcc.Store将状态存在前端浏览器中，回调通过读写Store来同步数据。

修改后的代码：

import os
import dash
from dash import DiskcacheManager, CeleryManager, Input, Output, html, dcc, State

class Global():
    test = None

if 'REDIS_URL' in os.environ:
    from celery import Celery
    celery_app = Celery(__name__, broker=os.environ['REDIS_URL'], backend=os.environ['REDIS_URL'])
    background_callback_manager = CeleryManager(celery_app)

else:
    import diskcache
    cache = diskcache.Cache("./cache")
    background_callback_manager = DiskcacheManager(cache)

app = dash.Dash(__name__, background_callback_manager=background_callback_manager)

app.layout = html.Div(
    [
        # 新增dcc.Store组件存储全局状态
        dcc.Store(id='global_store', data={'test': None}),
        html.Div([html.P(id="paragraph_id", children=["Button not clicked"])]),
        html.Button(id="button_id", children="Run Job!"),
        html.Button(id="button_test", children="Test Job!"),
        html.Div('test', id='div_test')
    ]
)

@app.callback(
    output=[Output("paragraph_id", "children"), Output('global_store', 'data')],
    inputs=Input("button_id", "n_clicks"),
    background=True,
    running=[
        (Output("button_id", "disabled"), True, False),
    ],
    prevent_initial_call=True 
)
def update_clicks(n_clicks):
    # 更新Store中的数据
    new_store_data = {'test': 5}
    print('Run : ' + str(new_store_data['test']))    
    return [f"Clicked {n_clicks} times"], new_store_data


@app.callback(
    Output('div_test', 'children'),
    Input('button_test', 'n_clicks'),
    State('global_store', 'data'),
    prevent_initial_call=True 
)
def btn_test(n, store_data):    
    test_val = store_data.get('test')
    print('test : ' + str(test_val))
    return test_val

if __name__ == "__main__":
    app.run_server(debug=True)

方案对比 #

• 方案1（缓存存储）：适合生产环境，支持跨多个worker进程共享状态，数据存储在后端缓存服务中。
• 方案2（dcc.Store）：适合简单场景或本地开发，数据存在前端浏览器，无需额外缓存服务，状态与前端会话绑定。

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