第九章：性能优化

前置知识：本章需要掌握 ColumnDataSource（第二章 2.2）、基础图表绘制（第三章）以及交互与 Bokeh Server 的基本使用（分别见第六章和第七章）。

当处理大数据集或复杂可视化时，性能优化变得至关重要。
不过，Bokeh 性能问题很少只来自“渲染慢”这一件事，它通常可能出现在：

• Python 端数据处理过慢
• 发送到浏览器的数据量过大
• 浏览器首次渲染或交互重绘成本过高
• Python 回调过于频繁
• Notebook / Bokeh Server 的运行方式不匹配

因此，本章不把“开启 WebGL”当作默认答案，而是按更稳妥的决策顺序来组织内容：

1. 先识别瓶颈到底发生在哪一层
2. 优先减少数据量和传输量
3. 再考虑增量更新、聚合和降采样
4. 最后再评估 WebGL 或更高阶工具（如 Datashader）

交叉引用：

• 如果你主要在处理 ColumnDataSource 的更新方式，请先回看第二章 2.3。
• 如果你遇到的是滑块、筛选器、实时监控导致的卡顿，请结合第六章和第七章一起看。
• 如果你还不确定图表应该如何输出或嵌入，请先回看第八章，因为 standalone HTML、Notebook、Bokeh Server 的性能瓶颈并不完全相同。

9.1 优化前的决策顺序

在真正开始“优化”之前，建议先按下面的顺序判断：

第一步：先问自己，慢在哪里？

第二步：按这个优先级优化

1. 减少数据量

   • 只传当前真正需要显示的数据
   • 先聚合、再绘图
   • 先降采样、再考虑样式

2. 减少更新量

   • 用 stream() / patch() 替代整表替换
   • 避免每次交互都重算和重传全部数据

3. 减少绘制复杂度

   • 简化 glyph
   • 减少透明叠加、文本、图例和复杂工具
   • 拆成多个联动图，而不是一张图塞全部信息

4. 最后再尝试 WebGL

   • WebGL 更适合大量散点、折线等场景
   • 不是所有 glyph 和场景都一定更快
   • 不是“性能问题的一键修复”

5. 数据规模再上一个数量级时，考虑更高阶方案

   • Datashader
   • 服务端聚合
   • 预计算结果
   • 分层加载

第三步：确认你的运行模式

性能优化必须结合运行模式来判断：

这一点和第八章中的输出方式选择直接相关。

9.2 性能分析方法

识别性能瓶颈

在优化之前，首先需要识别性能瓶颈所在：

import time
from bokeh.plotting import figure, show

def measure_performance(func):
    """性能测量装饰器"""
    def wrapper(*args, **kwargs):
        start_time = time.time()
        result = func(*args, **kwargs)
        end_time = time.time()
        print(f"{func.__name__} 执行时间: {end_time - start_time:.4f}秒")
        return result
    return wrapper

@measure_performance
def create_large_plot(n_points=100000):
    """创建大数据集图表"""
    import numpy as np
    x = np.random.randn(n_points)
    y = np.random.randn(n_points)
    
    p = figure(title=f"{n_points}个数据点")
    p.circle(x, y, size=2)
    return p

# 测试性能
p = create_large_plot(100000)

性能监控工具

import psutil
import os
from functools import wraps

