当前位置：首页 > news >正文

03 Temporal 详细介绍

news 文章来源：https://blog.csdn.net/spirit_8023/article/details/134954630 2025/1/11 23:54:12

前言

在后端开发中，大家是否有遇到如下类型的开发场景

需要处理较多的异步事件
需要的外部服务可靠性较低
需要记录保存某个对象的复杂状态

在以往的开发过程中，可能更多的直接使用数据库、定时任务、消息队列等作为基础，来解决上面的问题。然而即便如此，在代码开发中，也会有很多代码跟业务无关，比如外部服务低可靠性情况下的重试，多异步事件下的程序逻辑组织等。最终会影响开发效率，并且可能会降低代码后期的维护性。

针对上面的场景，抽象为 工作流 模式的话，可以减轻开发成本以及维护成本

工作流介绍

定义：指业务过程的部分或整体在计算机应用环境下的自动化。是对工作流程及其各操作步骤之间的业务规则的抽象、概括描述。

主要解决的问题：为了实现某个业务目标，利用计算机在多个参与者之间按照某种预订的规则自动传递问文档、信息或者任务。

适用场景

工作流通常适用于，有状态的、异步、长时间执行等特性的业务场景，比较典型的场景包括

视频、音频、图片处理工作流
订单、审批流程
数据处理流水线
自动化运维

常见工作流框架

工作流框架还是比较多的，按照语言分类的话，有

Java: jBPM、Activiti、SWF
PHP: Tpflow、PHPworkflow
Go: Cadence（Cadence由Uber开发并开源，Maxim Fateev是Cadence的主架构师）、Temporal（Maxim Fateev为了推广Workflow编排引擎的商业化，另立门户创建了Temporal）

Temporal 工作流基本概念

1、原理

在业务模块当中按规则编写 Workflow 流程以及其具体的 Activity，并注册到 worker 当中，启动 worker 外部⽤户触发 Workflow，Temporal 编排 workflow 形成⼀系列的 task 送到队列中，worker 去队列取任务，执⾏后将结果返回给Temporal。

举一个银行的流程示例：

由四部分组成Start、Temporal Server、Worker、Bank

Start：工作流的创建者/发起者
Temporal Server：存储所有工作流的数据、状态的中间件，整个工作依赖于该 server（后续简写为TS）
Worker：实际进行逻辑处理的执行者
Bank：官方给的示例，可以理解为DB

具体的流程描述：

启动 Temporal Server
启动 Worker 监听TS，循环获取待执行的工作流
Start 创建一个工作流，封装参数，调用 sdk 的 api(rpc) 发送到 TS
Worker 拉取到工作流开始逻辑处理

2、Workflow

workflow 即表示工作流，在 Temporal 中，工作流是由函数或对象方法来实现（工作流样例见下文）。

一个 workflow 通常完成一个业务目标。同时，当多个 workflow 中，有同样的处理流程时，可以封装为一个子的workflow，来达到代码复用的目的。

2.1、工作流选项

启动Workflow的时候，可以设置这个Wrokflow的执行超时时间，以及失败后的重试次数、任务队列名等参数，来更好的满足业务需求

支持的配置参数如下：

