跳到主要内容
Super Durable
简体中文
English简体中文
GitHub登录注册 Cloud
菜单

RPC

查看 Markdown

RPC（Remote Procedure Call）让 Flow 外的代码与正在运行的 Flow 交互。web handler、另一个服务或运维工具通过 Client 调用它。

Client 发起 RPC。Dex Server 将它发送给 Worker。Worker 执行后，将结果返回给 Dex Server，由它保存。Dex Server 再将输出返回给 Client。

什么时候用 RPC​

Flow 外的事情要改变 Flow，或者要问 Flow 一个问题时，用 RPC。例如一次审批、用户按下 Stop，或者请求查看 Flow 状态。

RPC handler 可以做四件事：

  • 读写 Attribute；
  • 向 Channel 发布消息；
  • 启动新的 Step；
  • 取消正在运行的 Step。

只是读取时，Client 可以不用 RPC，见什么时候读取需要 RPC。只读的 RPC，包括在已关闭的 Flow 上调用的，见 Read-only RPCs and terminal Flows。

保证​

Dex 像保存 Step 方法一样保存 RPC：它把 RPC 的 Attribute 写入和 Channel 发布与结果一起保存，而且只在 RPC 成功时保存。见 Step 的保证。

RPC handler 读取的是 Flow 的一个快照。在这次读取之后、Dex 保存 RPC 的修改之前，Step 和其他 RPC 可能已经修改了 Flow。transactional RPC 也不改变这一点。RPC 的修改依赖它读到的内容时，在这个 RPC 以及所有写这个 Attribute 的 Step 和 RPC 上，都声明这个 Attribute 的 lock。Dex 会让它们依次执行。如果 handler 运行期间 Flow 关闭了，Dex 会拒绝这个 RPC 的修改。但 handler 对外部系统的调用已经发生了。

SDK 发出的每次 RPC 调用都是一个新请求。重试一次调用时，Dex 可能再次运行 handler，并再次保存它的修改。RPC handler 要能安全地重复执行。

RPC 和 Step 一样在 Worker 上运行。每次 handler 调用都受 timeout 限制。见 RPC timeout。

定义和调用 RPC​

这个 RPC 写一个 Attribute，发布一条 Channel 消息，并启动一个 Step：

rpcPython
@rpc
def trigger(self, context: Context, input: str) -> RPCResult[str]:
    self.data.set(context, input)
    self.example_ch.publish(context, None)
    return RPCResult(input, next_steps=(StepMovement.of(Example, input),))

例子: examples/python/dex_examples/primitives/rpc

然后从 Client 调用这个 RPC：

rpcPython
result = await client.invoke_rpc(flow.trigger, flow_id, message)

例子: examples/python/dex_examples/primitives/rpc

Java、Kotlin 和 TypeScript 还有额外的 RPC 规则，见 RPC stubs and Kotlin 和 RPC codecs。

取消正在运行的 Step​

RPC 的结果可以按 Step type 取消所在 Flow 中的活跃 Step。外部请求让进行中的工作失去意义时用它，例如报告或模型调用运行时用户按下 Stop。Dex 在保存这个 RPC 的结果时执行取消。Step 代码必须检查取消并停下。示例见 Cancel running Steps，与先完成的 Step 的竞争见 Resolve the race from committed state，各 SDK 如何传递取消见 Make the Step stop。

Lock 和 transactional RPC​

RPC 可以声明 Attribute lock。这样它会和声明了同一个 lock 的 Step 和 RPC 依次执行。见 Locks 和 RPC lock conflicts。

Dex 保存 transactional RPC 的结果时，会多做一次检查。如果它删除的 Channel 消息已经不在队列中，RPC 失败，Dex 不保存它的任何修改。非 transactional RPC 会跳过缺失的删除，Dex 保存其余修改。持有 lock 的 RPC 总是 transactional。见 Transactional RPCs。

加载集合​

RPC 会收到普通 Attribute 和 Channel 大小。只有声明之后，它才加载 AttributeMap 条目和待处理的 Channel 消息。这样不用这些大集合的 RPC 就不会带上它们。见 Selective state loading 和 Runtime-selected map instances。

参考​

每个 SDK 参考的 RPC 页列出该语言的 API 名称。

Data primitive semantics 详细说明取消、已关闭 Flow 上的只读 RPC、状态加载、transactional RPC 和 timeout。