在 Encore 应用中集成 Temporal：工作流编排的完整实战指南

在 Encore 应用中集成 Temporal工作流编排的完整实战指南【免费下载链接】encoreThe infrastructure platform for the intelligence era项目地址: https://gitcode.com/GitHub_Trending/encor/encore导读Temporal 是一套用于构建高可靠系统的工作流编排Workflow Orchestration平台能够以持久化、可重放的方式编排分布式业务流程。Encore 作为面向云时代的后端开发平台与 Temporal 可以无缝共存——本指南将带你从零搭建一套Encore 业务 API Temporal 工作流的完整链路包括本地与云端 Temporal 集群的选型配置、在 Encore 服务中启动 Worker、定义 Workflow 与 Activity、通过 Encore API 触发工作流以及利用 Encore 的环境感知配置能力自动切换不同环境下的 Temporal 集群地址。读完本文你将掌握在 Encore 应用中生产级接入 Temporal 的完整实操方案。为什么在 Encore 中使用 TemporalTemporal 的核心价值在于把业务流程抽象为可持久化、可重试、可恢复的工作流即使进程崩溃、机器宕机工作流的执行状态依然被保留恢复后可以从断点继续执行。而 Encore 负责应用的后端基础设施API 路由、数据库、配置、部署等两者关注点正交、互不冲突因此 Encore 官方文档明确说明Encore works great with Temporal二者可以非常自然地组合使用。典型的组合方式是Encore负责对外暴露 HTTP API、管理服务生命周期、处理环境感知配置、部署到云端Temporal负责在 API 背后执行长时间运行、多步骤、需要可靠性保障的业务流程。第一步搭建 Temporal 集群在接入之前你需要准备至少两个 Temporal 集群一个用于本地开发一个用于云端环境。官方推荐的组合是场景推荐方案说明本地开发Temporalite轻量级、单进程即可运行的 Temporal 开发版云端环境Temporal CloudTemporal 官方托管服务或自建自托管的 Temporal 集群Temporalite 特别适合本地开发因为它不需要依赖外部数据库如 Cassandra/PostgreSQL开箱即用。云端则优先考虑 Temporal Cloud省去运维集群的成本如果你的组织已经有自托管集群同样可以在云端环境直接复用。第二步创建 Encore 服务并启动 Temporal Worker2.1 新建 greeting 服务首先在 Encore 应用中创建一个名为greeting的新服务。服务启动时会创建 Temporal client 与 worker核心代码如下greeting/greeting.gopackage greeting import ( context fmt go.temporal.io/sdk/client go.temporal.io/sdk/worker encore.dev ) // Use an environment-specific task queue so we can use the same // Temporal Cluster for all cloud environments. var ( envName encore.Meta().Environment.Name greetingTaskQueue envName -greeting ) //encore:service type Service struct { client client.Client worker worker.Worker } func initService() (*Service, error) { c, err : client.Dial(client.Options{}) if err ! nil { return nil, fmt.Errorf(create temporal client: %v, err) } w : worker.New(c, greetingTaskQueue, worker.Options{}) err w.Start() if err ! nil { c.Close() return nil, fmt.Errorf(start temporal worker: %v, err) } return Service{client: c, worker: w}, nil } func (s *Service) Shutdown(force context.Context) { s.client.Close() s.worker.Stop() }这段代码里有几个值得注意的设计点环境感知的任务队列命名通过encore.Meta().Environment.Name获取当前环境名本地开发时值为local云端环境则是对应的环境名拼接到任务队列名中。这样做的意义是同一个 Temporal 集群可以服务于所有云环境如 staging、prod不同环境使用不同的任务队列天然隔离、互不干扰。服务生命周期管理利用 Encore 的//encore:service声明式服务语法initService负责初始化创建 client、启动 workerShutdown负责优雅关闭停止 worker、关闭 client。2.2 encore.Meta() 的底层原理encore.Meta()由 Encore 运行时提供在 runtimes/go/pkgfn.go 中导出实际实现位于 runtimes/go/meta.go。其中EnvironmentMeta结构体包含三个字段见 runtimes/go/meta.goName当前环境名本地开发时为localType环境类型包括production、development、ephemeral、test等runtimes/go/meta.goCloud当前运行的云提供商本地运行时为CloudLocalruntimes/go/meta.go。这些元数据由运行时的配置注入mgr.runtime.EnvName、mgr.runtime.EnvCloud因此你可以在任何环境下安全地读取它们来定制业务逻辑——这正是环境特定任务队列能够成立的基础。第三步定义 Workflow 与 ActivityWorkflow 和 Activity 需要与服务放在同一个模块下以便注册。这里在greeting服务内部新建一个workflow子包将两者分文件组织3.1 定义 Workflowgreeting/workflow/workflow.gopackage workflow import ( time go.temporal.io/sdk/workflow ) func Greeting(ctx workflow.Context, name string) (string, error) { options : workflow.ActivityOptions{ StartToCloseTimeout: time.Second * 5, } ctx workflow.WithActivityOptions(ctx, options) var result string err : workflow.ExecuteActivity(ctx, ComposeGreeting, name).Get(ctx, result) return result, err }这里为 Activity 设置了StartToCloseTimeout: 5s超时这是 Temporal 可靠性的关键之一一旦 Activity 执行超过超时时间Temporal 会按照配置的重试策略处理避免工作流无限挂起。3.2 定义 Activitygreeting/workflow/activity.gopackage workflow import ( context fmt ) func ComposeGreeting(ctx context.Context, name string) (string, error) { greeting : fmt.Sprintf(Hello %s!, name) return greeting, nil }Activity 是真正执行业务动作的函数调用外部服务、读写数据库等而 Workflow 负责编排这些 Activity 的执行顺序与补偿逻辑。Temporal 对二者有明确的边界要求Workflow 代码必须是确定性的deterministic因此 IO 操作都应放在 Activity 中完成。3.3 注册 Workflow 与 Activity回到greeting服务在initService中把二者注册到 worker-- greeting/greeting.go -- // Import the package at the top: import encore.app/greeting/workflow // Add these lines to initService, below the call to worker.New: w.RegisterWorkflow(workflow.Greeting) w.RegisterActivity(workflow.ComposeGreeting)注册完成后Worker 就知道该监听greetingTaskQueue队列中的Greeting工作流及其关联的ComposeGreeting活动。第四步通过 Encore API 触发工作流新建greeting/greet.go定义一个公共 API 来触发工作流并返回结果package greeting import ( context encore.app/greeting/workflow encore.dev/rlog go.temporal.io/sdk/client ) type GreetResponse struct { Greeting string } //encore:api public path/greet/:name func (s *Service) Greet(ctx context.Context, name string) (*GreetResponse, error) { options : client.StartWorkflowOptions{ ID: greeting-workflow, TaskQueue: greetingTaskQueue, } we, err : s.client.ExecuteWorkflow(ctx, options, workflow.Greeting, name) if err ! nil { return nil, err } rlog.Info(started workflow, id, we.GetID(), run_id, we.GetRunID()) // Get the results var greeting string err we.Get(ctx, greeting) if err ! nil { return nil, err } return GreetResponse{Greeting: greeting}, nil }要点说明//encore:api public path/greet/:name声明了一个公共 API 端点路径参数:name会自动绑定到name参数无需手工解析 URLStartWorkflowOptions中指定了工作流 IDID与任务队列TaskQueue后者使用了之前的环境感知变量greetingTaskQueueExecuteWorkflow提交工作流后通过we.GetID()/we.GetRunID()拿到工作流标识并用rlog.Info记录日志Encore 的 rlog 会自动集成分布式追踪we.Get(ctx, greeting)会阻塞等待工作流执行完成并把结果反序列化到greeting。第五步本地运行与验证一切就绪后在两个独立终端分别启动 Temporalite 和 Encore 应用$ temporalite start --namespace default $ encore run然后通过 cURL 调用 API 进行验证$ curl http://localhost:4000/greeting/Temporal {Greeting: Hello Temporal!}如果看到上述输出说明整条链路Encore API → Temporal Client → Worker 队列 → Workflow → Activity → 返回结果已经跑通。你也可以通过 Encore 的 Local Development Dashboard 查看请求的追踪信息观察工作流的调用细节。第六步云端运行与环境感知配置本地验证通过后部署到云端时需要使用 Temporal Cloud 或自托管集群。最优雅的做法是利用Encore 的配置Config功能让不同环境自动使用正确的集群地址。6.1 定义配置结构greeting/config.gopackage greeting import encore.dev/config type Config struct { TemporalServer string } var cfg config.Load[*Config]()6.2 编写环境感知的 CUE 配置greeting/config.cuepackage greeting TemporalServer: [ // These act as individual case statements if #Meta.Environment.Cloud local { localhost:7233 }, // TODO: configure this to match your own cluster address my.cluster.address:7233, ][0] // Return the first value which matches the condition这里展示了 Encore 配置系统最强大的能力之一基于元数据#Meta的条件配置。列表中的每一项相当于一个 case 分支[0]取出第一个匹配条件成立的值。在本例中当环境云提供商为local即encore run本地运行时→ 使用localhost:7233其他任何环境staging、prod 等云端环境→ 回退到my.cluster.address:7233你只需把该地址替换为 Temporal Cloud 提供的命名空间地址或自托管集群地址。#Meta.Environment.Cloud与代码中encore.Meta().Environment.Cloud一脉相承都来自运行时元数据对应 runtimes/go/meta.go 中CloudProvider本地为CloudLocal。因此配置与代码共享同一套环境事实不会出现配置判断的环境与代码运行的环境不一致的问题。6.3 让 Client 使用配置的地址最后回到greeting/greeting.go把client.Dial改为使用配置值-- greeting/greeting.go -- client.Dial(client.Options{HostPort: cfg.TemporalServer})config.Load[*Config]()的实现位于 runtimes/go/config/pkgfn.goEncore 在编译期从各服务目录下的 CUE 文件计算配置并做校验运行时直接反序列化为类型化的配置结构体。换言之配置错误会在编译/启动阶段被尽早暴露而不是等到运行时才发现。完成以上改造后同一份代码在本地与云端会自动连接到各自对应的 Temporal 集群无需任何手工切换。完整文件结构总览至此greeting服务完整的目录结构如下greeting/ ├── greeting.go # 服务定义client、worker、注册 workflow/activity ├── greet.go # 公共 API触发工作流 ├── config.go # 配置结构体定义 ├── config.cue # 环境感知配置本地/云端集群地址 └── workflow/ ├── workflow.go # Workflow 定义 └── activity.go # Activity 定义实战要点小结环境隔离靠任务队列通过encore.Meta().Environment.Name生成{env}-greeting任务队列名让单个 Temporal 集群服务所有环境各环境工作流互不干扰生命周期交给 Encore//encore:serviceinitService/Shutdown让 Temporal client 与 worker 的创建、启动、优雅关闭与 Encore 服务的生命周期完全对齐超时与重试是可靠性的灵魂务必为 Activity 设置StartToCloseTimeout等超时选项Temporal 才能依据超时执行重试/补偿策略配置环境感知化善用 Encore 的 CUE 配置与#Meta.Environment.Cloud条件一处配置、全环境生效尽早验证本地用 Temporalite 跑通端到端链路后再部署云端减少云端调试成本。现在你的 Encore 应用已经具备了与 Temporal 深度集成的能力——既拥有 Encore 带来的极简基础设施体验也拥有 Temporal 提供的高可靠工作流编排能力。【免费下载链接】encoreThe infrastructure platform for the intelligence era项目地址: https://gitcode.com/GitHub_Trending/encor/encore创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考