Temporal 分布式工作流引擎实战：Worker、Activity、重试语义与生产部署

-- order 表加个 status 列和 next_retry_at
SELECT * FROM orders
WHERE status IN ('paying', 'reserving', 'shipping')
  AND next_retry_at < NOW()
LIMIT 100;

// 伪代码
compensations := []func() error{}
defer func() {
    if err != nil {
        for i := len(compensations) - 1; i >= 0; i-- {
            compensations[i]()
        }
    }
}()

if err = payment.Charge(ctx, orderID); err != nil { return }
compensations = append(compensations, func() error {
    return payment.Refund(ctx, orderID)
})

if err = inventory.Reserve(ctx, orderID); err != nil { return }
compensations = append(compensations, func() error {
    return inventory.Release(ctx, orderID)
})
// ...

func OrderFulfillmentWorkflow(ctx workflow.Context, orderID string) error {
    if err := workflow.ExecuteActivity(ctx, ChargePayment, orderID).Get(ctx, nil); err != nil {
        return err
    }
    if err := workflow.ExecuteActivity(ctx, ReserveInventory, orderID).Get(ctx, nil); err != nil {
        return err
    }
    return workflow.ExecuteActivity(ctx, ShipOrder, orderID).Get(ctx, nil)
}

package activities

import (
    "context"
    "errors"
    "fmt"

    "go.temporal.io/sdk/activity"
)

type OrderActivities struct {
    PaymentClient   PaymentClient
    InventoryClient InventoryClient
    ShippingClient  ShippingClient
}

type PaymentClient interface {
    Charge(ctx context.Context, orderID string, amount int64) (string, error)
    Refund(ctx context.Context, txnID string) error
}

type InventoryClient interface {
    Reserve(ctx context.Context, orderID string, sku string, qty int) (string, error)
    Release(ctx context.Context, reservationID string) error
}

type ShippingClient interface {
    CreateShipment(ctx context.Context, orderID string) (string, error)
    CancelShipment(ctx context.Context, shipmentID string) error
}

// ChargePayment 扣款，返回交易 ID
func (a *OrderActivities) ChargePayment(ctx context.Context, orderID string, amount int64) (string, error) {
    logger := activity.GetLogger(ctx)
    logger.Info("ChargePayment start", "orderID", orderID, "amount", amount)

    txnID, err := a.PaymentClient.Charge(ctx, orderID, amount)
    if err != nil {
        // 业务层确定不该重试的错误，用 NonRetryable 包一层
        if errors.Is(err, ErrInsufficientFunds) {
            return "", NewNonRetryable("insufficient_funds", err)
        }
        return "", fmt.Errorf("charge failed: %w", err)
    }
    return txnID, nil
}

// RefundPayment 补偿：退款
func (a *OrderActivities) RefundPayment(ctx context.Context, txnID string) error {
    activity.GetLogger(ctx).Info("RefundPayment", "txnID", txnID)
    return a.PaymentClient.Refund(ctx, txnID)
}

// ReserveInventory 扣库存
func (a *OrderActivities) ReserveInventory(ctx context.Context, orderID, sku string, qty int) (string, error) {
    logger := activity.GetLogger(ctx)
    logger.Info("ReserveInventory start", "orderID", orderID, "sku", sku, "qty", qty)

    resvID, err := a.InventoryClient.Reserve(ctx, orderID, sku, qty)
    if err != nil {
        if errors.Is(err, ErrOutOfStock) {
            return "", NewNonRetryable("out_of_stock", err)
        }
        return "", fmt.Errorf("reserve failed: %w", err)
    }
    return resvID, nil
}

// ReleaseInventory 补偿：释放库存
func (a *OrderActivities) ReleaseInventory(ctx context.Context, reservationID string) error {
    return a.InventoryClient.Release(ctx, reservationID)
}

// CreateShipment 创建物流单
func (a *OrderActivities) CreateShipment(ctx context.Context, orderID string) (string, error) {
    return a.ShippingClient.CreateShipment(ctx, orderID)
}

// CancelShipment 补偿：取消物流
func (a *OrderActivities) CancelShipment(ctx context.Context, shipmentID string) error {
    return a.ShippingClient.CancelShipment(ctx, shipmentID)
}

var (
    ErrInsufficientFunds = errors.New("insufficient_funds")
    ErrOutOfStock        = errors.New("out_of_stock")
)

package activities

import "go.temporal.io/sdk/temporal"

// NewNonRetryable 把一个业务错误包成 Temporal 的非重试错误
func NewNonRetryable(code string, cause error) error {
    return temporal.NewNonRetryableApplicationError(cause.Error(), code, nil)
}

