如何从独立进程控制Bokeh Server并实现图表动态替换与数据更新？

阿华AIGC实验室

2026-5-25

嘿，这个需求我之前落地过类似的场景，刚好给你一套完整的可执行方案——从外部进程控制Bokeh Server、动态替换图表、支持流式更新，再到嵌入Flask页面，一步步拆解：

1. 核心思路：进程间通信+Bokeh的IO循环调度

Bokeh Server本身是单线程事件循环模型，外部进程不能直接修改它的文档对象，得通过消息中间件传递指令，再让Bokeh Server在自己的IO循环里执行修改操作。我个人推荐用Redis Pub/Sub做消息通道，轻量易部署，跨进程/跨语言都能用。

2. 第一步：搭建可被外部控制的Bokeh Server

先安装依赖：

pip install bokeh redis flask

然后写核心的Bokeh控制脚本bokeh_controller.py：

from bokeh.server.server import Server
from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource
import redis
import threading
import json

# 初始化Redis客户端，用来做进程间通信
r = redis.Redis(host='localhost', port=6379, db=0)
pubsub = r.pubsub()
pubsub.subscribe('bokeh_control_channel')  # 订阅指定频道

# 全局变量存储当前图表和数据源，方便后续替换/更新
current_plot = None
source = ColumnDataSource(data={'x': [], 'y': []})

def update_plot(new_config):
    """根据外部传的配置完全替换当前图表"""
    global current_plot, source
    doc = curdoc()

    # 先移除旧图表
    if current_plot is not None:
        doc.remove_root(current_plot)

    # 根据配置生成新图表，支持多种类型
    plot_type = new_config.get('type', 'line')
    p = figure(
        title=new_config.get('title', '新图表'),
        x_axis_label=new_config.get('x_label', 'X轴'),
        y_axis_label=new_config.get('y_label', 'Y轴')
    )

    # 生成对应类型的图表
    if plot_type == 'line':
        p.line('x', 'y', source=source, line_width=2)
    elif plot_type == 'scatter':
        p.circle('x', 'y', source=source, size=10, color='orange')
    elif plot_type == 'bar':
        p.vbar(x='x', top='y', width=0.5, source=source, color='green')

    # 添加新图表到文档
    doc.add_root(p)
    current_plot = p

    # 如果有初始数据，直接加载
    if 'initial_data' in new_config:
        source.data = new_config['initial_data']

def listen_for_messages():
    """后台线程监听Redis消息，不阻塞Bokeh的IO循环"""
    for message in pubsub.listen():
        if message['type'] == 'message':
            try:
                payload = json.loads(message['data'].decode('utf-8'))
                # 区分是替换图表还是流式更新数据
                if payload.get('action') == 'replace_plot':
                    # 必须用add_next_tick_callback把操作放到Bokeh的IO循环里执行，线程安全！
                    curdoc().add_next_tick_callback(lambda: update_plot(payload['config']))
                elif payload.get('action') == 'stream_data':
                    curdoc().add_next_tick_callback(lambda: source.stream(payload['data']))
            except Exception as e:
                print(f"处理消息出错: {str(e)}")

def bokeh_app(doc):
    """Bokeh应用的入口函数"""
    # 启动后台监听线程，设为daemon自动随主进程退出
    threading.Thread(target=listen_for_messages, daemon=True).start()
    # 初始化一个默认图表
    update_plot({'type': 'line', 'title': '初始默认图表'})

if __name__ == '__main__':
    # 启动Bokeh Server，注意如果要嵌入Flask，需要允许Flask的域名访问WebSocket
    server = Server(
        {'/': bokeh_app},
        port=5006,
        allow_websocket_origin=['localhost:5000']  # Flask运行的端口
    )
    server.start()
    print("Bokeh Server运行在 http://localhost:5006/")
    server.io_loop.start()

3. 第二步：外部进程发送控制指令

写一个external_controller.py，模拟外部进程发送替换图表、流式更新的指令：

import redis
import json
import time

r = redis.Redis(host='localhost', port=6379, db=0)

# 1. 发送替换图表的指令：切换为散点图
replace_msg = {
    'action': 'replace_plot',
    'config': {
        'type': 'scatter',
        'title': '外部控制的散点图',
        'x_label': '时间点',
        'y_label': '数值',
        'initial_data': {'x': [1,2,3,4,5], 'y': [12,25,18,30,22]}
    }
}
r.publish('bokeh_control_channel', json.dumps(replace_msg))
print("已发送替换图表指令")

# 2. 模拟流式更新数据，每隔1秒发一条
time.sleep(2)
for i in range(6, 11):
    stream_msg = {
        'action': 'stream_data',
        'data': {'x': [i], 'y': [i*2 + 8]}
    }
    r.publish('bokeh_control_channel', json.dumps(stream_msg))
    time.sleep(1)
print("流式更新完成")

4. 第三步：将Bokeh图表嵌入Flask应用

先写Flask的主脚本flask_app.py：

from flask import Flask, render_template
from bokeh.embed import server_document

app = Flask(__name__)

@app.route('/')
def index():
    # 生成Bokeh Server的嵌入脚本，自动连接到Bokeh的WebSocket
    bokeh_script = server_document('http://localhost:5006/')
    return render_template('index.html', bokeh_script=bokeh_script)

if __name__ == '__main__':
    app.run(port=5000, debug=True)

然后在templates文件夹下创建index.html：

<!DOCTYPE html>
<html>
<head>
    <title>Flask + Bokeh Server</title>
</head>
<body>
    <h1>Flask中嵌入的Bokeh动态图表</h1>
    <!-- 渲染Bokeh的嵌入脚本 -->
    {{ bokeh_script|safe }}
</body>
</html>

5. 运行流程&关键注意事项

先启动Redis服务（本地默认端口6379）
启动Bokeh Server：python bokeh_controller.py
启动Flask应用：python flask_app.py，访问http://localhost:5000就能看到初始图表
运行外部控制脚本：python external_controller.py，就能看到Flask页面里的图表自动切换成散点图，然后开始流式更新数据

重点提醒：

线程安全：所有修改Bokeh文档的操作必须通过add_next_tick_callback放到Bokeh的IO循环中执行，绝对不能在后台线程直接修改文档对象，否则会导致崩溃或者UI不同步。
跨域问题：Bokeh Server启动时必须指定allow_websocket_origin，允许Flask的域名/端口访问，否则WebSocket连接会失败。
消息中间件替换：如果不想用Redis，也可以用Python的multiprocessing.Pipe（仅限同机器的Python进程）或者ZeroMQ，但Redis是最通用的选择。

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