--- title: "Channel" description: "Channel 是单个 Flow run 内部一条 durable 的 first-in, first-out (FIFO) 队列。Step 和 RPC 向它发布有类型的消息。Step 在 WaitFor 里等待这个 Channel，从而消费消息。消息不会跨 Flow run。" canonical_url: "https://docs.superdurable.io/zh-Hans/concepts/primitives/channel/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/concepts/primitives/channel.mdx" locale: "zh-Hans" --- # Channel **Channel** 是单个 Flow run 内部一条 durable 的 first-in, first-out (FIFO) 队列。Step 和 RPC 向它发布有类型的消息。Step 在 **WaitFor** 里等待这个 Channel，从而消费消息。消息不会跨 Flow run。 Flow run 拥有一个由其 Step 和 RPC 共享的 FIFO Channel。发布者添加消息，每条消息由一个等待中的消费 Step 消费。 ## 什么时候用 Channel Channel 适合 Step 要等待的事件，例如一次审批、用户回复或新的工作项。会被覆盖、多次读取的状态，用 [Attribute](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md)。可以丢失的实时进度，用 [Stream](https://docs.superdurable.io/zh-Hans/concepts/primitives/stream.md)。 要等待并行分支完成，用 **Join**。Join 建在 ChannelMap 之上，它会替你数分支，并区分每一轮。见 [等待并行 Step](https://docs.superdurable.io/zh-Hans/design-patterns/parallel/await.md)。 ## 保证 - 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 的保证](https://docs.superdurable.io/zh-Hans/concepts/step.md#guarantees)。 ## 定义、发布和等待 Channel 有稳定的名字和消息类型。先定义它，再加到 Flow 的 persistence schema。 在这个例子里，start Step 等待 **ApprovalMessages** Channel 上的一条消息，或者一个 Timer。**publishApprovalMessage** RPC 发布审批消息。 **Python** ```python 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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/channel/channel_flow.py) **Go** ```go var ( ApprovalMessages = dex.DefineChannel[string]("ApprovalMessages") QueuedMessages = dex.DefineChannel[string]("QueuedMessages") PrioritizedMessages = dex.DefineChannel[string]("PrioritizedMessages") ) func (channelWait) WaitFor(_ dex.Context, input int) (*dex.Wait, error) { return dex.AnyOf( ApprovalMessages.ForOne(), dex.Timer(time.Duration(input)*time.Second), ), nil } func (channelWait) Execute(ctx dex.Context, _ int) (*dex.StepDecision, error) { pendingQueuedMessages, err := QueuedMessages.PendingMessages(ctx) if err != nil { return nil, err } if len(pendingQueuedMessages) > 0 { if err := QueuedMessages.Delete(ctx, pendingQueuedMessages[0].MessageID); err != nil { return nil, err } return dex.GracefulComplete(pendingQueuedMessages[0].Value), nil } if ctx.HasTimerFired() { return dex.GracefulComplete("approval timed out"), nil } approvalMessageValues, err := ApprovalMessages.GetConditionResults(ctx) if err != nil { return nil, err } return dex.GracefulComplete(approvalMessageValues[0]), nil } func (*ChannelFlow) PublishApprovalMessage(ctx dex.Context, _ dex.None) (*dex.RPCResult[dex.None], error) { if err := ApprovalMessages.Publish(ctx, "approved"); err != nil { return nil, err } return &dex.RPCResult[dex.None]{}, nil } ``` 例子: [examples/go/primitives/channel/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/channel/workflow.go) **Java** ```java public final Channel approvalMessages = Channel.define("ApprovalMessages", String.class); @Override public Wait waitFor(final Context context, final Integer input) { return Wait.anyOf( approvalMessages.forOne(), Timer.byDuration(Duration.ofSeconds(input))); } @Override public StepDecision execute(final Context context, final Integer input) { final List> pendingQueuedMessages = queuedMessages.pendingMessages(context); if (!pendingQueuedMessages.isEmpty()) { queuedMessages.delete(context, pendingQueuedMessages.get(0).getMessageId()); return StepDecision.gracefulComplete(pendingQueuedMessages.get(0).getValue()); } if (context.hasTimerFired()) { return StepDecision.gracefulComplete("approval timed out"); } final List approvalMessageValues = approvalMessages.getConditionResults(context); return StepDecision.gracefulComplete(approvalMessageValues.get(0)); } @RPC public void publishApprovalMessage(final Context context) { approvalMessages.publish(context, "approved"); } ``` 例子: [examples/java/src/main/java/io/superdurable/dex/primitives/channel/ChannelFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/channel/ChannelFlow.java) **TypeScript** ```typescript const approvalMessages = new Channel("ApprovalMessages", stringCodec); public waitFor(_context: Context, input: number): Wait { return Wait.anyOf( approvalMessages.forOne(), Timer.byDuration(input * 1000), ); } public execute(context: Context, _input: number): StepDecision { const pendingQueuedMessages = queuedMessages.pendingMessages(context); if (pendingQueuedMessages.length > 0) { queuedMessages.delete(context, pendingQueuedMessages[0]!.messageId); return gracefulComplete(pendingQueuedMessages[0]!.value); } if (context.hasTimerFired()) { return gracefulComplete("approval timed out"); } const approvalMessageValues = approvalMessages.results(context); return gracefulComplete(approvalMessageValues[0]!); } @rpc() public publishApprovalMessage(context: Context): void { approvalMessages.publish(context, "approved"); } ``` 例子: [examples/typescript/src/primitives/channel/channel-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/channel/channel-flow.ts) **Rust** ```rust static APPROVAL_MESSAGES: LazyLock> = LazyLock::new(|| Channel::new("ApprovalMessages")); fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { Ok(Wait::any_of([ APPROVAL_MESSAGES.for_one(), Timer::by_duration(Duration::from_secs(input.max(0) as u64)), ])) } fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { let pending_queued_messages = QUEUED_MESSAGES.pending_messages(context)?; if let Some(queued_message) = pending_queued_messages.first() { QUEUED_MESSAGES.delete(context, &queued_message.message_id)?; return Ok(StepDecision::graceful_complete( queued_message.value.clone(), )); } if context.has_any_timer_fired() { return Ok(StepDecision::graceful_complete( "approval timed out".to_owned(), )); } let approval_message_values = APPROVAL_MESSAGES.condition_results(context)?; Ok(StepDecision::graceful_complete( approval_message_values[0].clone(), )) } fn publish_approval_message(&self, context: &mut Context) -> HandlerResult<()> { APPROVAL_MESSAGES.publish(context, "approved".to_string()) } ``` 例子: [examples/rust/src/primitives/channel/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/channel/flow.rs) 如果第二个 Channel **QueuedMessages** 上有 RPC 放入的待处理消息，**Execute** 会先取走它，见[管理待处理消息](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md#manage-pending-messages)。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](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#manage-pending-messages)。 ## ChannelMap 一个 Flow 需要很多独立队列时，例如每个订单、客户或任务一条，用 **ChannelMap**。ChannelMap 有一个名字和一种消息类型。每个 instance key 有自己的 FIFO 队列。 ## 参考 每个 SDK 参考的 [Channel](https://docs.superdurable.io/zh-Hans/reference/python/channel.md) 页列出该语言的 API 名称。 [Data primitive semantics](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#channels) 详细说明 wait condition、待处理消息、删除、状态加载和 ChannelMap key。