Bokeh Server 回调机制详解bokeh.server.callbacks 中的会话回调与事件循环调度原理【免费下载链接】bokehInteractive Data Visualization in the browser, from Python项目地址: https://gitcode.com/GitHub_Trending/bo/bokehBokeh 服务端应用bokeh serve的动态行为——周期轮询数据、延迟执行任务、下一 tick 立即更新——都建立在bokeh.server.callbacks这一模块之上。本文以该模块的 API 参考页 callbacks.rst 及其实现源码 src/bokeh/server/callbacks.py 为主体完整梳理SessionCallback、NextTickCallback、PeriodicCallback、TimeoutCallback四个类的职责与构造参数并结合 src/bokeh/util/asyncio.py、src/bokeh/document/document.py、src/bokeh/server/session.py 讲清“从Document.add_periodic_callback到 asyncio 事件循环执行”的完整调用链帮助你在编写 Bokeh 服务端应用时正确注册、取消与调试调度回调。模块定位为 Document 与 Session 提供可关联的回调对象模块文档 对该模块的定义是提供一组表示“可关联到 Bokeh Document 与 Session 的回调代码”的类。模块的__all__导出了 4 个公开名称src/bokeh/server/callbacks.pySessionCallback回调对象基类NextTickCallback在下一次事件循环 “tick” 执行一次的回调PeriodicCallback按指定周期反复执行的回调TimeoutCallback等待指定毫秒后执行一次的回调。需要与前端区分这些是Python 服务端的调度对象与前端bokeh.models.callbacks.Callback即浏览器侧的CustomJS完全是两套机制。模块内部还定义了一个 Dev API 类型别名src/bokeh/server/callbacks.pyCallback Callable[[], None | Awaitable[None]]也就是说回调函数既可以是普通同步函数也可以是async协程函数——这一点直接决定了实现层必须处理“awaitable 结果”后文_invoke与_AsyncPeriodic分别处理了两种场景。四个回调类的结构与参数基类 SessionCallbackid 被包裹的 callableSessionCallbacksrc/bokeh/server/callbacks.py是所有会话回调的基类构造与属性如下成员类型/取值说明__init__(callback, *, callback_id)callback: Callable[[], None \| Awaitable[None]]callback_id: ID关键字参数callback_id是回调的唯一标识后续取消回调都靠它定位idproperty返回ID唯一 IDcallbackproperty返回Callback该回调对象实际包裹的可调用对象ID 从何而来在 src/bokeh/document/document.py 中可以看到Document在创建这三类回调对象时统一用make_id()生成 IDfrom ..server.callbacks import NextTickCallback cb NextTickCallback(callback_no_op_callback, callback_idmake_id())一个值得注意的细节SessionCallback对象构造时传入的callback是_no_op_callback真正的用户回调并不直接存在对象里而是在DocumentCallbackManager.add_session_callback中被重新赋值并包裹见 src/bokeh/document/callbacks.py 第 204 行的callback_obj._callback _wrap_with_curdoc(doc, actual_callback)。这意味着回调执行时自动处于“当前 Document 已设置”的上下文中用户回调里可以直接调用curdoc()。NextTickCallback / PeriodicCallback / TimeoutCallback三个子类都在基类之上增加了调度语义src/bokeh/server/callbacks.py类额外构造参数额外属性调度语义源码 docstringNextTickCallback无无在下一次事件循环 “tick” 执行一次PeriodicCallbackperiod: int毫秒period按指定周期毫秒在事件循环上反复执行TimeoutCallbacktimeout: int毫秒timeout等待指定毫秒毫秒后执行一次两者均为毫秒单位PeriodicCallback.period的 docstring 明确写着 “The period time (in milliseconds) that this callback should repeat execution at”TimeoutCallback.timeout同理。时间到秒的换算发生在底层调度处src/bokeh/util/asyncio.py 中call_later(timeout_milliseconds / 1000.0, wrapper)。DocumentCallbackGroup把回调对象分发到事件循环DocumentCallbackGroup位于 Dev API 区src/bokeh/server/callbacks.py它是“回调对象”与真正的 asyncio 调度器之间的适配器内部持有一个_CallbackGroup来自 src/bokeh/util/asyncio.py。核心方法是add_session_callback它按isinstance把三类对象分发到对应的调度入口if isinstance(callback_obj, PeriodicCallback): self._group.add_periodic_callback(callback_obj.callback, callback_obj.period, callback_obj.id) elif isinstance(callback_obj, TimeoutCallback): self._group.add_timeout_callback(callback_obj.callback, callback_obj.timeout, callback_obj.id) elif isinstance(callback_obj, NextTickCallback): self._group.add_next_tick_callback(callback_obj.callback, callback_obj.id) else: raise ValueError(...)对应关系一目了然PeriodicCallback → add_periodic_callback、TimeoutCallback → add_timeout_callback、NextTickCallback → add_next_tick_callback每个调度入口都接收回调本体、时间参数毫秒与callback_id。remove_session_callbacksrc/bokeh/server/callbacks.py做了同样三类分发但有一处关键的健壮性处理——源码注释解释了原因“we may be called multiple times because of multiple views on a document - the document has to notify that the callback was removed even if only one view invoked it. So we need to silently no-op if were already removed.”即当同一 Document 有多个浏览器视图多标签页时取消操作可能被触发多次因此这里用try/except ValueError: pass吞掉“重复移除”异常。remove_all_callbacks则转发给_CallbackGroup.remove_all_callbackssrc/bokeh/util/asyncio.py对三类回调 ID 逐一移除并同样静默忽略ValueError——测试test_remove_all_callbacks验证了“一次性移除全部后后续周期/超时/next-tick 回调都不再执行”tests/unit/bokeh/server/test_callbacks__server.py。用户侧入口Document 上的 add/remove API真正被应用代码调用的是Document的方法。三个 add 方法src/bokeh/document/document.py统一走“构造回调对象 → 交给DocumentCallbackManager.add_session_callback”的路径区别只在one_shot参数Document 方法创建的回调对象one_shot对应移除方法add_next_tick_callback(callback)→ 返回NextTickCallbackNextTickCallbackTrueremove_next_tick_callback(callback_obj)add_periodic_callback(callback, period_milliseconds)→ 返回PeriodicCallbackPeriodicCallbackFalseremove_periodic_callback(callback_obj)add_timeout_callback(callback, timeout_milliseconds)→ 返回TimeoutCallbackTimeoutCallbackTrueremove_timeout_callback(callback_obj)one_shotTrue的语义在 src/bokeh/document/callbacks.py 中实现管理器会给一次性回调套一层remove_then_invoke先把自己从_session_callbacks集合中移除、再执行用户函数。这解释了为什么 next-tick 与 timeout 回调“执行完即自毁”而周期回调需要用户显式取消。one_shot的自动移除与调度层的移除是两道独立防线_CallbackGroup.add_next_tick_callback的 wrapper 在触发时会先remove_next_tick_callback(callback_id)再执行src/bokeh/util/asyncio.py保证 removers 表不泄漏。测试test_next_tick_runs专门断言“循环结束后len(ctx.group._next_tick_callback_removers) 0”tests/unit/bokeh/server/test_callbacks__server.pytimeout 用例同理。移除方法remove_next_tick_callback/remove_periodic_callback/remove_timeout_callbacksrc/bokeh/document/document.py都委托给DocumentCallbackManager.remove_session_callback若回调从未添加、已执行或已移除会抛出ValueError“callback already ran or was already removed, cannot be removed again”src/bokeh/document/callbacks.py。测试对三种类型均验证了“第二次移除抛ValueError且消息包含 twice”tests/unit/bokeh/server/test_callbacks__server.py。每个 add/remove 还会产生SessionCallbackAdded/SessionCallbackRemoved两个 Document 变更事件定义在 src/bokeh/document/events.py它们是 Document 层与 Session 层之间的桥梁——下一节展开。三个 add 方法的 docstring 还带有一条重要使用限制以add_periodic_callback为例src/bokeh/document/document.py“Periodic callbacks only work within the context of a Bokeh server session. This function will no effect when Bokeh outputs to standalone HTML or Jupyter notebook cells.”即这三类回调只在bokeh serve的服务会话上下文中生效把应用输出为独立 HTML 或 notebook 单元格时它们不产生效果。从注册到执行完整调用链与文档锁把前面各层串起来一个doc.add_periodic_callback(tick, 500)的完整生命周期如下每步均给出源码位置Document 层Document.add_periodic_callbacksrc/bokeh/document/document.py创建PeriodicCallbackcallback_no_op_callbackcallback_idmake_id()调用self.callbacks.add_session_callback(cb, callback, one_shotFalse)。DocumentCallbackManager 层src/bokeh/document/callbacks.py校验 Document 未被销毁否则抛RuntimeError按one_shot决定是否套自动移除包装用_wrap_with_curdoc(doc, ...)包裹真实回调执行时自动 patchcurdoc()见 src/bokeh/document/callbacks.py加入_session_callbacks集合并trigger_on_change(SessionCallbackAdded(doc, callback_obj))发出事件。ServerSession 层服务端会话订阅了这两类事件。ServerSession._session_callback_addedsrc/bokeh/server/session.py调用_wrap_session_callback后再交给本模块的DocumentCallbackGroup.add_session_callbackdef _wrap_session_callback(self, callback: SessionCallback) - SessionCallback: wrapped copy(callback) wrapped._callback self._wrap_document_callback(callback.callback) return wrapped_wrap_document_callbacksrc/bokeh/server/session.py体现了 Bokeh 服务端的并发模型若回调函数带有nolock属性则以async方式丢进 executor 运行不持文档锁否则每次调用都先self.with_document_locked(callback, *args, **kwargs)持文档锁执行。这保证了回调修改source.data等模型属性时与来自其他连接的 patch 互斥避免并发写入 Document。 4.DocumentCallbackGroup → _CallbackGroup本模块 src/bokeh/server/callbacks.py按类型分发到 src/bokeh/util/asyncio.py 的_CallbackGroup。 5.asyncio 事件循环三种调度的底层实现各不相同next tickself._loop.call_soon_threadsafe(wrapper)src/bokeh/util/asyncio.py下一轮循环即触发call_soon_threadsafe使“从其他线程注册”也是线程安全的——测试test_adding_next_tick_from_another_thread用 1000 线程并发注册并断言全部执行tests/unit/bokeh/server/test_callbacks__server.pytimeoutstart()在事件循环内通过call_later(timeout_ms / 1000.0, wrapper)创建TimerHandlesrc/bokeh/util/asyncio.py移除时remover会取消该 handle因此“立即移除则不执行”测试test_timeout_does_not_run_if_removed_immediatelyperiodic创建一个_AsyncPeriodic任务src/bokeh/util/asyncio.py其_run循环为sleep(period) → 执行 → 记录耗时 → sleep(max(0, period - elapsed))。这一“补偿剩余时间”的设计使各轮执行不会重叠且周期以墙钟时间对齐而非“执行完后重新计时”异常在循环内被捕获并log.error单个 tick 失败不会杀死整个周期任务。回调执行与异常处理_CallbackGroup._invokesrc/bokeh/util/asyncio.py调用用户函数后若结果是 awaitable 则asyncio.ensure_future调度并挂_log_task_exception回调任务内未处理的异常会以 “Error thrown from callback:” 打印完整堆栈src/bokeh/util/asyncio.py而不会中断事件循环。取消方向对称doc.remove_periodic_callback(cb)→DocumentCallbackManager.remove_session_callback发出SessionCallbackRemovedsrc/bokeh/document/callbacks.py→ServerSession._session_callback_removedsrc/bokeh/server/session.py→DocumentCallbackGroup.remove_session_callback→_CallbackGroup中对应的 remover周期回调走_AsyncPeriodic.stop()置_stopped并call_soon_threadsafe(task.cancel)src/bokeh/util/asyncio.py。实战用法注册、取消与协程回调结合上述 API 语义典型的服务端动态更新写法如下curdoc()在 Bokeh 应用文件中返回当前 Document与 src/bokeh/document/document.py 的add_periodic_callback配合from bokeh.document import curdoc from bokeh.models import ColumnDataSource doc curdoc() source ColumnDataSource(data{t: [], v: []}) # 周期回调每 500ms 轮询一次毫秒 def tick(): new_v poll_remote_value() # 自定义数据获取 source.stream({t: [time.time()], v: [new_v]}, max_len1000) periodic doc.add_periodic_callback(tick, 500) # 延迟回调2 秒后执行一次 def warmup(): source.data[ready] [True] doc.add_timeout_callback(warmup, 2000) # 下一 tick 回调常用于“稍后再执行”的立即型更新 doc.add_next_tick_callback(lambda: print(on next loop tick))取消周期回调防止页面逻辑停止后定时器空转doc.remove_periodic_callback(periodic)由于Callback类型签名是Callable[[], None | Awaitable[None]]回调可以直接写成协程next-tick 与 timeout 场景由_invoke中的ensure_future承接周期场景由_AsyncPeriodic._run内await result承接src/bokeh/util/asyncio.pyasync def fetch_remote(): async with aiohttp.ClientSession() as s: # 任意三方 HTTP 库 data await s.get(url).json() source.data data doc.add_timeout_callback(fetch_remote, 1000)使用时需要牢记的三点均有源码/文档依据仅限服务会话三类回调在独立 HTML 或 notebook 输出下不生效docstringsrc/bokeh/document/document.py重复移除会报错Document 层remove_*_callback对“已运行/已移除”的回调抛ValueError业务代码若要容忍重复取消应自行捕获——DocumentCallbackGroup内部虽已静默处理多视图场景但DocumentCallbackManager.remove_session_callback在 src/bokeh/document/callbacks.py 仍会抛错周期执行有文档锁语义未标记nolock的回调在ServerSession._wrap_document_callback中持锁执行src/bokeh/server/session.py回调内避免做阻塞 I/O耗时工作建议放入协程回调或线程否则事件循环的其余 tick 会被推迟。测试视角的行为保证调度层的单元测试 tests/unit/bokeh/server/test_callbacks__server.py 直接驱动_CallbackGroup IOLoop为本模块的行为提供了可验证的边界三种回调均可正常触发且触发后 removers 表清空无泄漏断言第 80–98 行“立即移除则不执行”对三种类型均成立第 113–134 行同一函数可同时以三种类型注册互不干扰test_same_callback_as_all_three_types第 136–143 行重复添加/移除同一类型的不同 ID 合法同一 ID 移除两次抛ValueError第 145–210 行跨线程注册 next-tick 回调安全test_adding_next_tick_from_another_thread第 212–222 行remove_all_callbacks可一次性清空并阻止后续执行第 166–177 行。关键文件索引关注点文件本文对应的 API 参考页docs/bokeh/source/docs/reference/server/callbacks.rst四类回调对象 DocumentCallbackGroupsrc/bokeh/server/callbacks.pyasyncio 调度实现_CallbackGroup / _AsyncPeriodicsrc/bokeh/util/asyncio.pyDocument 的 add/remove 用户 APIsrc/bokeh/document/document.py回调注册/移除 事件SessionCallbackAdded/Removedsrc/bokeh/document/callbacks.py、src/bokeh/document/events.py服务端会话侧的回调包装与文档锁src/bokeh/server/session.py调度行为单元测试tests/unit/bokeh/server/test_callbacks__server.py综上bokeh.server.callbacks是 Bokeh 服务端“时间驱动”能力的核心四个轻量回调对象承载调度语义DocumentCallbackGroup负责类型分发事件机制连接 Document 与 ServerSession 两层最终由 src/bokeh/util/asyncio.py 中的_CallbackGroup落到 asyncio 事件循环上执行理解了这条链路就能正确实现并排查服务端应用中周期刷新、延迟任务与跨线程更新相关的各类问题。【免费下载链接】bokehInteractive Data Visualization in the browser, from Python项目地址: https://gitcode.com/GitHub_Trending/bo/bokeh创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考