使用反应式数据流管理复杂应用状态

# 使用响应式数据流管理复杂应用状态 来源：https://yogthos.net/posts/2026-09-12-reactive-dataflow.html 在小型应用中，响应式 UI 表面看起来很容易实现，无需太多努力即可保持同步。但随着应用规模增长并积累实际业务逻辑，问题就开始显现。通常最终会形成一系列依赖派生值的级联规则。此外，部分数据需要流向外部服务，同时外部数据又不断流入应用内部。要确保所有数据在用户持续点击界面、输入数据时保持一致性并非易事，任何构建过此类应用的开发者都深有体会。 ## 四大构建模块 好消息是我们可以使用四个构建模块来分解问题。Datastar (https://data-star.dev/) 和 glimmer (https://github.com/jolt-lang/glimmer) 提供了创建响应式 UI 的简便方式，使其能响应数据变化。Domino (https://github.com/domino-clj/domino) 提供事务性数据流引擎，可编码所有业务逻辑。Ebb (https://github.com/jlt-commons/ebb) 提供了协调系统内外数据流的清晰方式。 这些组件恰好能巧妙地结合在一起。Ebb 位于边缘层，协调进入系统的外部事件。这些事件在 Domino 中进行事务处理，计算所有派生值，随后由 glimmer 响应式原子根据最终状态驱动 UI 更新。用户输入则反向流动，从 UI 进入 Domino，经过事务处理后触发效果，再通过 Ebb 流出系统。 ## 边缘层的 Ebb Ebb 最终充当服务总线角色，可访问数据库、外部 API 等外部资源，或对接收邮件、生成 PDF 等应用所需功能。这些都是你希望与业务逻辑保持距离的应用边缘数据流。 它是 Missionary (https://github.com/leonoel/missionary) JVM 库的移植版本，该库严重依赖 Java 生态系统处理重负载。虽然这导致无法直接使用 Missionary，但 Jolt 纤程在概念上与 Missionary 的工作方式高度契合。Ebb 在 Jolt 纤程基础上用纯 Clojure 实现了 Missionary 的 API，并通过了 Missionary 的测试套件。拥有真正的纤程甚至在某方面优于原版：Missionary 的 `?` 操作符仅当语法上出现在进程主体内时才能挂起，因其依赖的协程转换是词法作用域的。而每个纤程拥有真实栈，因此在 Ebb 中函数可在任意调用深度挂起。 Missionary 背后的核心思想是提供一个用于监督式数据流编程的库，其中异步效应可被视为可组合值。它通过将任何分配资源的精确生命周期与其被消费者实际需要的时期绑定，解决了并发应用中协调时间和状态的问题。这通过有向无环图监督模型实现：共享依赖在首次请求时分配，在最终释放时销毁。 这种方法最大的优势在于消除了困扰响应式软件开发的内存泄漏和状态不一致问题。由于架构强制规定了异步事件流保持活跃的方式和时间，你永远不必担心孤立的 websocket 或僵尸线程消耗系统资源等问题。当组件卸载时，所有依赖资源都会递归清理，这为连续时间响应性奠定了数学上可靠的基础。 Missionary 设计的一个有趣方面是使用双向流协议，允许生产者和消费者进程协商背压，以便在昂贵的重新计算触发前使过时数据失效。文章末尾的仪表盘示例通过两个通道读取生产者，对比了处理背压的两种方式。 通道 A 使用 `m/observe` 订阅，通过读取线程将值推送给消费者。由于 `m/observe` 本身没有背压机制，需要配合 `m/relieve` 使用，在消费者滞后时保留最新值并丢弃其他值。取消流会运行 `cleanup` 函数，销毁子进程以确保不会遗留管道或 pid。 ``` (defn- observed-lines "A flow of producer lines, pushed from a reader thread." [k produced] (m/observe (fn [!] (let [proc (spawn-producer! k) rdr (io/reader (:out proc))] ;; a reader thread loops over (.readLine rdr), bumping produced ;; and pushing each line with (! line) (fn cleanup [] (kill! proc)))))) (defn- lane-a-flow [produced delivered] (let [source (m/relieve (fn [_ x] x) (observed-lines :a produced))] (m/ap (let [line (m/?> source)] ;; parking here lets the upstream ;; relieve collapse values (when (pos? @consumer-delay-ms) (m/? (m/sleep @consumer-delay-ms))) (swap! delivered inc) (parse-line line))))) ``` 通道 B 则按需求单位逐行拉取，当消费者减速时，操作系统管道会填满导致生产者停滞。在这种场景下，压力保持在源头，因此不会丢弃值。 ``` (defn- pulled-lines "A flow that reads one line per unit of demand. `m/via m/blk` moves the blocking read off the flow's thread; because nothing reads ahead, the pipe fills and the producer blocks in write(2)." [k] (let [proc (spawn-producer! k) rdr (io/reader (:out proc))] (m/ap (loop [] (if-let [line (m/? (m/via m/blk (.readLine rdr)))] (m/amb line (recur)) (m/amb)))))) ``` ## Domino Missionary 使用的数据流模型恰好完美适配 Domino，后者用于管理应用状态。文档数据模型位于 Domino 核心，跟踪应用数据的所有字段。业务逻辑通过将上下文无关函数附加到文档路径上作为规则来表达。当值在其声明为输入的路径上发生变化时触发规则，规则在事务中级联执行，产生文档的新状态。文档完成事务后，可触发效果将数据传递给 Ebb 管理的流层。 我倾向于将应用视为状态机，这确实是 Domino 设计背后的核心思想。触发一个事件（可以是用户输入、系统事件、服务调用等），并将其作为输入馈送到数据流引擎。规则级联触发，最终获得新状态。然后可以触发效果、更新 UI 等。 在演示仪表盘中我们可以看到具体体现。样本作为单个事务到达文档的 `[:sample]` 路径，触发一系列规则级联。事件声明为数据形式，每个事件明确说明其读写的路径，这允许计算关系图。 ``` (def events [{:id :record-history :inputs [:sample] :outputs [:history] ;; append the sample to the capped history series :handler ...} {:id :compute-stats :inputs [:history :window] :outputs [:stats] ;; stats over the last :window samples :handler ...} {:id :compute-pressure :inputs [:stats] :outputs [:pressure] :handler (fn [_ {:keys [stats]} _] {:pressure (+ (* 0.55 (get-in stats [:cpu :avg] 0.0)) (* 0.35 (get-in stats [:mem :last] 0.0)) (* 0.10 (get-in stats [:io-wait :last] 0.0)))})} {:id :classify-alert :inputs [:pressure :warn-threshold :crit-threshold] :outputs [:alert-level] :handler (fn [_ {:keys [pressure warn-threshold crit-threshold]} _] {:alert-level (cond (>= pressure (or crit-threshold 0.85)) :critical (>= pressure (or warn-threshold 0.55)) :warn :else :ok)})}]) ``` 事件向量也作为应用行为的规范，清晰声明了从原始样本到警报级别触发的每个业务规则。阈值和窗口连接到滑块，可拖动以重新运行相同的纯事件，而无需新样本到达。值得注意的一个微妙之处是：Domino 对每个变更的输入路径仅运行一次事件，这要求处理程序必须是幂等的。 Domino 提供了一个事务性数据流引擎来管理应用状态。输入进入，事务发生，输出产生。真正的好处在于能精确了解文档中所有字段之间的关系及其关联的业务规则。在我参与的大多数大型应用中，我发现这才是真正的业务难题。你最终会面对大量业务逻辑和许多派生字段，其关系的复杂性大到无法在脑海中完整把握。这时有人要求添加新业务规则，就根本无法保证它不会破坏系统中的其他规则。 税收和贷款是很好的例子。许多事项需要一起计算，而随着法律更新，公式会随时间变化，因此必须为每个场景维护清晰的规则集。当计算分散在代码库中时，无法轻易看出特定更改会影响什么，也难以证明更改后规则仍然一致。 另一个我有直接工作经验的场景是医院，患者数据需要在不同团队间协调。用于手术患者评估的应用需要在护士、外科医生、营养师和其他临床人员之间协调数据。最终会产生跟踪数百个不同字段的大型表单，这些字段用于计算术前评估分数。没有人能记住所有这些内容，每个字段都可能影响结果，因此确保分数正确派生至关重要。 ## 可重用规则和视图 Domino 的方法使业务逻辑可重用且可组合，因为规则函数和 UI 小部件都是上下文无关的。如果你编写计算 BMI 的公式，该公式将成为构建模块，可附加到代表身高和体重的两个字段以及 BMI 输出字段。表格小部件可以收集行信息，图表小部件可以附加到相同路径并渲染随时间变化的趋势。 Domino 还允许你创建附加到模式的视图，这些视图用于将文档中的字段映射到 UI。如果有护士和外科医生两个角色，他们可能关心文档数据的不同子集，且这些子集很可能重叠。能够附加具有各自小部件、命名约定和显示字段的不同视图，使得基于上下文以不同方式表达相同底层数据变得容易。由于视图仍通过整个文档的通用事务机制处理，无论字段是否出现在给定视图中，值都会被重新计算。护士可能正在收集患者的身高体重，而医生只关心最终的 BMI。由于护士无需看到 BMI 就能让其被计算，视图中显示的内容与需要触发的业务规则没有直接关联。无论是否向用户展示某项数据，业务逻辑都必须在整个文档中保持一致。 这种方法解决的另一个问题是并发多用户工作流。由于预先知道任何规则集影响的字段子图，当用户编辑属于该集合的字段时，可以锁定这些字段。不同用户可以安全地处理文档的不同部分，无需担心覆盖彼此数据。相关字段在用户编辑期间保持锁定，编辑完成后逻辑以事务方式应用。 ## UI 层 这留下了谜题的最后一块——界面本身。Glimmer 是一个响应式 GUI 工具包，你编写 Reagent 风格的组件，返回 hiccup 结构。它的唯一职责是在响应式状态变化时保持小部件树同步，glimmer-datastar (https://github.com/jolt-lang/glimmer-datastar) 在 glimmer 上实现了 Datastar 协议的服务端。页面保持一个开放的服务器发送事件流，每当状态变化时，服务器重新渲染片段并将其推送出去。浏览器保持为哑终端，所有业务逻辑都存在于服务器上。 在我参与的几乎所有大型应用中，我总发现需要将应用状态集中保存在某处。要么完全位于前端，后端作为服务总线；要么完全位于后端，客户端负责收集输入和显示 UI 小部件。将状态分割在两端意味着双方必须不断协商所有权，这成为微妙错误的根源。 在仪表盘中，权威的 Domino 上下文存在于一个原子中，以机器产生样本的速率写入数据。它再按计时器发布到响应式 glimmer 原子，SSE 流订阅的就是这个原子。这避免了页面以数据流入系统的速率重绘，也防止了行为异常的客户端回溯到数据摄入层。 ``` ;; the authoritative domino context is a plain atom (defonce ctx (atom nil)) ;; the single reactive cell the SSE renders subscribe to (defonce view (ratom/atom {:db nil :cascade [] :log []})) (defn publish! "Mirror the authoritative state into the view. One caller, on a timer." [] (ratom/reset! view {:db (db) :cascade (change-history) :log @log-entries})) ``` 页面本身随后就是该快照的函数。 ``` (defn fragment "The live region, rendered from one published snapshot." [live?] (let [{:keys [db cascade log]} @state/view {:keys [sample history stats alert pressure controls]} db] [:div#app-body (alert-banner alert) (gauge pressure (:warn controls) (:crit controls)) ;; metrics, the side panels, and the controls rail ...])) ``` ## 仪表盘示例 我创建了一个仪表盘 (https://github.com/jolt-lang/examples/tree/main/reactive-dashboard)，将所有理念融合在一起。它是一个实时系统监控器，渲染从 /proc 读取的 CPU、内存和网络数据，页面展示机器当前状态。 每一层位于自己的命名空间，划分遵循上述讨论的架构。底层是 app.pipeline，属于 Ebb 层，拥有可休眠、需要重试或可取消的流。app.state 由 Domino 层管理，位于数据流和 UI 之间。最后 app.ui 从发布快照渲染 hiccup 并传递给 Datastar。 管道并行运行三个摄入通道，每个通道对活跃生产者展示不同规范。通道 A 通过 m/observe 推入 m/relieve，因此生产者无需等待慢速消费者。通道 B 通过 m/via m/blk 按需求单位逐行拉取，确保在压力到达源头