BLOGE 0.9.8-RC1 在线版 · 事实校验 2026-09-15 · English

附录 C —— Streaming

普通 graph 会在上游节点完全结束后才把输出交给下游。Streaming 改变了这个契约：数据在上游仍在产出的过程中，通过 channel 逐条流向下游。

什么场景需要 Streaming

大多数 BLOGE graph 采用"批量完成"模式：每个节点执行完毕、输出被存储，之后下游节点才能看到该输出。当数据集有界且可以舒适地放进内存时，这种模式完全够用。

以下场景需要 streaming：

上游增量产出数据 —— 音频块、数据库游标分页、实时传感器读数——你希望一边收到一边处理。
数据集大到一次性物化很浪费 —— 你更希望通过有界缓冲区做流水线。
延迟比吞吐量更重要 —— 在剩余数据还在计算时就向用户展示部分结果。

如果以上都不适用，普通节点和 foreach 更简单，也更容易测试。如果你真正困难的不是数据量，而是生命周期 —— 多轮外部交互、命名状态或信号所有权 —— 那么应该继续阅读第 14 章到第 16 章，而不是优先选择 streaming 原语。

三种 Streaming 原语

BLOGE 提供三种产出流式输出的构造，各服务于不同模式。

1. `stream node` —— 流式源或变换

stream node 声明一个 operator，其输出通过内部 NodeChannel 逐条发射，而非作为单个值返回。

stream node audioCapture : AudioCapture {
  buffer = 64
}

Operator 按自身节奏向 channel 推送条目。下游节点可以实时消费 channel，也可以等待 channel 关闭后接收物化后的完整列表。

适用场景： 单个 operator 产出无界或长时运行的条目序列（音频采集、事件流、分页 API）。

2. `stream foreach` —— 遍历集合并流式输出

stream foreach 是 foreach 的流式变体。它遍历集合，每个元素处理完成后立即将结果发射到下游，而非等所有元素都处理完。

stream foreach processOrders : item in loadOrders.output.orders {
  buffer = 16
  node processItem : OrderProcessor {
    input {
      order = item
    }
  }
}

适用场景： 集合是有界的，但你希望下游节点在最后一个元素处理完之前就开始消费结果。

3. `stream loop` —— 轮询并流式输出

stream loop 是 loop 的流式变体。每次迭代的输出会立即发射到下游，循环持续到 until 满足或达到 max_iterations。

stream loop checkStatus {
  max_iterations = 100
  delay = 2s
  depends_on = [initMonitor]
  buffer = 8
  node pollStatus : StatusPoller {
    input {
      serviceId = initMonitor.output.serviceId
      iteration = loopIteration
    }
  }
  until pollStatus.output.status == "ready"
}

适用场景： 你在轮询或持续产出条目，希望立即转发每个结果，而不是收集所有迭代后再一次性交付。

`.stream` 与 `.output` —— 两种边类型

当下游节点引用流式节点的结果时，边的类型决定了数据何时以及如何到达。

语法	边类型	行为
`upstream.stream`	StreamEdge	下游实时接收条目，逐条到达。下游 operator 必须能处理增量输入。
`upstream.output`	DirectEdge	下游等待流关闭后，接收完整的物化 `List<T>`。适用于下游 operator 需要全部数据才能开始工作的场景。

如何选择

问自己：下游 operator 是否需要所有数据才能开始？

是 → 使用 .output。例如：报告生成器需要所有已处理的订单来计算汇总。
否 → 使用 .stream。例如：语音转文字节点在音频块到达时即开始转录。

voice pipeline 示例在同一个 graph 中展示了两种边：

/// 通过 StreamEdge 实时接收音频块
stream node speechToText : SpeechToText {
  input {
    audio = audioCapture.stream       // StreamEdge —— 实时转发
  }
  buffer = 16
}

/// 通过 DirectEdge 等待完整转录
node textAnalysis : TextAnalyzer {
  depends_on = [speechToText]
  input {
    transcript = speechToText.output   // DirectEdge —— 物化 List<T>
  }
}

缓冲区大小

每个流式构造都有 buffer 参数，用于设置内部 NodeChannel 环形缓冲区的容量。

