- 前言
- 一、Temporal是什么？
- 二、组件有哪些？
- 三、TS启动
- 四、样例代码
- 1.目录结构
- 2.代码
- 2.1 start
- 2.2 activity
- 2.3 workflow
- 2.4 worker
- 五、运行结果
- 六、生产部署
- END
公司前员工优化了内部的视频转码流程，引入了工作流，使用的是Temporal框架，其离职后没人了解该项目。本人出于学习的态度便学习Temporal框架，并希望后续能独立接手该项目，积累实战经验，所以深入学习了Temporal。
我们的视频工作流只要包括：去水印、视频下载、多码率转码等。
官方说明：Temporal是一种分布式、可扩展、持久且高度可用的编排引擎，用于以可扩展和弹性的方式执行一步长时间运行的业务逻辑。Temporal Go SDK是使用Go语言创作工作流和活动的框架。
从官方说明SDK里可以看到2个关键词 “工作流” 和 “活动”，这两个是主体。
官方给的组件示意图如下，由四部分组成Start、Temporal Server、Worker、Bank。
- Start：工作流的创建者/发起者
- Temporal Server：存储所有工作流的数据、状态的中间件，整个工作依赖于该server（后续简写为TS）
- Worker：实际进行逻辑处理的执行者
- Bank：官方给的示例，可以理解为DB
具体的流程描述：启动Temporal Server。启动Worder监听TS，循环获取待执行的工作流。Start创建一个工作流，封装参数，调用sdk的api(rpc)发送到TS。Worker拉取到工作流开始逻辑处理。
woker是Activity和Workflow的包装，worker的唯一工作就是执行Activity和Workflow并将结果返回给TS。
一个Wokeflow包含多个Activity，对Activity进行编排，多个Activity可以并行，也可以同步（阻塞到都某个Activity执行完毕）。其底层会阻塞到Future.Get()方法上。
在本地下载并启动TS。
$ git clone https://github.com/temporalio/docker-compose.git $ cd docker-compose $ docker-compose up
使用temporal的命令行工具与本地TS交互。
$ alias tctl="docker exec temporal-admin-tools tctl" $ tctl namespace list $ tctl workflow list
启动后，网页里输入 http://localhost:8088，检查TS是否正常运行。
package main
import (
"context"
"go.temporal.io/sdk/client"
"log"
"temporal/workflow"
)
type Student struct {
Id int64
Name string
}
func main() {
c, err := client.NewClient(client.Options{})
if err != nil {
log.Fatalln("unable to create Temporal client", err)
}
defer c.Close()
options := client.StartWorkflowOptions{
ID: "xl_test2", // ID应唯一
TaskQueue: "xl_test_queue",
}
student := Student{
Id: 1,
Name: "xiaoming",
}
we, err := c.ExecuteWorkflow(context.Background(), options, workflow.HandleStudent, student)
if err != nil {
log.Println("failed to ExecuteWorkflow err: %s", err.Error())
return
}
log.Println(we.GetID(), we.GetRunID())
}
2.2 activity
package activity
import (
"context"
"fmt"
)
func PrintStudent(ctx context.Context, student interface{}) (err error) {
// 实现活动的具体逻辑
fmt.Println("print student %v", student)
return
}
2.3 workflow
package workflow
import (
"fmt"
"go.temporal.io/sdk/temporal"
"go.temporal.io/sdk/workflow"
"temporal/activity"
"time"
)
func HandleStudent(ctx workflow.Context, student interface{}) (err error) {
retrypolicy := &temporal.RetryPolicy{
InitialInterval: time.Second,
BackoffCoefficient: 2.0,
MaximumInterval: time.Minute,
MaximumAttempts: 500,
}
options := workflow.ActivityOptions{
// Timeout options specify when to automatically timeout Actvitivy functions.
StartToCloseTimeout: time.Minute,
// Optionally provide a customized RetryPolicy.
// Temporal retries failures by default, this is just an example.
RetryPolicy: retrypolicy,
}
// 特定位置执行特定活动，对活动可以进行编排
ctx = workflow.WithActivityOptions(ctx, options)
// activity.PrintStudent 底层会反射拿到该方法名
// 也可以直接传字符串，但是会失去方法的参数校验
printFuture := workflow.ExecuteActivity(ctx, activity.PrintStudent, student)
// Future.Get()方法会阻塞，获取活动结果
err = printFuture.Get(ctx, nil)
if err != nil {
fmt.Println("failed handle student err: %s", err.Error())
return
}
return
}
2.4 worker
package main
import (
"go.temporal.io/sdk/client"
"go.temporal.io/sdk/worker"
"log"
"temporal/activity"
"temporal/workflow"
)
func main() {
c, err := client.NewClient(client.Options{})
if err != nil {
log.Fatal("new client failed")
return
}
defer c.Close()
// 创建worker，并注册workflow和activity
w := worker.New(c, "xl_test_queue", worker.Options{})
w.RegisterWorkflow(workflow.HandleStudent)
w.RegisterActivity(activity.PrintStudent)
err = w.Run(worker.InterruptCh())
if err != nil {
log.Fatalln("unable to start Worker", err)
}
}
五、运行结果
1、写入一个任务
localhost:temporal zuiyou$ go run start/main.go 2021/07/19 20:47:38 INFO No logger configured for temporal client. Created default one. 2021/07/19 20:47:38 xl_test2 1bed122b-7be5-489b-a896-c9b4d1320f35
2、启动worker处理
localhost:temporal zuiyou$ go run worker/main.go 2021/07/19 20:49:26 INFO No logger configured for temporal client. Created default one. 2021/07/19 20:49:26 INFO Started Worker Namespace default TaskQueue xl_test_queue WorkerID 21499@xxxdeMacBook-Pro.local@ 2021/07/19 20:49:26 DEBUG ExecuteActivity Namespace default TaskQueue xl_test_queue WorkerID 21499@xxxdeMacBook-Pro.local@ WorkflowType HandleStudent WorkflowID xl_test2 RunID 1bed122b-7be5-489b-a896-c9b4d1320f35 Attempt 1 ActivityID 5 ActivityType PrintStudent print student %v map[Id:1 Name:xiaoming]六、生产部署
我们视频转码业务中有三个服务：
- mediastarter：触发网关层，用于接收转码请求，并向Temporal发起工作流请求
- temporal 引擎，负责工作流的调度
- video-worker 工作组件，负责从Temporal中pull任务，执行每一个转码动作
temporal引擎的部署主要涉及以下几个步骤： - 修改配置。主要修改 development_xxx.yaml 即可，其中xxx表示的是temporal的身份。我们有2台机器，active即主机，standby为备机。数据库mysql，ES(v7及以上)。
- 准备组件。主要是处理mysql和es，在temporal工程下，需要执行make install-schema-mysql-cdc和make install-schema-es。注意要先修改其中的对应段的内容，主要是替换地址和密码，es要注意是否使用v7版本。
- 注册命名空间。当前的视频转码工作流都是运行在media的namespace下。在temporal工程下执行make bins，能得到tctl可执行工具，之后执行以下shell，创建全局命名空间，一定要是全局
$ ./tctl --ad active_ip:7233 --namespace media_v2 n re --gd true --cl active standby
- 查看工作流信息
$ ./tctl --ad 10.12.4.32:7233 --namespace media workflow show -w video_workflow_vid_353416365557END