def monitor_resources(func):
    """监控资源使用"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        process = psutil.Process(os.getpid())
        
        # 记录开始状态
        start_memory = process.memory_info().rss / 1024 / 1024  # MB
        start_time = time.time()
        
        # 执行函数
        result = func(*args, **kwargs)
        
        # 记录结束状态
        end_memory = process.memory_info().rss / 1024 / 1024
        end_time = time.time()
        
        print(f"执行时间: {end_time - start_time:.2f}秒")
        print(f"内存使用: {end_memory:.2f}MB (增加: {end_memory - start_memory:.2f}MB)")
        
        return result
    return wrapper

@monitor_resources
def process_data(n_points):
    """处理大数据"""
    import numpy as np
    data = np.random.randn(n_points, 10)
    return data.sum(axis=1)

# 使用
result = process_data(1000000)

9.3 渲染性能优化

WebGL 渲染

WebGL 是浏览器中的硬件加速图形后端，在部分场景下可以明显提升性能，尤其常见于：

• 大量散点
• 大量折线段
• 浏览器端绘制压力明显大于 Python 端处理压力

但要注意，WebGL 不是默认最优解：

• 它并不替代降采样、聚合和数据裁剪
• 某些 glyph、文本、导出或交互行为不一定更理想
• 兼容性、视觉效果和调试体验也可能与 Canvas 不同

如果你还没有先做“减少点数 / 减少列数 / 减少重绘范围”，建议先完成这些步骤，再尝试 WebGL。

from bokeh.plotting import figure
import numpy as np

# 创建大数据集
n = 100000
x = np.random.randn(n)
y = np.random.randn(n)

# 使用Canvas渲染（默认）
p_canvas = figure(title="Canvas渲染", width=600, height=400,
    output_backend="canvas")
p_canvas.circle(x, y, size=2, alpha=0.1)

# 使用WebGL渲染（适合大数据集）
p_webgl = figure(title="WebGL渲染", width=600, height=400,
    output_backend="webgl")
p_webgl.circle(x, y, size=2, alpha=0.1)

# 注意：这里只是说明 WebGL 的典型用法，不代表它总是更快
show(p_webgl)

Canvas vs WebGL 对比

渲染优化最佳实践

from bokeh.plotting import figure
from bokeh.models import ColumnDataSource

# 1. 减少不必要的视觉元素
p = figure(
    title="优化图表",
    width=800,
    height=400,
    tools="pan,wheel_zoom,reset",  # 只保留必要工具
    toolbar_location=None          # 隐藏工具栏
)

# 2. 使用简单的标记
source = ColumnDataSource(data={
    'x': range(10000),
    'y': [i**0.5 for i in range(10000)]
})

# 简单的圆形标记通常比复杂标记更快
p.circle(
    'x', 'y', source=source,
    size=2,
    alpha=0.5,
    color="navy"
)

# 3. 避免不必要的交互
p.toolbar.autohide = True

9.5 增量更新优先于全量替换

在 Bokeh 中，很多“越交互越卡”的问题，不是因为数据绝对太大，而是因为每次都在完整替换 source.data。

什么时候优先考虑增量更新？

• 实时监控面板
• 追加式时间序列
• 只改局部值的控制面板
• Bokeh Server 周期回调

推荐方式

from bokeh.models import ColumnDataSource

source = ColumnDataSource(data={'x': [1, 2, 3], 'y': [2, 4, 6]})

# 追加新点
source.stream({'x': [4], 'y': [8]}, rollover=200)

# 只更新局部值
source.patch({'y': [(1, 5)]})

不推荐方式

# 每次都整体替换，尤其在高频回调里成本更高
source.data = {
    'x': new_x,
    'y': new_y,
}

当然，整体替换不是“错误”，而是要看场景：

• 数据结构整体发生变化：可以整体替换
• 只是追加几个点：优先 stream()
• 只是修改个别位置：优先 patch()

交叉引用：ColumnDataSource 的更新方式见第二章 2.3.3，实时回调见第七章 7.5。

9.6 Bokeh 特有的性能瓶颈

除了普通 Python 程序常见的 CPU / 内存问题，Bokeh 还有几类很典型的性能瓶颈：

1. 文档体积过大

症状：

• 页面首次打开很慢
• Notebook 单元格输出很大
• Bokeh Server 初次加载耗时长

常见原因：

• 传了太多列
• 传了太多历史数据
• 一个页面里堆了太多图和控件
• 每个图都带了完整 tooltip 字段

2. 浏览器重绘成本高

症状：

• 图表能打开，但缩放、平移、刷选卡顿
• 选中高亮后页面明显延迟

常见原因：

• 点数太多
• 文本、透明叠加、复杂 marker 太多
• 多图联动同时重绘

3. Bokeh Server 回调频率过高

症状：

• 滑块拖动时明显掉帧
• 实时面板越跑越慢
• 服务器 CPU 持续升高

常见原因：

• 每次 on_change() 都做重计算
• add_periodic_callback() 频率过高
• 每次回调都发送整份数据到前端

4. 输出模式和需求不匹配

症状：

• 只是展示数据，却用了 Server
• 只是简单筛选，却写了复杂 Python 回调
• notebook 和 standalone、server 混用后调试困难

建议：

• 纯前端交互优先 CustomJS
• 需要 Python 逻辑再使用 Bokeh Server
• 需要分享静态交互页面时优先 standalone HTML

9.7 实用排查清单

优化时建议按下面顺序排查，而不是一上来就改后端或启用 WebGL：

• 当前图表真的需要显示这么多点吗？
• 能否先聚合、分箱或降采样？
• ColumnDataSource 是否只保留了必要列？
• 是否在高频回调里反复整体替换 source.data？
• 是否给每张图都配置了过多工具、图例和悬停字段？
• 是否误把应该用 CustomJS 的前端交互写成了 Python 回调？
• 当前瓶颈到底在 Python、网络传输，还是浏览器渲染？
• 是否真的有必要启用 WebGL？
• 数据量是否已经到了应该考虑 Datashader 的级别？

9.8 本章小结

本章的核心不是“怎么把所有性能技巧都用上”，而是学会按正确顺序优化：

1. 先识别瓶颈
2. 先减少数据量，再减少更新量
3. 优先聚合、降采样、增量更新
4. 再考虑简化绘制和 WebGL
5. 超大规模数据再考虑 Datashader 或更高层方案

如果你接下来主要要优化的是：

• 交互回调卡顿：优先回看第六章
• Bokeh Server 实时更新：优先回看第七章
• 输出方式导致的问题：优先回看第八章

9.9 常见坑

• 看到卡顿就立刻启用 WebGL，而没有先减少数据量
• 在 Bokeh Server 中高频整体替换 source.data
• 在 tooltip 中塞入大量其实不需要显示的字段
• 用散点图硬画百万级点，而不是先聚合或栅格化
• 混淆 standalone HTML、Notebook、Bokeh Server 三种运行模式
• 只测 Python 执行时间，却忽略浏览器渲染和网络传输成本

更实用的经验顺序通常是：

1. 先减少点数
2. 再简化视觉编码
3. 再减少工具和联动
4. 最后再试 output_backend="webgl"

如果你是在做实时监控或 Bokeh Server 应用，还要结合第七章中的 stream()、patch() 和回调频率一起优化。

9.4 数据优化策略

数据降采样

当数据量过大时，降采样通常比“直接把所有点都画出来”更有效。
对于趋势图、监控图、长时间序列，降采样往往是第一选择，而不是 WebGL。

交叉引用：如果你这里处理的是实时更新数据，而不是一次性大数据，请先回看第二章 2.3.3中的 stream() / patch()，以及第七章 7.5 的实时更新模式。

当数据量过大时，降采样可以显著提升性能：

import numpy as np
from bokeh.plotting import figure, show
from bokeh.models import ColumnDataSource

def downsample_data(x, y, max_points=5000):
    """
    降采样数据，保留数据特征
    
    参数:
        x, y: 原始数据
        max_points: 最大点数
    返回:
        降采样后的数据
    """
    if len(x) <= max_points:
        return x, y
    
    # 计算采样步长
    step = len(x) // max_points
    
    # 均匀采样
    indices = np.arange(0, len(x), step)[:max_points]
    return x[indices], y[indices]

# 创建大数据集
n_points = 1000000
x_original = np.linspace(0, 10 * np.pi, n_points)
y_original = np.sin(x_original) + np.random.randn(n_points) * 0.1

# 降采样
x_sampled, y_sampled = downsample_data(x_original, y_original, max_points=5000)

print(f"原始数据: {len(x_original)} 点")
print(f"降采样后: {len(x_sampled)} 点")

# 绘制降采样后的数据
p = figure(title="降采样示例", width=800, height=400)
p.line(x_sampled, y_sampled, line_width=1)
show(p)

智能降采样算法

def lttb_downsample(x, y, target_points):
    """
    Largest Triangle Three Buckets (LTTB) 算法
    保持数据视觉特征的降采样方法
    """
    if len(x) <= target_points:
        return x, y
    
    # 始终保留第一个和最后一个点
    sampled_x = [x[0]]
    sampled_y = [y[0]]
    
    # 计算每个桶的大小
    bucket_size = (len(x) - 2) / (target_points - 2)
    
    for i in range(1, target_points - 1):
        # 当前桶的范围
        bucket_start = int((i - 1) * bucket_size) + 1
        bucket_end = int(i * bucket_size) + 1
        next_bucket_start = int(i * bucket_size) + 1
        next_bucket_end = int((i + 1) * bucket_size) + 1
        
        # 找到下一个桶中面积最大的点
        max_area = -1
        best_idx = next_bucket_start
        
        for j in range(next_bucket_start, min(next_bucket_end, len(x))):
            area = abs(
                (x[bucket_start] - x[j]) * (y[i-1] - y[bucket_start]) -
                (x[bucket_start] - x[i-1]) * (y[j] - y[bucket_start])
            )
            if area > max_area:
                max_area = area
                best_idx = j
        
        sampled_x.append(x[best_idx])
        sampled_y.append(y[best_idx])
    
    # 保留最后一个点
    sampled_x.append(x[-1])
    sampled_y.append(y[-1])
    
    return np.array(sampled_x), np.array(sampled_y)

# 使用LTTB降采样
x_lttb, y_lttb = lttb_downsample(x_original, y_original, 1000)
print(f"LTTB降采样: {len(x_lttb)} 点")

数据聚合

对于分类数据或矩阵数据，聚合通常比“直接把明细全画出来”更稳妥。
如果你发现散点已经重叠成一团，或者柱状图类别过多，优先考虑：

• 按组聚合
• 分箱统计
• 预先计算区间摘要
• 改用热力图、直方图、箱线图等更适合聚合表达的图形

这和第三章里的“图表选择”是一体两面：很多性能问题，本质上其实是图表选择问题。

对于分类数据，使用聚合可以减少数据点数量：

import pandas as pd
import numpy as np
from bokeh.plotting import figure, show
from bokeh.models import ColumnDataSource

# 创建大量数据
np.random.seed(42)
n = 100000
df = pd.DataFrame({
    'category': np.random.choice(['A', 'B', 'C', 'D', 'E'], n),
    'value': np.random.randn(n) * 100
})

# 聚合数据
aggregated = df.groupby('category')['value'].agg(['mean', 'std', 'count']).reset_index()
print("聚合后的数据:")
print(aggregated)

# 绘制聚合后的图表
source = ColumnDataSource(aggregated)
p = figure(x_range=list(aggregated['category']), title="聚合数据", height=400)
p.vbar(x='category', top='mean', source=source, width=0.5, color="navy")

show(p)

9.4 内存优化

内存使用监控

import psutil
import os
from bokeh.plotting import figure, show
from bokeh.models import ColumnDataSource
import numpy as np

def get_memory_usage():
    """获取当前内存使用量（MB）"""
    process = psutil.Process(os.getpid())
    return process.memory_info().rss / 1024 / 1024

print(f"初始内存: {get_memory_usage():.2f} MB")

# 创建大数据集
n = 1000000
x = np.random.randn(n)
y = np.random.randn(n)
print(f"创建数据后: {get_memory_usage():.2f} MB")

# 创建ColumnDataSource
source = ColumnDataSource(data={'x': x, 'y': y})
print(f"创建数据源后: {get_memory_usage():.2f} MB")

# 创建图表
p = figure(title="内存优化示例", width=800, height=400)
p.circle('x', 'y', source=source, size=1, alpha=0.1)
print(f"创建图表后: {get_memory_usage():.2f} MB")

内存优化技巧

import numpy as np
from bokeh.models import ColumnDataSource

# 技巧1：使用适当的数据类型
# 使用float32代替float64可以节省一半内存
x = np.random.randn(1000000).astype(np.float32)  # 节省内存
y = np.random.randn(1000000).astype(np.float32)

# 技巧2：避免不必要的数据复制
source = ColumnDataSource(data={'x': x, 'y': y})
# 不要这样做：source.data = {'x': x.copy(), 'y': y.copy()}

# 技巧3：及时释放不需要的数据
large_data = np.random.randn(10000000)
# 处理完后释放
del large_data

# 技巧4：使用流式更新而不是完全替换
source.stream({'x': [1.0], 'y': [2.0]})  # 追加数据
source.patch({'x': [(0, 1.5)]})  # 局部更新

内存泄漏检测

import gc
from bokeh.models import ColumnDataSource

def check_memory_leaks():
    """检查内存泄漏"""
    # 强制垃圾回收
    gc.collect()
    
    # 检查未引用的对象
    print(f"未回收的对象: {len(gc.garbage)}")
    
    # 检查数据源引用
    sources = [obj for obj in gc.get_objects() if isinstance(obj, ColumnDataSource)]
    print(f"活跃的ColumnDataSource数量: {len(sources)}")

# 定期检查
check_memory_leaks()

9.5 网络优化

数据压缩

from bokeh.plotting import figure, show
from bokeh.models import ColumnDataSource
import numpy as np

# 压缩大数据集
def create_compressed_source(x, y, compression_ratio=10):
    """创建压缩的数据源"""
    if len(x) > compression_ratio:
        step = len(x) // compression_ratio
        x_compressed = x[::step]
        y_compressed = y[::step]
    else:
        x_compressed = x
        y_compressed = y
    
    return ColumnDataSource(data={'x': x_compressed, 'y': y_compressed})

# 使用压缩数据源
n = 100000
x = np.linspace(0, 10, n)
y = np.sin(x)

source = create_compressed_source(x, y, compression_ratio=1000)
p = figure(title="压缩数据", width=800, height=400)
p.line('x', 'y', source=source)
show(p)

增量更新优化

from bokeh.plotting import figure, curdoc
from bokeh.models import ColumnDataSource
import numpy as np

# 创建数据源
source = ColumnDataSource(data={'x': [], 'y': []})
p = figure(title="实时数据", width=800, height=400)
p.line('x', 'y', source=source)

# 使用增量更新而不是完全替换
def update_data():
    """增量更新数据"""
    # 只添加新数据
    new_x = [len(source.data['x'])]
    new_y = [np.random.randn()]
    
    # 使用stream追加数据（高效）
    source.stream({'x': new_x, 'y': new_y}, rollover=100)  # 只保留最近100个点

# 定时更新
curdoc().add_periodic_callback(update_data, 100)  # 每100ms更新
curdoc().add_root(p)

缓存策略

from functools import lru_cache
import numpy as np

@lru_cache(maxsize=32)
def compute_expensive_operation(n_points):
    """缓存耗时的计算"""
    print(f"计算 {n_points} 个点...")
    x = np.linspace(0, 10 * np.pi, n_points)
    y = np.sin(x) * np.exp(-x / 10)
    return x, y

# 第一次调用会计算
x1, y1 = compute_expensive_operation(10000)

# 第二次调用使用缓存（瞬间返回）
x2, y2 = compute_expensive_operation(10000)

print("缓存生效：第二次调用无需重新计算")

9.6 性能检查清单

渲染性能检查

• 使用WebGL渲染大数据集（>10k点）
• 限制同时显示的数据点数量
• 减少不必要的视觉元素（网格线、标签等）
• 使用简单的标记形状（圆形比复杂形状快）
• 设置适当的透明度（alpha < 1 可能影响性能）

内存使用检查

• 使用适当的数据类型（float32 vs float64）
• 及时释放不需要的数据
• 使用流式更新（stream/patch）而不是完全替换
• 监控内存使用，防止内存泄漏
• 设置数据保留限制（rollover参数）

网络传输检查

• 压缩大数据集后再传输
• 使用增量更新减少数据传输量
• 启用数据压缩（Bokeh Server自动处理）
• 优化数据序列化格式
• 使用CDN加载BokehJS

代码优化检查

• 避免在回调中执行耗时操作
• 使用异步处理长时间任务
• 缓存重复计算结果
• 优化数据结构和算法
• 减少不必要的对象创建

9.7 常见性能问题排查

问题1：图表渲染缓慢

症状：图表加载时间长，交互卡顿

可能原因：

1. 数据点过多
2. 未使用WebGL渲染
3. 视觉元素过于复杂

解决方案：

# 方案1：启用WebGL
p = figure(output_backend="webgl")

# 方案2：降采样数据
def downsample(x, y, max_points=10000):
    if len(x) > max_points:
        step = len(x) // max_points
        return x[::step], y[::step]
    return x, y

# 方案3：简化视觉元素
p.circle(x, y, size=2, alpha=0.1, color="navy")  # 简单样式

问题2：内存占用过高

症状：应用内存持续增长，最终崩溃

可能原因：

1. 数据未及时释放
2. 内存泄漏
3. 数据类型不当

解决方案：

# 方案1：使用适当的数据类型
import numpy as np
x = np.array(data, dtype=np.float32)  # 使用float32

# 方案2：及时释放数据
del large_array
import gc
gc.collect()

# 方案3：使用流式更新
source.stream(new_data, rollover=1000)  # 限制保留的数据量

问题3：交互响应延迟

症状：缩放、平移操作不流畅

可能原因：

1. 回调函数耗时过长
2. 数据更新频率过高
3. 网络延迟

解决方案：

# 方案1：优化回调函数
def update(attr, old, new):
    # 避免在回调中执行耗时操作
    # 使用异步处理或延迟执行
    pass

# 方案2：降低更新频率
curdoc().add_periodic_callback(update, 500)  # 500ms而不是100ms

# 方案3：使用防抖动
from bokeh.models import CustomJS
slider.js_on_change('value', CustomJS(code="""
    clearTimeout(this.timeout);
    this.timeout = setTimeout(() => {
        // 实际处理逻辑
    }, 300);
