temporal-golang-pro

Temporal Go SDK（temporal-golang-pro）

概述

这是一份专家级指南，介绍如何使用 Temporal Go SDK 构建弹性、可扩展且确定性的分布式系统。这项技能可将模糊的编排需求转化为 Go 实现草案，但仍需针对具体项目进行验证，重点关注持久化执行、严格确定性以及企业级 Worker 配置。

何时使用此技能

• 设计分布式系统：构建需要持久化状态和可靠编排的微服务时。
• 实现复杂工作流：使用 Go SDK 处理长时间运行的流程（数天或数月），或复杂的 Saga 模式时。
• 优化性能：Worker 需要精细调整并发、mTLS 安全性或自定义拦截器时。
• 确保可靠性：实现幂等 Activity、优雅的错误处理和复杂的重试策略时。
• 维护与演进：为正在运行的工作流进行版本管理，或在不停机的情况下更新 Worker 时。

以下情况请勿使用此技能

• 使用其他 SDK（Python、Java、TypeScript）与 Temporal 交互时，请参考相应的 -pro 技能。
• 任务是简单的请求/响应，不涉及持久化或协调需求时。
• 只需高层设计而无需实现时（请使用 workflow-orchestration-patterns）。

分步指南

1. 收集上下文：主动询问：
   • 目标 Temporal 集群（Cloud 或自托管）和 命名空间。
   • 任务队列名称和预期吞吐量。
   • 安全要求（mTLS 路径、身份验证）。
   • 故障模式和期望的重试/超时策略。
2. 验证确定性：在提出工作流代码之前，根据以下 5 条规则进行验证：
   • 不使用 Go 原生并发（goroutine）。
   • 不使用原生时间函数（time.Now、time.Sleep）。
   • 不使用非确定性的 map 遍历（必须对键排序）。
   • 不进行直接外部 I/O 或网络调用。
   • 不使用非确定性的随机数。
3. 逐步实现：先实现共享的 Protobuf/数据类，再实现 Activity，然后是工作流，最后是 Worker。
4. 利用资源：如果实现需要高级模式（Saga、拦截器、Replay 测试），请明确参考实现手册和测试策略。

功能

Go SDK 实现

• Worker 管理：深入了解 worker.Options，包括 MaxConcurrentActivityTaskPollers、WorkerStopTimeout 和 StickyScheduleToStartTimeout。
• 拦截器：实现客户端、Worker 和工作流拦截器，以处理横切关注点（日志记录、追踪、身份验证）。
• 自定义数据转换器：集成 Protobuf、加密负载或自定义 JSON 编组。

高级工作流模式

• 持久化并发：使用 workflow.Go、workflow.Channel 和 workflow.Selector，而不是原生并发原语。
• 版本管理：使用 workflow.GetVersion 和 workflow.GetLogger 安全地演进代码。
• 大规模处理：使用 ContinueAsNew 管理历史记录大小限制的模式（默认值：50MB 或 50K 个事件）。
• 子工作流：管理生命周期、取消操作以及父子信号传播。

测试与可观测性

• 精通测试套件：使用 WorkflowTestSuite 进行单元和功能测试，并确定性地控制时间。
• 模拟：采用复杂的 Activity 和子工作流模拟策略。
• Replay 测试：根据生产环境的事件历史记录验证代码更改。
• 指标：配置 Prometheus/OpenTelemetry 导出器，以跟踪 Worker 性能。

示例

示例 1：版本化工作流（确定性）

// Note: imports omitted. Requires 'go.temporal.io/sdk/workflow', 'go.temporal.io/sdk/temporal', and 'time'.
func SubscriptionWorkflow(ctx workflow.Context, userID string) error {
    // 1. Versioning for logic evolution (v1 = DefaultVersion)
    v := workflow.GetVersion(ctx, "billing_logic", workflow.DefaultVersion, 2)

    for i := 0; i < 12; i++ {
        ao := workflow.ActivityOptions{
            StartToCloseTimeout: 5 * time.Minute,
            RetryPolicy: &temporal.RetryPolicy{MaximumAttempts: 3},
        }
        ctx = workflow.WithActivityOptions(ctx, ao)

        // 2. Activity Execution (Always handle errors)
        err := workflow.ExecuteActivity(ctx, ChargePaymentActivity, userID).Get(ctx, nil)
        if err != nil {
            workflow.GetLogger(ctx).Error("Payment failed", "Error", err)
            return err
        }

        // 3. Durable Sleep (Time-skipping safe)
        sleepDuration := 30 * 24 * time.Hour
        if v >= 2 {
            sleepDuration = 28 * 24 * time.Hour
        }

        if err := workflow.Sleep(ctx, sleepDuration); err != nil {
            return err
        }
    }
    return nil
}

示例 2：完整的 mTLS Worker 配置

