如何用资源采集脚本的输出更新Bokeh机器监控时序图？

你现在的需求很明确：用Bokeh做一个时序图展示多台机器的CPU和内存使用率，已经有了数据拉取脚本，不想用文件读写这种比较笨重的方式，想找更高效的方案对吧？我给你整理几个实用的思路，结合你提供的示例代码来改就行：

1. 最直接：把数据拉取逻辑集成到Bokeh脚本里 #

如果你的数据拉取脚本只是一个简单的函数（不是独立运行的服务），那直接把拉取逻辑搬到Bokeh的更新函数里是最省事的，完全不用跨进程通信。比如把示例里的随机数替换成你的拉取函数返回值：

import numpy as np
from bokeh.plotting import figure, curdoc
from bokeh.driving import linear

# 替换成你实际的资源数据拉取函数
def fetch_system_stats():
    # 这里调用你的数据拉取逻辑，返回CPU和内存使用率（0-100的数值）
    cpu_usage = get_cpu_usage()  # 假设这是你原脚本里的拉取函数
    mem_usage = get_mem_usage()
    return cpu_usage, mem_usage

tools = 'pan'

@linear()
def update(step):
    # 用真实数据替换随机数
    cpu, mem = fetch_system_stats()
    ds1.data['x'].append(step)
    ds1.data['y'].append(cpu)
    ds2.data['x'].append(step)
    ds2.data['y'].append(mem)
    ds1.trigger('data', ds1.data, ds1.data)
    ds2.trigger('data', ds2.data, ds2.data)

p = figure(plot_width=400, plot_height=400)
r1 = p.line([], [], color="firebrick", line_width=2, legend='CPU使用率')
r2 = p.line([], [], color="navy", line_width=2, legend='内存使用率')

ds1 = r1.data_source
ds2 = r2.data_source

curdoc().add_root(p)
curdoc().add_periodic_callback(update, 500)

这种方案的好处是没有额外的开销，逻辑也清晰，适合拉取逻辑不复杂的场景。

2. 进程间通信（IPC）：用队列传递数据 #

如果你的数据拉取脚本是独立运行的进程（比如已经在后台持续跑着），那可以用Python的multiprocessing.Queue来做进程间的数据传递，比文件读写高效多了，还不会有IO阻塞和同步问题。

第一步：改造数据拉取脚本为生产者进程 #

把拉取逻辑封装成一个进程，持续往队列里推送数据：

from multiprocessing import Process, Queue
import time

def stats_producer(queue):
    while True:
        # 你的数据拉取逻辑
        cpu = get_cpu_usage()
        mem = get_mem_usage()
        queue.put((cpu, mem))  # 把数据放进队列
        time.sleep(0.5)  # 和Bokeh的更新频率保持一致

第二步：在Bokeh脚本里启动生产者并消费数据 #

import numpy as np
from bokeh.plotting import figure, curdoc
from bokeh.driving import linear
from multiprocessing import Queue, Process

# 导入你的数据拉取函数或者生产者进程
from stats_producer_script import stats_producer

# 创建队列并启动生产者进程
queue = Queue()
producer = Process(target=stats_producer, args=(queue,))
producer.daemon = True  # 让生产者进程随Bokeh主进程一起退出
producer.start()

tools = 'pan'

@linear()
def update(step):
    # 检查队列里有没有新数据
    if not queue.empty():
        cpu, mem = queue.get()
        ds1.data['x'].append(step)
        ds1.data['y'].append(cpu)
        ds2.data['x'].append(step)
        ds2.data['y'].append(mem)
        ds1.trigger('data', ds1.data, ds1.data)
        ds2.trigger('data', ds2.data, ds2.data)

p = figure(plot_width=400, plot_height=400)
r1 = p.line([], [], color="firebrick", line_width=2, legend='CPU使用率')
r2 = p.line([], [], color="navy", line_width=2, legend='内存使用率')

ds1 = r1.data_source
ds2 = r2.data_source

curdoc().add_root(p)
curdoc().add_periodic_callback(update, 500)

3. 共享内存：适合高频小数据传递 #

如果你的数据只是简单的数值（比如每次只传最新的CPU和内存值），还可以用multiprocessing的共享内存对象（Value或Array），这种方式的开销比队列还小，适合超高频率的数据更新场景。

比如：

生产者进程部分 #

from multiprocessing import Process, Value
import time

def stats_producer(cpu_val, mem_val):
    while True:
        cpu_val.value = get_cpu_usage()
        mem_val.value = get_mem_usage()
        time.sleep(0.5)

Bokeh脚本部分 #

import numpy as np
from bokeh.plotting import figure, curdoc
from bokeh.driving import linear
from multiprocessing import Value, Process
from stats_producer_script import stats_producer

# 创建共享内存变量
cpu_usage = Value('d', 0.0)  # 'd'表示双精度浮点数
mem_usage = Value('d', 0.0)

# 启动生产者进程
producer = Process(target=stats_producer, args=(cpu_usage, mem_usage))
producer.daemon = True
producer.start()

tools = 'pan'

@linear()
def update(step):
    # 直接读取共享内存里的最新值
    cpu = cpu_usage.value
    mem = mem_usage.value
    ds1.data['x'].append(step)
    ds1.data['y'].append(cpu)
    ds2.data['x'].append(step)
    ds2.data['y'].append(mem)
    ds1.trigger('data', ds1.data, ds1.data)
    ds2.trigger('data', ds2.data, ds2.data)

# 后面的图表初始化代码和之前一致...

对比一下几种方案 #

• 集成拉取逻辑：最简单，无额外开销，适合拉取逻辑不独立的场景。
• 队列IPC：灵活，支持复杂数据结构，适合拉取进程独立运行的情况。
• 共享内存：性能最高，适合高频小数据传递，缺点是只能存简单类型。

而你一开始想到的文件读写方案，不仅有IO性能开销，还需要处理文件锁定、读写同步的问题，确实不如上面这几种方案靠谱。

内容的提问来源于stack exchange，提问作者Mr. Fegur