Temporal介绍_temporal 工作原理

作者：小舞很执着 | 2024-07-13 22:26:49

踩

temporal 工作原理

1、前言

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

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

在以往的开发过程中，可能更多的直接使用数据库、定时任务、消息队列等作为基础，来解决上面的问题

然而即便如此，在代码开发中，也会有很多代码跟业务无关，比如外部服务低可靠性情况下的重试，多异步事件下的程序逻辑组织等。

最终会影响开发效率，并且可能会降低代码后期的维护性。

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

2、工作流介绍

2.1、定义

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

2.2、目标

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

2.3、适用场景

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

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

2.4、常见工作流框架

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

Java: jBPM、Activiti、SWF

PHP: Tpflow、PHPworkflow

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

3、Temporal工作流基本概念

3.1、Workflow

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

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

3.1.1、工作流选项

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

支持的配置参数如下：

超时设置

ExecuteTimeout：Workflow的最大运行时间，包括失败后重试的时间。默认值是10年
RunTimeout：Workflow单次运行的时间，默认值为 ExecuteTimeout
TaskTimeout：从Worker从任务队列拉取到Workflow任务，到Worker开始执行Workflow的时间。如果超时，Server会认为Worker已经挂掉，会重新调度该Workflow给其他Worker，默认值10s

重试策略

InitialInterval：第一次重试前，需要等待多久。无默认值
BackoffCeofficient：退避系数，表示多次重试时，下次等待的时间是上次的多少倍，默认值：2
MaximumnInterval：下次重试时，最大等待时间。默认值：100*初始等待时间
MaximumAttempts：做大重试次数，默认值：0，表示无限重试
Non-Retryable：表示Workflow遇到哪些Error后，不再进行重试

Workflow Id：

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

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

多个Workflow使用相同ID时的策略配置

Allow duplicate failed only policy：只有前一个相同ID的Workflow失败后，才可以再启动下一个相同ID的Workflow
Allow duplicate policy：允许两个相同ID的Workflow同时运行。默认策略
Reject duplicate policy：任何时候，不允许有相同ID的Workflow

定时运行

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

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

3.2、 Activities

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

Activities可以理解为一个业务操作单元

在Workflow执行过程中，会将Activity放入消息队列，由其他Worker获取后，执行该Activity，并将结果再返回给Workflow。

3.2.1、Activity选项

超时配置

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

重试策略

和Workflow的重试策略完全一致

执行时间超长的Activity

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

4、工作流样例


package subscription
 
import (
    "log"
    "time"
 
    "go.temporal.io/sdk/workflow"
)
 
// 此处定义该函数作为一个工作流，用于处理用户订阅的服务
// 工作流的编写方式和普通函数相似，需要注意的地方是，函数中涉及到可能失败的情况都应该封装到Activity中执行
func SubscriptionWorkflow(ctx workflow.Context, customer Customer) (string, error) {
    workflowCustomer := customer
    subscriptionCancelled := false
    billingPeriodNum := 0
    actResult := ""
 
    QueryCustomerIdName := "customerid"
    QueryBillingPeriodNumberName := "billingperiodnumber"
    QueryBillingPeriodChargeAmountName := "billingperiodchargeamount"
 
    logger := workflow.GetLogger(ctx)
 
    // 通过SetQueryHandler方法，外部服务向Temporal Server发送请求查询该工作流的内部变量状态
    // 比如，根据下面QueryHandler的定义，外部服务可以查看该工作流的账单信息
    err = workflow.SetQueryHandler(ctx, QueryBillingPeriodChargeAmountName, func() (int, error) {
        return workflowCustomer.Subscription.BillingPeriodCharge, nil
    })
    if err != nil {
        logger.Info("QueryBillingPeriodChargeAmountName handler failed.", "Error", err)
        return "Error", err
    }
     
 
    // 下面的代码定义了一个Selector，Selector通常用于等待一个或多个异步时间或超时通知
    // 此处定义的Selector，用于接受用户取消订阅的事件，收到该事件后，修改了变量 subscriptionCancelled 的值
    cancelSelector := workflow.NewSelector(ctx)
    cancelCh := workflow.GetSignalChannel(ctx, "cancelsubscription")
    cancelSelector.AddReceive(cancelCh, func(ch workflow.ReceiveChannel, _ bool) {
        var cancelSubSignal bool
        ch.Receive(ctx, &cancelSubSignal)
        subscriptionCancelled = cancelSubSignal
    })
     
 
    // 此处配置了Activity的执行超时时间
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: time.Minute * 5,
    }
 
    ctx = workflow.WithActivityOptions(ctx, ao)
    logger.Info("Subscription workflow started for: " + customer.Id)
 
    var activities *Activities
 
    // 执行一个向用户发送欢迎邮件的Activity
    err = workflow.ExecuteActivity(ctx, activities.SendWelcomeEmail, workflowCustomer).Get(ctx, &actResult)
    if err != nil {
        log.Fatalln("Failure executing SendWelcomeEmail", err)
    }
 
    // 程序等待，直到超时或用户取消订阅
    workflow.AwaitWithTimeout(ctx, workflowCustomer.Subscription.TrialPeriod, func() bool {
        return subscriptionCancelled == true
    })
 
    // 如果用户在试用期，取消订阅，发送取消订阅的通知邮件
    if subscriptionCancelled == true {
        err = workflow.ExecuteActivity(ctx, activities.SendCancellationEmailDuringTrialPeriod, workflowCustomer).Get(ctx, &actResult)
        if err != nil {
            log.Fatalln("Failure executing SendCancellationEmailDuringTrialPeriod", err)
        }
        // 结束此工作流
        return "Subscription finished for: " + workflowCustomer.Id, err
    }
    ...
 
    return "Completed Subscription Workflow", err
}