package workflows

import (
    "time"

    "go.temporal.io/sdk/temporal"
    "go.temporal.io/sdk/workflow"

    "example.com/orders/activities"
)

// OrderRequest 工作流入参
type OrderRequest struct {
    OrderID string
    UserID  string
    SKU     string
    Qty     int
    Amount  int64 // 分
}

// OrderResult 工作流返回值
type OrderResult struct {
    OrderID    string
    TxnID      string
    ShipmentID string
}

// OrderFulfillmentWorkflow 订单履约
func OrderFulfillmentWorkflow(ctx workflow.Context, req OrderRequest) (*OrderResult, error) {
    logger := workflow.GetLogger(ctx)
    logger.Info("OrderFulfillment start", "orderID", req.OrderID)

    // Activity 通用选项
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    time.Minute,
            MaximumAttempts:    5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    var a *activities.OrderActivities // 运行时由 worker 注册，nil 只是用来引用方法名

    // Step 1: 扣款
    var txnID string
    if err := workflow.ExecuteActivity(ctx, a.ChargePayment, req.OrderID, req.Amount).Get(ctx, &txnID); err != nil {
        return nil, err
    }

    // Step 2: 扣库存；失败必须回滚扣款
    var resvID string
    if err := workflow.ExecuteActivity(ctx, a.ReserveInventory, req.OrderID, req.SKU, req.Qty).Get(ctx, &resvID); err != nil {
        _ = workflow.ExecuteActivity(ctx, a.RefundPayment, txnID).Get(ctx, nil)
        return nil, err
    }

    // Step 3: 创建物流单；失败要回滚库存和扣款
    var shipmentID string
    if err := workflow.ExecuteActivity(ctx, a.CreateShipment, req.OrderID).Get(ctx, &shipmentID); err != nil {
        _ = workflow.ExecuteActivity(ctx, a.ReleaseInventory, resvID).Get(ctx, nil)
        _ = workflow.ExecuteActivity(ctx, a.RefundPayment, txnID).Get(ctx, nil)
        return nil, err
    }

    logger.Info("OrderFulfillment done", "orderID", req.OrderID)
    return &OrderResult{
        OrderID:    req.OrderID,
        TxnID:      txnID,
        ShipmentID: shipmentID,
    }, nil
}

package main

import (
    "log"
    "os"
    "os/signal"
    "syscall"

    "go.temporal.io/sdk/client"
    "go.temporal.io/sdk/worker"

    "example.com/orders/activities"
    "example.com/orders/workflows"
)

func main() {
    c, err := client.Dial(client.Options{
        HostPort:  getenv("TEMPORAL_ADDRESS", "temporal-frontend.example.com:7233"),
        Namespace: getenv("TEMPORAL_NAMESPACE", "order"),
    })
    if err != nil {
        log.Fatalf("dial temporal: %v", err)
    }
    defer c.Close()

    // Activity 依赖注入：真实 worker 里 PaymentClient 等是 gRPC stub
    acts := &activities.OrderActivities{
        PaymentClient:   newPaymentClient(),
        InventoryClient: newInventoryClient(),
        ShippingClient:  newShippingClient(),
    }

    w := worker.New(c, "order-fulfillment", worker.Options{
        MaxConcurrentActivityExecutionSize:     200,
        MaxConcurrentWorkflowTaskExecutionSize: 100,
    })

    w.RegisterWorkflow(workflows.OrderFulfillmentWorkflow)
    w.RegisterActivity(acts)

    // 优雅退出
    stop := make(chan os.Signal, 1)
    signal.Notify(stop, syscall.SIGINT, syscall.SIGTERM)
    go func() {
        <-stop
        w.Stop()
    }()

    if err := w.Run(worker.InterruptCh()); err != nil {
        log.Fatalf("worker run: %v", err)
    }
}

func getenv(k, def string) string {
    if v := os.Getenv(k); v != "" {
        return v
    }
    return def
}

package main

import (
    "context"
    "log"

    "go.temporal.io/sdk/client"

    "example.com/orders/workflows"
)

