分布式任务调度与工作流编排完全指南 2026：Temporal vs Asynq vs XXL-JOB + 生产自愈实战

1
package main
2

3
import (
4
  "context"
5
  "fmt"
6
)
7

8
type OrderActivities struct{}
9

10
func (a *OrderActivities) CheckPaymentStatus(ctx context.Context, orderID string) (bool, error) {
11
  fmt.Printf("🔍 Checking payment status for order: %s\n", orderID)
12
  // 模拟查询支付网关
13
  return false, nil // 模拟用户尚未支付
14
}
15

16
func (a *OrderActivities) CancelOrder(ctx context.Context, orderID string) error {
17
  fmt.Printf("❌ Auto cancelling expired order: %s, releasing inventory...\n", orderID)
18
  // 执行库存解冻与订单状态取消
19
  return nil
20
}

1
package main
2

3
import (
4
  "time"
5
  "go.temporal.io/sdk/temporal"
6
  "go.temporal.io/sdk/workflow"
7
)
8

9
func OrderFulfillmentWorkflow(ctx workflow.Context, orderID string) error {
10
  // 1. 配置 Activity 自动重试策略与超时
11
  ao := workflow.ActivityOptions{
12
    StartToCloseTimeout: 10 * time.Second,
13
    RetryPolicy: &temporal.RetryPolicy{
14
      InitialInterval:    time.Second,
15
      BackoffCoefficient: 2.0,
16
      MaximumAttempts:    5,
17
    },
18
  }
19
  ctx = workflow.WithActivityOptions(ctx, ao)
20

21
  var activities *OrderActivities
22

23
  // 2. 核心：原生休眠 30 分钟（等待用户支付，期间不占用任何机器内存！）
24
  workflow.GetLogger(ctx).Info("Waiting 30 minutes for user payment...", "OrderID", orderID)
25
  _ = workflow.Sleep(ctx, 30*time.Minute)
26

27
  // 3. 30 分钟后唤醒：调用 Activity 检查支付状态
28
  var isPaid bool
29
  err := workflow.ExecuteActivity(ctx, activities.CheckPaymentStatus, orderID).Get(ctx, &isPaid)
30
  if err != nil {
31
    return err
32
  }
33

34
  // 4. 若未支付，执行自动取消
35
  if !isPaid {
36
    err = workflow.ExecuteActivity(ctx, activities.CancelOrder, orderID).Get(ctx, nil)
37
    if err != nil {
38
      return err
39
    }
40
    workflow.GetLogger(ctx).Info("Order cancelled successfully due to timeout", "OrderID", orderID)
41
  }
42

43
  return nil
44
}

1
package main
2

3
import (
4
  "context"
5
  "log"
6

7
  "go.temporal.io/sdk/client"
8
  "go.temporal.io/sdk/worker"
9
)
10

11
const TaskQueue = "ORDER_TASK_QUEUE"
12

13
func main() {
14
  // 1. 连接 Temporal 服务端集群
15
  c, err := client.Dial(client.Options{
16
    HostPort: "localhost:7233",
17
  })
18
  if err != nil {
19
    log.Fatalln("Unable to create Temporal client", err)
20
  }
21
  defer c.Close()
22

23
  // 2. 注册并启动 Worker 监听任务队列
24
  w := worker.New(c, TaskQueue, worker.Options{})
25
  w.RegisterWorkflow(OrderFulfillmentWorkflow)
26
  w.RegisterActivity(&OrderActivities{})
27

28
  go func() {
29
    err = w.Run(worker.InterruptCh())
30
    if err != nil {
31
      log.Fatalln("Unable to start Worker", err)
32
    }
33
  }()
34

35
  // 3. 业务触发：发起一个长生命周期工作流实例
36
  workflowOptions := client.StartWorkflowOptions{
37
    ID:        "order-wf-20260920-8888", // 业务唯一 ID，天然防重复幂等！
38
    TaskQueue: TaskQueue,
39
  }
40

41
  we, err := c.ExecuteWorkflow(context.Background(), workflowOptions, OrderFulfillmentWorkflow, "ORD-998811")
42
  if err != nil {
43
    log.Fatalln("Unable to execute workflow", err)
44
  }
45

46
  log.Printf("🚀 Workflow started successfully! RunID: %s\n", we.GetRunID())
47
  select {}
48
}