workflowOptions := client.StartWorkflowOptions{ID:                       "hello_world_" + uuid.New(), //用于业务级别标识。不可能有两个一样的workflowId同时工作TaskQueue:                "hello-world",               //活动任务队列。让收到任务的 Worker 知道下一步要执行哪一段代码。 Workflows（工作流）只能使用一个任务队列WorkflowExecutionTimeout: 10 * time.Minute,            //Workflow的最大运行时间，包括失败后重试的时间。默认无限制WorkflowRunTimeout:       3 * time.Minute,             //单次运行的时间。默认值为 ExecuteTimeoutWorkflowTaskTimeout:      10 * time.Second,            //从Worker从任务队列拉取到Workflow任务，到Worker开始执行Workflow的时间。如果超时，Server会认为Worker已经挂掉，会重新调度该Workflow给其他Worker，默认值10sRetryPolicy: &temporal.RetryPolicy{ //重试策略InitialInterval:        30 * time.Second,                         //初始间隔 描述：第一次重试前，需要等待多久。无默认值。如果提供重试策略，则必须提供一个值。用例：这用作退避系数乘以对抗的基本间隔时间BackoffCoefficient:     2,                                        //退避系数 描述：退避系数，表示多次重试时，下次等待的时间是上次的多少倍。默认值设置为 2.0。回退系数为 1.0 表示重试间隔始终等于初始间隔。用例：使用此来增加重试之间的时间间隔。通过具有回退系数，前几次重试相对较快地发生以克服间歇性故障，但随后的重检将发生越来越远的距离，以考虑更长的持久中断。使用 maximum interval 最大间隔选项来防止系数过多地增加重试间隔。MaximumInterval:        5 * time.Minute,                          //最大间隔 描述：下次重试时，最大等待时间。默认值：100*初始等待时间。用例：这对于大于 1.0 的系数很有用，因为它可防止间隔以指数级无限增长。MaximumAttempts:        1,                                        //最大重试次数 描述：默认值：0，表示无限重试，但在大多数情况下，建议依靠执行超时来限制检索的持续时间，而不是此选项。NonRetryableErrorTypes: []string{"TemporalTimeout:StartToClose"}, //表示Workflow遇到哪些Error后，不再进行重试},}

2.2、Workflow Id

一个Workflow，可由命名空间，Workflow Id 和 Run id 唯一标识

启动Workflow的时候，可以指定一个ID，这个ID一般采用业务级的ID，如一个要处理的客户的ID或订单ID

2.3、定时运行

启动Workflow的时候，可以设置为定时启动。

📢 注意。如果到了下次运行Workflow的时候，但上次的Workflow还没执行完（可能任务执行耗时长，或由于失败后重试等原因），会跳过下次运行Workflow

2.4、查询工作流状态

we = client.GetWorkflow(workflowID)var result stringwe.Get(ctx, &result) // 获取是阻塞的

3、 Activities

Activities可以理解为一个业务操作单元。在Workflow执行过程中，会将Activity放入消息队列，由其他Worker获取后，执行该Activity，并将结果再返回给Workflow。

3.1、Activity 选项

（1）超时配置

ScheduleToStartTimeout：表示Activity任务放到消息队列，到Worker获取到的超时时间。如果超时后，也不会触发重试（不建议设置该值）
StartToCloseTimeout：Activity实际执行超时时间。如果Activity执行时间不确定，最好按照最长时间设置。比如一个Activity可能需要2分钟、有时需要5分钟，那就设置为5分钟
ScheduleToCloseTimeout：从Activity放入消息队列，到Activity执行完成的时间
HeartbeatTimeout：Activity和Server的心跳超时时间。在Activity运行需要较长时间时需要。用于Server检查执行 Activity的Worker 是否已经挂掉

（2）重试策略

和Workflow的重试策略完全一致

（3）执行时间超长的Activity

如果一个Activity运行时间较长，最好设置一个心跳间隔超时。这样当执行Activity的Woker挂掉时，Server可以及时知道

3.2、工作流代码有变化时如何更改

三种方法可供使用

基于任务队列的版本控制（建议）
- 更改新工作流的任务队列名，让旧的 worker 继续运行，可减少实例的数量，知道全部执行
- 优点：概念简单
- 缺点：旧的 worker 长时间运行，且不能当前正在运行的 bug
基于工作流名称的版本控制
Patch和GetVersion api
- 根据 SKD 的 api workflow.GetVersion() 分支运行的新旧工作流代码
- 优点：兼容性好
- 缺点：长时间变更会引起使用上的歧义

4、Signal

对于正在运行的 WorkflowExecution，可以发送携带参数的信号，Workflow中可以等待或根据条件处理信号，动态控制工作流的执行逻辑。示例

5、Child Workflows

考虑工作流执行事件历史大小限制需要对父 Workflow 做拆分。希望将每个子工作流执行视为单独的服务。示例

6、Selectors

类似Go的select，允许goroutine等待多个通信。一个select块直到它的一个case可以运行，然后执行。如果多个已准备就绪，则随机选择一个。

Temporal 集群架构

前端组件（Frontend Service）
- 是一个单点网关，提供 Proto API。可以接受来自浏览器、tctl（Temporal的命令行工具）、以及业务方的调用请求。（主要用于接口限速、授权认证、校验和请求路由）
记录服务（History service）
- 用于记录Workflow的执行状态，并且支持横向拓展。
匹配服务（Matching service）
- 用于管理任务队列，及任务分发，并且支持横向拓展
后台服务（Worker service）
- 用于维护拷贝队列和执行一些Temporal服务自己的Wrokflow。
数据库
- 保存了用于分发的任务信息、以及Workflow的执行状态、命名空间元数据，前端可视化配置

工作流样例

Golang SDK文档: Temporal Go SDK developer's guide | Temporal Documentation
Golang 示例代码：GitHub - temporalio/samples-go: Temporal Go SDK samples
与正常业务代码对比：Open Source Durable Execution Platform | Temporal Technologies

常用 API 使用示例代码

Workflow执行状态、结果查看

Temporal 官网提供了Web UI组建。可以通过浏览器查看Workflow的执行状态和结果

1、看个简单的 hello world

1）worker 部分：主要是做工作流的注册，是执行 Activity、Workflow 逻辑的进程，并将执行结果返回给 Temporal Server

点击此处展开...

2）starter 部分：发起工作流的进程