---
title: "RPC"
description: "RPC（Remote Procedure Call）让 Flow 外的代码与正在运行的 Flow 交互。web handler、另一个服务或运维工具通过 Client 调用它。"
canonical_url: "https://docs.superdurable.io/zh-Hans/concepts/primitives/rpc/"
source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/concepts/primitives/rpc.mdx"
locale: "zh-Hans"
---

# RPC

**RPC**（Remote Procedure Call）让 Flow 外的代码与正在运行的 Flow 交互。web handler、另一个服务或运维工具通过 [Client](https://docs.superdurable.io/zh-Hans/concepts/client.md) 调用它。

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

Client 将 RPC 请求发送给 Dex Server，由 Dex Server 转发给 Application Worker。Worker 通过 Dex Server 将结果返回给 Client。

## 什么时候用 RPC

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

RPC handler 可以做四件事：

- 读写 [Attribute](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md)；
- 向 [Channel](https://docs.superdurable.io/zh-Hans/concepts/primitives/channel.md) 发布消息；
- 启动新的 [Step](https://docs.superdurable.io/zh-Hans/concepts/step.md)；
- 取消正在运行的 Step。

<a id="read-only-rpcs-and-terminal-flows"></a>

只是读取时，Client 可以不用 RPC，见[什么时候读取需要 RPC](https://docs.superdurable.io/zh-Hans/concepts/client.md#related-primitives)。只读的 RPC，包括在已关闭的 Flow 上调用的，见 [Read-only RPCs and terminal Flows](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#read-only-rpcs-and-terminal-flows)。

## 保证

Dex 像保存 Step 方法一样保存 RPC：它把 RPC 的 Attribute 写入和 Channel 发布与结果一起保存，而且只在 RPC 成功时保存。见 [Step 的保证](https://docs.superdurable.io/zh-Hans/concepts/step.md#guarantees)。

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 要能安全地重复执行。

<a id="rpc-timeout"></a>

RPC 和 Step 一样在 Worker 上运行。每次 handler 调用都受 timeout 限制。见 [RPC timeout](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#rpc-timeout)。

## 定义和调用 RPC

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

**Python**

```python
@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](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/rpc)

**Go**

```go
func (flow *RpcFlow) GetRPCs() []dex.RPCDef {
	return []dex.RPCDef{
		dex.DefineRPC(flow.Trigger, &dex.RPCOptions{Timeout: 30 * time.Second}),
	}
}

func (*RpcFlow) Trigger(ctx dex.Context, input string) (*dex.RPCResult[string], error) {
	if err := Data.Set(ctx, input); err != nil {
		return nil, err
	}
	if err := ExampleCh.Publish(ctx, nil); err != nil {
		return nil, err
	}
	return &dex.RPCResult[string]{
		Output: input,
		NextSteps: []dex.StepMovement{
			dex.MovementOf(exampleStep{}, input),
		},
	}, nil
}
```

例子: [examples/go/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/go/primitives/rpc)

**Java**

```java
@RPC
public RPCResult<String> trigger(final Context context, final String input) {
    data.set(context, input);
    exampleCh.publish(context, null);
    return RPCResult.of(input, StepMovement.of(Example.class, input));
}
```

例子: [examples/java/src/main/java/io/superdurable/dex/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/rpc)

**TypeScript**

```typescript
@rpc({ inputCodec: stringCodec, outputCodec: stringCodec })
public trigger(context: Context, input: string): RPCResult<string> {
  this.data.set(context, input);
  exampleCh.publish(context, undefined);
  return { output: input, nextSteps: [StepMovement.of(Example, input)] };
}
```

例子: [examples/typescript/src/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/rpc)

**Rust**

```rust
impl RpcFlow {
    fn trigger(&self, context: &mut Context, input: String) -> HandlerResult<RpcResult<String>> {
        data().set(context, input.clone())?;
        example_ch().publish(context, ())?;
        Ok(RpcResult::new(input.clone()).then(StepMovement::to(&self.example_step, input)))
    }
}
```

例子: [examples/rust/src/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/rpc)

然后从 Client 调用这个 RPC：

**Python**

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

例子: [examples/python/dex_examples/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/rpc)

**Go**

```go
var result string
err := client.InvokeRPC(
	ctx,
	flowID,
	flow.Trigger,
	message,
	&result,
)
```

例子: [examples/go/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/go/primitives/rpc)

**Java**

```java
final RpcFlow flow = client.newRpcStub(RpcFlow.class, flowId);
final String result = client.invokeRPC(flow::trigger, message);
```

例子: [examples/java/src/main/java/io/superdurable/dex/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/rpc)

**TypeScript**

```typescript
const result = await client.invokeRPC(flow.trigger, flowId, message);
```

例子: [examples/typescript/src/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/rpc)

**Rust**

```rust
let result = client.invoke_rpc(&workflow_id, TRIGGER_RPC, message)?;
```

例子: [examples/rust/src/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/rpc)

<a id="java-和-kotlin"></a>

<a id="typescript-codec"></a>

Java、Kotlin 和 TypeScript 还有额外的 RPC 规则，见 [RPC stubs and Kotlin](https://docs.superdurable.io/zh-Hans/reference/java/rpc.md#rpc-stubs-and-kotlin) 和 [RPC codecs](https://docs.superdurable.io/zh-Hans/reference/typescript/rpc.md#rpc-codecs)。

<a id="cancel-running-steps"></a>

## 取消正在运行的 Step

<a id="cancel-race"></a>

<a id="cancel-honour"></a>

RPC 的结果可以按 Step type 取消所在 Flow 中的活跃 Step。外部请求让进行中的工作失去意义时用它，例如报告或模型调用运行时用户按下 Stop。Dex 在保存这个 RPC 的结果时执行取消。Step 代码必须检查取消并停下。示例见 [Cancel running Steps](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#cancel-running-steps)，与先完成的 Step 的竞争见 [Resolve the race from committed state](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#cancel-race)，各 SDK 如何传递取消见 [Make the Step stop](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#cancel-honour)。

<a id="transactional-reads-and-writes"></a>

## Lock 和 transactional RPC

<a id="attribute-locks"></a>

RPC 可以声明 [Attribute lock](https://docs.superdurable.io/zh-Hans/concepts/primitives/attribute.md#locking)。这样它会和声明了同一个 lock 的 Step 和 RPC 依次执行。见 [Locks](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#attribute-locks) 和 [RPC lock conflicts](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#rpc-lock-conflicts)。

Dex 保存 **transactional** RPC 的结果时，会多做一次检查。如果它删除的 Channel 消息已经不在队列中，RPC 失败，Dex 不保存它的任何修改。非 transactional RPC 会跳过缺失的删除，Dex 保存其余修改。持有 lock 的 RPC 总是 transactional。见 [Transactional RPCs](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#transactional-reads-and-writes)。

<a id="load-collection-state"></a>

<a id="运行时选择-map-instance"></a>

## 加载集合

RPC 会收到普通 Attribute 和 Channel 大小。只有声明之后，它才加载 AttributeMap 条目和待处理的 Channel 消息。这样不用这些大集合的 RPC 就不会带上它们。见 [Selective state loading](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#load-collection-state) 和 [Runtime-selected map instances](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#runtime-selected-map-instances)。

## 参考

每个 SDK 参考的 [RPC](https://docs.superdurable.io/zh-Hans/reference/python/rpc.md) 页列出该语言的 API 名称。

[Data primitive semantics](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#rpcs) 详细说明取消、已关闭 Flow 上的只读 RPC、状态加载、transactional RPC 和 timeout。