stream node audioCapture : AudioCapture {
  buffer = 64    // 环形缓冲区最多持有 64 个条目
}

默认值： 16 个条目。

如何思考缓冲区大小

缓冲区解耦了生产者速率和消费者速率。当缓冲区满时，生产者会阻塞（背压），直到消费者追上来。

场景	建议
生产者远快于消费者	增大缓冲区以吸收突发流量；差距过大时背压仍然会启动。
生产者和消费者速率接近	默认值 16 通常够用。
内存受限或条目较大	减小缓冲区以限制内存使用；接受更频繁的背压暂停。
延迟敏感的流水线	保持缓冲区较小，避免条目排队时间过长。

没有万能公式——从默认值开始，测量，再调整。当可观测性监听器激活时，引擎会记录背压事件，调优变得很直观。

Back-pressure、取消与持久化边界

事件	契约	要测试什么
Buffer 已满	Producer 等待，不能静默丢 item	放慢 consumer，观察内存有界且最终继续
下游取消	取消向上游传播并关闭 channel	断言 producer 停止，取消后不再发射
Producer 失败	Channel 以失败关闭，物化 `.output` 不能伪装完整	断言下游失败与部分输出策略
进程崩溃	内存 channel 内容不是 durable checkpoint	重启并证明哪些 item 来自 source 或 store 重放

Streaming 与 durable execution 解决不同问题。有界 channel 控制运行进程里的实时流动，不会让途中 item 自动跨崩溃保存。需要 replay 时，必须说出 durable source offset 或 checkpoint，并定义重复处理。不要从 buffer 大小推导 exactly-once delivery。

完整示例

仓库中以下文件用真实场景演示了三种 streaming 原语。

文件	原语	展示内容
`streaming-batch.bloge`	`stream foreach`	流式处理客户订单；下游报告使用 `.output`（DirectEdge）物化全部结果。
`streaming-status-monitor.bloge`	`stream loop`	每 2 秒轮询服务状态，将每次状态更新流式转发到下游；服务就绪时提前退出循环。
`voice-pipeline.bloge`	`stream node`	音频采集 → 语音转文字 → 文本分析。在同一个 graph 中展示 `.stream`（实时）和 `.output`（物化）两种边。

打开这些文件，阅读文档注释，追踪数据如何从流式源经过缓冲区流向下游消费者。修改 buffer 值，观察背压行为的变化。

速查表

Diagram: appendix-c-streaming figure 1

常见错误

错误	为什么会犯	如何纠正
在 operator 需要全部数据时使用 `.stream`	习惯性地选择"更快的路径"	如果 operator 在看到所有条目之前无法产出输出，就使用 `.output`。
把缓冲区设得很大"以防万一"	想完全避免背压	过大的缓冲区会延迟背压信号并浪费内存。从默认值开始。
忘了 `stream foreach` 同时也在收集 `.output`	以为 streaming 意味着什么都不会物化	条目被流式转发的同时也在收集；`.output` 会在流关闭后提供完整列表。
不测试背压路径	只用小数据集测试，缓冲区从未被填满	在测试中用 `MockOperator.delaying(...)` 模拟慢消费者，验证 graph 在背压下的行为。

与主线章节的关系

Streaming 没有被安排在单独一章，因为它依赖多个概念：

节点和依赖 —— 第 2 章和第 3 章
批处理和迭代 —— 第 11 章引入了 foreach 和 loop；streaming 变体在此基础上增加了 channel。
测试 —— 第 18 章展示了如何对流式 graph 结果做断言。
可观测性 —— 第 19 章涵盖了记录背压事件的监听器。

本附录将 streaming 的心智模型集中在一处，方便你在需要增量数据流时随时查阅。

什么场景需要 Streaming​

三种 Streaming 原语​

1. stream node —— 流式源或变换​

2. stream foreach —— 遍历集合并流式输出​

3. stream loop —— 轮询并流式输出​

.stream 与 .output —— 两种边类型​

如何选择​

缓冲区大小​

如何思考缓冲区大小​

Back-pressure、取消与持久化边界​

完整示例​

速查表​

常见错误​

与主线章节的关系​