func main() {
    c, err := client.Dial(client.Options{
        HostPort:  "temporal-frontend.example.com:7233",
        Namespace: "order",
    })
    if err != nil {
        log.Fatalf("dial: %v", err)
    }
    defer c.Close()

    req := workflows.OrderRequest{
        OrderID: "ord-20260408-0001",
        UserID:  "u-1001",
        SKU:     "sku-book-a",
        Qty:     1,
        Amount:  9900,
    }

    run, err := c.ExecuteWorkflow(context.Background(),
        client.StartWorkflowOptions{
            ID:        "order-" + req.OrderID,
            TaskQueue: "order-fulfillment",
        },
        workflows.OrderFulfillmentWorkflow, req)
    if err != nil {
        log.Fatalf("start workflow: %v", err)
    }

    var result workflows.OrderResult
    if err := run.Get(context.Background(), &result); err != nil {
        log.Fatalf("workflow failed: %v", err)
    }
    log.Printf("done: %+v", result)
}

// 错：
// id := uuid.NewString()
// now := time.Now()

// 对：
var id string
_ = workflow.SideEffect(ctx, func(ctx workflow.Context) interface{} {
    return uuid.NewString()
}).Get(&id)

now := workflow.Now(ctx)

// sleep
_ = workflow.Sleep(ctx, 30*time.Minute)

// goroutine
workflow.Go(ctx, func(ctx workflow.Context) {
    _ = workflow.ExecuteActivity(ctx, someAct).Get(ctx, nil)
})

func TestOrderWorkflow_Replay(t *testing.T) {
    replayer := worker.NewWorkflowReplayer()
    replayer.RegisterWorkflow(workflows.OrderFulfillmentWorkflow)

    err := replayer.ReplayWorkflowHistoryFromJSONFile(nil, "testdata/order-history.json")
    if err != nil {
        t.Fatalf("replay failed (NDE?): %v", err)
    }
}

RetryPolicy{
    InitialInterval:        time.Second,       // 第一次失败后等多久重试
    BackoffCoefficient:     2.0,               // 每次退避翻倍
    MaximumInterval:        time.Minute,       // 退避上限
    MaximumAttempts:        5,                 // 最多试几次，0 表示无限
    NonRetryableErrorTypes: []string{          // 匹配到这些错误类型直接放弃
        "out_of_stock",
        "insufficient_funds",
    },
}

// 方案 A: 预设 NonRetryable 错误类型，workflow 配 NonRetryableErrorTypes
return "", temporal.NewNonRetryableApplicationError(
    "out of stock for sku "+sku,
    "out_of_stock", // type 字段，匹配 NonRetryableErrorTypes
    nil,
)

// 方案 B: 直接标记 non-retryable
return "", temporal.NewApplicationError(
    "bad request",
    "bad_request",
).(*temporal.ApplicationError) // 需要 cast 设置 nonRetryable

func (a *OrderActivities) BulkExport(ctx context.Context, jobID string) error {
    // 读取上次的 heartbeat details
    var lastOffset int
    if activity.HasHeartbeatDetails(ctx) {
        _ = activity.GetHeartbeatDetails(ctx, &lastOffset)
    }

    for offset := lastOffset; offset < 1_000_000; offset += 1000 {
        if err := processBatch(ctx, offset); err != nil {
            return err
        }
        activity.RecordHeartbeat(ctx, offset)
    }
    return nil
}

workflow.ActivityOptions{
    StartToCloseTimeout: time.Hour,
    HeartbeatTimeout:    30 * time.Second,
}

ao := workflow.ActivityOptions{
    StartToCloseTimeout: 30 * time.Second,  // 单次执行不能超过 30s
    RetryPolicy: &temporal.RetryPolicy{
        InitialInterval:    time.Second,
        BackoffCoefficient: 2.0,
        MaximumInterval:    time.Minute,
        MaximumAttempts:    5,
    },
}

ao := workflow.ActivityOptions{
    StartToCloseTimeout: time.Hour,
    HeartbeatTimeout:    30 * time.Second,
    RetryPolicy: &temporal.RetryPolicy{
        MaximumAttempts: 3,
    },
}

ao := workflow.ActivityOptions{
    StartToCloseTimeout:    time.Minute,
    ScheduleToCloseTimeout: 10 * time.Minute, // 不管重试多少次，总共不超过 10 分钟
}

const SignalCancelOrder = "cancel-order"