"""))

问题4：Bokeh Server响应慢

症状：服务器处理请求时间长

可能原因：

1. 并发用户过多
2. 服务器资源不足
3. 应用代码效率低

解决方案：

# 方案1：增加工作进程
bokeh serve app.py --num-procs=4

# 方案2：使用负载均衡
# 配置Nginx或其他负载均衡器

# 方案3：优化应用代码
# 使用性能分析工具识别瓶颈

9.8 性能基准测试

"""
Bokeh性能基准测试脚本
用于测量不同场景下的性能表现
"""
import time
import numpy as np
from bokeh.plotting import figure, show
from bokeh.models import ColumnDataSource

def benchmark_scatter(n_points, use_webgl=False):
    """散点图性能测试"""
    x = np.random.randn(n_points)
    y = np.random.randn(n_points)
    
    backend = "webgl" if use_webgl else "canvas"
    
    start = time.time()
    p = figure(title=f"散点图 {n_points}点 ({backend})", 
        width=800, height=400, output_backend=backend)
    p.circle(x, y, size=2, alpha=0.5)
    creation_time = time.time() - start
    
    print(f"{n_points}点散点图创建时间 ({backend}): {creation_time:.4f}秒")
    return p

def benchmark_line(n_points):
    """折线图性能测试"""
    x = np.linspace(0, 10 * np.pi, n_points)
    y = np.sin(x)
    
    start = time.time()
    p = figure(title=f"折线图 {n_points}点", width=800, height=400)
    p.line(x, y, line_width=1)
    creation_time = time.time() - start
    
    print(f"{n_points}点折线图创建时间: {creation_time:.4f}秒")
    return p

def run_benchmarks():
    """运行所有基准测试"""
    print("=== Bokeh 性能基准测试 ===\n")
    
    # 测试不同数据量
    test_sizes = [1000, 10000, 100000, 500000]
    
    for n in test_sizes:
        print(f"\n--- {n} 数据点 ---")
        
        # Canvas渲染
        benchmark_scatter(n, use_webgl=False)
        
        # WebGL渲染（如果数据量足够大）
        if n >= 10000:
            benchmark_scatter(n, use_webgl=True)
        
        # 折线图
        benchmark_line(n)

if __name__ == "__main__":
    run_benchmarks()

下一步：掌握了性能优化技巧后，下一章我们将通过实战案例来综合运用所学知识。

延伸阅读：WebGL 渲染也可在第三章和第四章 4.7中找到简要介绍。Bokeh Server 的实时数据更新见第七章。