Channel
查看 MarkdownChannel 是单个 Flow run 内部一条 durable 的 first-in, first-out (FIFO) 队列。Step 和 RPC 向它发布有类型的消息。Step 在 WaitFor 里等待这个 Channel，从而消费消息。消息不会跨 Flow run。
requests_queue.for_one() 什么时候用 Channel
Channel 适合 Step 要等待的事件，例如一次审批、用户回复或新的工作项。会被覆盖、多次读取的状态，用 Attribute。可以丢失的实时进度，用 Stream。
要等待并行分支完成，用 Join。Join 建在 ChannelMap 之上，它会替你数分支，并区分每一轮。见 等待并行 Step。
保证
- Dex 把每条消息和 Flow 一起 durable 存储。
- 消息保持到达顺序。
- 一条消息由一个 wait 消费。Dex 不会把它复制给其他等待中的 Step。
- Step 的 wait 消费了消息之后、Dex 保存它的 Execute 结果之前，Step 可能被取消。这时 Dex 把这些消息按原来的顺序放回 Channel 的最前面。之后的 wait 可以再消费它们。
- 被取消的 Execute 可能已经看到过放回的消息。所以一条消息可能两次到达 Execute。消息处理要做成幂等的。
- 如果 Step 是因为 Flow 关闭而被取消的，Dex 会丢弃这些消息，不再放回。StopFlow 请求、Flow timeout，以及用 ForceComplete 或 ForceFail 关闭 Flow 的决定，都属于这种情况。
- 其他取消都会放回消息，即使 Flow 之后关闭，例如在 GracefulComplete 之后。Flow 关闭后，不会再有 Step 消费它们。
- 一个 Step 方法取消某个 Step、同时发布消息时，被取消的 Step 不会收到这些消息。
- 多个 Step 等待同一个 Channel 时，最先开始等待的 Step 最先有机会拿到新消息。如果一个 Step 的 wait 需要的消息比待处理的多，它会被跳过，之后的 Step 可以拿走这些消息。
- 所以当其他 Step 一条一条地拿走消息时，等待多条消息的 Step 可能永远等不到。在同一个 Channel 上，不要同时有等待一条消息的 Step 和等待多条消息的 Step。
- 消息可以在 Step 开始等待之前到达。Dex 会保留它，直到之后的 wait 消费它。
- 只有发布消息的 Step 方法或 RPC 成功时，Dex 才会发布这条消息。见 Step 的保证。
定义、发布和等待
Channel 有稳定的名字和消息类型。先定义它，再加到 Flow 的 persistence schema。
在这个例子里，start Step 等待 ApprovalMessages Channel 上的一条消息，或者一个 Timer。publishApprovalMessage RPC 发布审批消息。
class ChannelWait(Step[int]):
def __init__(
self,
approval_messages: Channel[str],
queued_messages: Channel[str],
) -> None:
self.approval_messages = approval_messages
self.queued_messages = queued_messages
def get_step_options(self) -> StepOptions:
return StepOptions(execute_load_channels=(self.queued_messages,))
def wait_for(self, context: Context, input: int) -> Wait:
return Wait.any_of(
self.approval_messages.for_one(),
Timer.by_duration(timedelta(seconds=input)),
)
def execute(self, context: Context, input: int) -> StepDecision:
pending_queued_messages = self.queued_messages.pending_messages(context)
if pending_queued_messages:
self.queued_messages.delete(context, pending_queued_messages[0].message_id)
return graceful_complete(pending_queued_messages[0].value)
if context.has_timer_fired():
return graceful_complete("approval timed out")
approval_message_values = self.approval_messages.results(context)
return graceful_complete(approval_message_values[0])
class ChannelFlow(Flow[int]):
approval_messages = Channel("ApprovalMessages", str)
@rpc
def publish_approval_message(self, context: Context) -> None:
self.approval_messages.publish(context, "approved")
例子: examples/python/dex_examples/primitives/channel/channel_flow.py
如果第二个 Channel QueuedMessages 上有 RPC 放入的待处理消息，Execute 会先取走它，见管理待处理消息。Timer 分支在读取 Channel 结果之前返回。审批胜出时，Channel condition 已经消费了一条消息，所以第一个结果一定存在。完整示例还会在 Step options 中加载 QueuedMessages，在 persistence schema 中注册这些 Channel，并注册 RPC。
等待并消费
WaitFor 中的 Channel condition 会等到队列里有足够的消息。然后 Dex 按 FIFO 顺序移除选中的消息，作为 Channel results 传给 Execute。
| Condition | 何时继续 | 消费什么 |
|---|---|---|
| ForOne | 至少有一条消息在队列中。 | 正好一条消息。 |
| ForN | 队列中至少有请求的数量。 | 正好这个数量的消息。 |
| AtLeast | 队列中至少有请求的数量。 | 当前所有可用消息。 |
| AtMost | 立即继续，队列为空时也一样。 | 当前队列中最多请求数量的消息。 |
| AtLeastAtMost | 队列中至少有下限数量。 | 当前队列中从下限到上限数量的消息。 |
发布消息
在 Step 或 RPC 里，用 handler 的 Context 发布。Worker 之外的代码通过调用 Flow 的 RPC 来发布。这样校验、授权和相关的状态变更都留在 Flow 内。
管理待处理消息
Dex 接受每条消息时会给它一个 ID。Step 或 RPC 可以加载待处理消息，按 FIFO 顺序读取，并按 ID 删除其中一条。读取不会消费消息。handler 运行时，其他 Step 和 RPC 可能改变队列，所以加载到的列表可能已经过时。
在 Step 或非 transactional RPC 中，删除一条已经不在的消息不会有任何效果。transactional RPC 则会失败，Dex 不保存它的任何修改。从 dexcli 或 Dex Web 直接删除时，会返回 not-found 错误。见 Pending messages。
ChannelMap
一个 Flow 需要很多独立队列时，例如每个订单、客户或任务一条，用 ChannelMap。ChannelMap 有一个名字和一种消息类型。每个 instance key 有自己的 FIFO 队列。
参考
每个 SDK 参考的 Channel 页列出该语言的 API 名称。
Data primitive semantics 详细说明 wait condition、待处理消息、删除、状态加载和 ChannelMap key。