func OrderFulfillmentWorkflow(ctx workflow.Context, req OrderRequest) (*OrderResult, error) {
    cancelCh := workflow.GetSignalChannel(ctx, SignalCancelOrder)

    ao := workflow.ActivityOptions{StartToCloseTimeout: 30 * time.Second}
    ctx = workflow.WithActivityOptions(ctx, ao)

    var a *activities.OrderActivities

    // Step 1: 扣款
    var txnID string
    if err := workflow.ExecuteActivity(ctx, a.ChargePayment, req.OrderID, req.Amount).Get(ctx, &txnID); err != nil {
        return nil, err
    }

    // 等库存扣减，同时接受取消 signal
    var resvID string
    stockFuture := workflow.ExecuteActivity(ctx, a.ReserveInventory, req.OrderID, req.SKU, req.Qty)

    sel := workflow.NewSelector(ctx)
    var canceled bool
    sel.AddReceive(cancelCh, func(c workflow.ReceiveChannel, more bool) {
        var reason string
        c.Receive(ctx, &reason)
        canceled = true
    })
    sel.AddFuture(stockFuture, func(f workflow.Future) {
        _ = f.Get(ctx, &resvID)
    })
    sel.Select(ctx)

    if canceled {
        _ = workflow.ExecuteActivity(ctx, a.RefundPayment, txnID).Get(ctx, nil)
        return nil, temporal.NewApplicationError("order canceled", "canceled")
    }

    // ... 继续后续步骤
    return nil, nil
}

_ = c.SignalWorkflow(context.Background(),
    "order-ord-20260408-0001", // workflow ID
    "",                         // run ID 留空 = 当前 run
    SignalCancelOrder,
    "user requested")

const QueryStatus = "status"

type OrderStatus struct {
    Step    string
    TxnID   string
    ResvID  string
}

func OrderFulfillmentWorkflow(ctx workflow.Context, req OrderRequest) (*OrderResult, error) {
    status := &OrderStatus{Step: "init"}

    if err := workflow.SetQueryHandler(ctx, QueryStatus, func() (*OrderStatus, error) {
        return status, nil
    }); err != nil {
        return nil, err
    }

    // 后面每推进一步更新 status.Step
    status.Step = "charging"
    // ...
    return nil, nil
}

resp, _ := c.QueryWorkflow(ctx, "order-ord-xxx", "", QueryStatus)
var status workflows.OrderStatus
_ = resp.Get(&status)
fmt.Println(status.Step)

cwo := workflow.ChildWorkflowOptions{
    WorkflowID: "shipment-" + req.OrderID,
    TaskQueue:  "shipping",
}
ctx = workflow.WithChildOptions(ctx, cwo)
var shipmentID string
if err := workflow.ExecuteChildWorkflow(ctx, ShippingWorkflow, req.OrderID).Get(ctx, &shipmentID); err != nil {
    return nil, err
}

func SubscriptionWorkflow(ctx workflow.Context, state SubState) error {
    for i := 0; i < 12; i++ { // 一年循环 12 次就换 run
        _ = workflow.Sleep(ctx, 30*24*time.Hour)
        _ = workflow.ExecuteActivity(ctx, ChargeMonthly, state.UserID).Get(ctx, nil)
        state.MonthsPaid++
    }
    // 开新 run，带上最新状态
    return workflow.NewContinueAsNewError(ctx, SubscriptionWorkflow, state)
}

client.StartWorkflowOptions{
    ID:                    "order-" + orderID,
    TaskQueue:             "order-fulfillment",
    WorkflowIDReusePolicy: enums.WORKFLOW_ID_REUSE_POLICY_REJECT_DUPLICATE,
}

func (a *OrderActivities) ChargePayment(ctx context.Context, orderID string, amount int64) (string, error) {
    // 用 orderID 做幂等键
    idempotencyKey := "charge-" + orderID
    return a.PaymentClient.ChargeWithKey(ctx, idempotencyKey, amount)
}

package workflows

import (
    "time"

    "go.temporal.io/sdk/temporal"
    "go.temporal.io/sdk/workflow"

    "example.com/orders/activities"
)

// compensation 是一个待执行的补偿动作
type compensation struct {
    name string
    fn   func(ctx workflow.Context) error
}

// saga 收集补偿链
type saga struct {
    comps []compensation
}

func (s *saga) add(name string, fn func(ctx workflow.Context) error) {
    s.comps = append(s.comps, compensation{name, fn})
}

// compensate 从后往前执行所有补偿；单个失败继续执行下一个
func (s *saga) compensate(ctx workflow.Context) {
    logger := workflow.GetLogger(ctx)
    for i := len(s.comps) - 1; i >= 0; i-- {
        c := s.comps[i]
        logger.Info("compensate", "name", c.name)
        if err := c.fn(ctx); err != nil {
            logger.Error("compensate failed", "name", c.name, "err", err)
        }
    }
}