func RunSecureWorker() error {
    // 1. Load Client Certificate and Key
    cert, err := tls.LoadX509KeyPair("client.pem", "client.key")
    if err != nil {
        return fmt.Errorf("failed to load client keys: %w", err)
    }

    // 2. Load CA Certificate for Server verification (Proper mTLS)
    caPem, err := os.ReadFile("ca.pem")
    if err != nil {
        return fmt.Errorf("failed to read CA cert: %w", err)
    }
    certPool := x509.NewCertPool()
    if !certPool.AppendCertsFromPEM(caPem) {
        return fmt.Errorf("failed to parse CA cert")
    }

    // 3. Dial Cluster with full TLS config
    c, err := client.Dial(client.Options{
        HostPort:  "temporal.example.com:7233",
        Namespace: "production",
        ConnectionOptions: client.ConnectionOptions{
            TLS: &tls.Config{
                Certificates: []tls.Certificate{cert},
                RootCAs:      certPool,
            },
        },
    })
    if err != nil {
        return fmt.Errorf("failed to dial temporal: %w", err)
    }
    defer c.Close()

    w := worker.New(c, "payment-queue", worker.Options{})
    w.RegisterWorkflow(SubscriptionWorkflow)
    w.RegisterActivity(ChargePaymentActivity) // implement with a stable per-charge idempotency key

    if err := w.Run(worker.InterruptCh()); err != nil {
        return fmt.Errorf("worker run failed: %w", err)
    }
    return nil
}

示例 3：Selector 与信号集成

func ApprovalWorkflow(ctx workflow.Context) (string, error) {
    var approved bool
    signalCh := workflow.GetSignalChannel(ctx, "approval-signal")

    // Use Selector to wait for multiple async events
    s := workflow.NewSelector(ctx)
    s.AddReceive(signalCh, func(c workflow.ReceiveChannel, _ bool) {
        c.Receive(ctx, &approved)
    })

    // Add 72-hour timeout timer
    s.AddFuture(workflow.NewTimer(ctx, 72*time.Hour), func(f workflow.Future) {
        approved = false
    })

    s.Select(ctx)

    if !approved {
        return "rejected", nil
    }
    return "approved", nil
}

最佳实践

• ✅ 应该： 始终处理 ExecuteActivity 和 client.Dial 返回的错误。
• ✅ 应该： 使用 workflow.Go 和 workflow.Channel 实现并发。
• ✅ 应该： 迭代前对映射键排序，以保持确定性。
• ✅ 应该： 对耗时超过 1 分钟的活动使用 activity.RecordHeartbeat。
• ✅ 应该： 使用 replayer.ReplayWorkflowHistoryFromJSON 测试逻辑兼容性。
• ❌ 不要： 在生产工作进程中用 _ 或 log.Fatal 忽略错误。
• ❌ 不要： 在工作流函数中直接执行网络／磁盘 I/O。
• ❌ 不要： 依赖原生 time.Now() 或 rand.Int()。
• ❌ 不要： 将此方案用于不需要持久性的简单 cron 任务。

故障排除

• Panic：确定性不匹配：通常由修改逻辑时未使用 workflow.GetVersion 或使用非确定性代码（例如原生映射）导致。
• 错误：超出历史记录大小限制：已达到历史记录上限（默认 50K 个事件）。请确保已实现 ContinueAsNew。
• 工作进程挂起：检查 WorkerStopTimeout，并确保所有活动都能处理上下文取消。

限制

• 不涵盖 Temporal Cloud UI 导航或 TLS 证书配置工作流。
• 不涵盖 Temporal Java、Python 或 TypeScript SDK；请参阅它们专门的 -pro 技能。
• 假设使用 Temporal Server v1.20+ 和 Go SDK v1.25+；较旧的 SDK 版本可能有不同的 API。
• 不涵盖 Nexus 或多集群复制；请查阅固定版本的 SDK／服务端文档，确认当前支持情况。
• 不涉及全局命名空间配置或多区域故障转移设置。
• 不涵盖当前的 Worker Deployment／Versioning API；请根据固定版本的文档选择升级策略。

资源

• 实现手册 - 深入介绍 Go SDK 模式。
• 测试策略 - Go 的单元测试、重放测试和集成测试。
• Temporal Go SDK 参考
• Temporal Go 示例

相关技能

• grpc-golang - 内部传输协议与 Protobuf 设计。
• golang-pro - 通用 Go 性能调优与高级语法。
• workflow-orchestration-patterns - 与语言无关的编排策略。

生产环境注意事项

示例省略了导入、活动类型和持久化存储，因此并非可编译的应用程序。请注册所有引用的 Activity。计费循环使用 28／30 天的时长，而非日历月；请将明确的逻辑扣费 ID 纳入原子化的服务商／数据库幂等契约，以确保重试不会重复扣费，且不同周期不会被合并。Saga 需要设计补偿机制，且无法保证回滚任意外部操作。请配置 HeartbeatTimeout，以便基于心跳进行检测／取消，并区分工作进程本地速率限制与任务队列级速率限制。取消未使用的计时器，并将工作流取消与拒绝分别处理。固定 SDK／服务端版本，并在迁移前核实 API、部署／版本保留策略、重置／重放需求以及实际历史记录限制；仅凭没有正在运行的实例，不足以舍弃所有重放兼容性。请对敏感历史记录负载进行加密并尽量减少其内容，同时保护导出的历史记录／证书；切勿禁用 TLS 验证。