从多处理过程更新散景图答案

【问题标题】：Update Bokeh figure from multiprocessing Process从多处理过程更新散景图
【发布时间】：2016-12-22 02:27:55
【问题描述】：

我正在使用 Bokeh 服务器绘制来自传感器的在线数据。我编写了一个multiprocessing.Process 子类，它从传感器读取数据，然后通过管道将其馈送到另一个子类，该子类应该使用传入的数据更新散景图。

如何使绘图 Process 从管道中异步读取并将其绘制到 Bokeh？

考虑Grapher类，假设另一个进程已经向input_pipe发送数据：

from multiprocessing import Process

from bokeh.client import push_session
from bokeh.models import ColumnDataSource
from bokeh.plotting import curdoc, figure

from functools import partial

class Grapher(Process):
  def __init__(self, name, input_pipe):
    super(Grapher, self).__init__(name=name, daemon=True)
    self.input_pipe = input_pipe

    self.doc = curdoc()
    self.source = ColumnDataSource(dict(time=[], value=[])
    self.fig = figure()
    self.fig.line(source=self.source, x='time', y='value')
    self.doc.add_root(self.fig)
    self.session = push_session(self.doc)  # To keep session updated.

  def run(self):
    while True:
      time, value = self.input_pipe.recv()
      self.doc.add_next_tick_callback(partial(self.update, time, value))

  @gen.coroutine
  def update(self, time, value):
    self.source.stream(dict(time=[time], value=[value]))

此处的代码取自 Bokeh 文档中的 example。另外我运行bokeh serve 以便curdoc() 有一些东西可以连接。从 Bokeh 服务器日志中，我看到该连接存在。

但是，问题似乎在于update 在此处的while 循环的下一个滴答声之后并没有真正执行。我可以通过在update 中添加一条日志消息来检查这一点，该消息永远不会被打印出来。

如果我尝试将add_next_tick_callback 更改为仅update 调用，则该函数确实已运行，但会话不会绘制任何内容。

什么可能导致这里的问题？代码看起来合乎逻辑，我找不到使用这种方法可能失败的文档。

谢谢。

【问题讨论】：

标签： python multiprocessing bokeh

【解决方案1】：

当对应用程序使用bokeh.client 方法时，如果您希望为事件提供服务，并在它们发生时调用回调，那么您必须在最后调用阻塞函数session.loop_until_closed()。这就是保持监视事件和调用回调的东西，实际上是留在那里，以便监视事件和调用回调。

如果调用阻塞函数有问题，并且您不想使用 bokeh serve app.py 风格的应用程序，那么您可以尝试在另一个线程中调用 session.loop_until_closed()（不是 100% 这会起作用），或者启动您的拥有 Tornado ioloop 并直接在其上搭载 Bokeh 服务器应用程序。该技术将在即将发布的0.12.4 版本中得到更好的展示，但您现在可能会发现此笔记本是一个有用的参考：

https://gist.github.com/bryevdv/ff84871fcd843aceea4f0197be9c57e0

请注意，该笔记本是概念证明，您需要 0.12.4 开发版本才能使其工作，并且某些使用细节可能会发生变化。

【讨论】：