func OrderFulfillmentSagaWorkflow(ctx workflow.Context, req OrderRequest) (*OrderResult, error) {
    ao := workflow.ActivityOptions{
        StartToCloseTimeout: 30 * time.Second,
        RetryPolicy: &temporal.RetryPolicy{
            InitialInterval:    time.Second,
            BackoffCoefficient: 2.0,
            MaximumInterval:    time.Minute,
            MaximumAttempts:    5,
        },
    }
    ctx = workflow.WithActivityOptions(ctx, ao)

    s := &saga{}
    var a *activities.OrderActivities

    defer func() {
        if r := recover(); r != nil {
            s.compensate(ctx)
            panic(r)
        }
    }()

    // Step 1: 扣款
    var txnID string
    if err := workflow.ExecuteActivity(ctx, a.ChargePayment, req.OrderID, req.Amount).Get(ctx, &txnID); err != nil {
        return nil, err
    }
    s.add("refund", func(ctx workflow.Context) error {
        return workflow.ExecuteActivity(ctx, a.RefundPayment, txnID).Get(ctx, nil)
    })

    // Step 2: 扣库存
    var resvID string
    if err := workflow.ExecuteActivity(ctx, a.ReserveInventory, req.OrderID, req.SKU, req.Qty).Get(ctx, &resvID); err != nil {
        s.compensate(ctx)
        return nil, err
    }
    s.add("release-inventory", func(ctx workflow.Context) error {
        return workflow.ExecuteActivity(ctx, a.ReleaseInventory, resvID).Get(ctx, nil)
    })

    // Step 3: 创建物流单
    var shipmentID string
    if err := workflow.ExecuteActivity(ctx, a.CreateShipment, req.OrderID).Get(ctx, &shipmentID); err != nil {
        s.compensate(ctx)
        return nil, err
    }
    s.add("cancel-shipment", func(ctx workflow.Context) error {
        return workflow.ExecuteActivity(ctx, a.CancelShipment, shipmentID).Get(ctx, nil)
    })

    return &OrderResult{
        OrderID:    req.OrderID,
        TxnID:      txnID,
        ShipmentID: shipmentID,
    }, nil
}

// 在 ChargePayment 前面加
_ = workflow.ExecuteActivity(ctx, a.RiskCheck, req.OrderID).Get(ctx, nil)

v := workflow.GetVersion(ctx, "add-risk-check", workflow.DefaultVersion, 1)
if v == 1 {
    if err := workflow.ExecuteActivity(ctx, a.RiskCheck, req.OrderID).Get(ctx, nil); err != nil {
        return nil, err
    }
}

// 后续 ChargePayment 保持不变

v := workflow.GetVersion(ctx, "add-risk-check", workflow.DefaultVersion, 2)
if v >= 1 { /* ... */ }
if v == 2 { /* v2 的新逻辑 */ }

helm repo add temporal https://go.temporal.io/helm-charts
helm repo update

helm install temporal temporal/temporal \
  --namespace temporal \
  --create-namespace \
  --values values.yaml

server:
  replicaCount: 3
  config:
    persistence:
      default:
        driver: "sql"
        sql:
          driver: "postgres12"
          host: "pg.example.com"
          port: 5432
          database: "temporal"
          user: "temporal"
          existingSecret: "temporal-db-secret"
          maxConns: 50
          maxConnLifetime: "1h"
      visibility:
        driver: "elasticsearch"
        elasticsearch:
          version: "v7"
          url:
            scheme: "https"
            host: "es.example.com:9200"
          indices:
            visibility: "temporal_visibility_v1_prod"
    numHistoryShards: 512

cassandra:
  enabled: false
elasticsearch:
  enabled: false  # 用外部 ES
prometheus:
  enabled: true
grafana:
  enabled: true

web:
  replicaCount: 2
  ingress:
    enabled: true
    hosts:
      - temporal-ui.example.com

frontend:
  replicas: 3
  resources:
    requests:
      cpu: 500m
      memory: 1Gi
    limits:
      cpu: 2
      memory: 4Gi

history:
  replicas: 3
  resources:
    requests:
      cpu: 1
      memory: 2Gi
    limits:
      cpu: 4
      memory: 8Gi

matching:
  replicas: 3
  resources:
    requests:
      cpu: 500m
      memory: 1Gi
    limits:
      cpu: 2
      memory: 4Gi

worker:
  replicas: 2
  resources:
    requests:
      cpu: 200m
      memory: 512Mi
    limits:
      cpu: 1
      memory: 2Gi

_, _ = c.ScheduleClient().Create(ctx, client.ScheduleOptions{
    ID: "daily-reconcile",
    Spec: client.ScheduleSpec{
        CronExpressions: []string{"0 2 * * *"},
    },
    Action: &client.ScheduleWorkflowAction{
        ID:        "reconcile",
        Workflow:  ReconcileWorkflow,
        TaskQueue: "reconcile",
    },
})