1
package saga
2

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

9
// Step 定义 saga 的一个步骤
10
type Step struct {
11
  Name        string                                 // 步骤名称
12
  Execute     func(ctx context.Context) error        // 正向操作
13
  Compensate func(ctx context.Context) error         // 补偿操作
14
}
15

16
// Saga 协调者
17
type Saga struct {
18
  name  string
19
  steps []Step
20
}
21

22
func NewSaga(name string, steps []Step) *Saga {
23
  return &Saga{name: name, steps: steps}
24
}
25

26
// Execute 执行 saga：顺序执行所有步骤，失败则逆序补偿
27
func (s *Saga) Execute(ctx context.Context) error {
28
  // 记录已完成的步骤，用于补偿
29
  completed := make([]int, 0, len(s.steps))
30

31
  for i, step := range s.steps {
32
    log.Printf("[Saga:%s] 执行步骤 %d: %s", s.name, i+1, step.Name)
33
    if err := step.Execute(ctx); err != nil {
34
      log.Printf("[Saga:%s] 步骤 %d 失败: %v，开始补偿", s.name, i+1, err)
35
      s.compensate(ctx, completed)
36
      return fmt.Errorf("步骤 %q 失败: %w，已执行补偿", step.Name, err)
37
    }
38
    completed = append(completed, i)
39
  }
40

41
  log.Printf("[Saga:%s] 所有步骤完成", s.name)
42
  return nil
43
}
44

45
// compensate 逆序执行已完成步骤的补偿操作
46
func (s *Saga) compensate(ctx context.Context, completed []int) {
47
  // 逆序遍历
48
  for i := len(completed) - 1; i >= 0; i-- {
49
    idx := completed[i]
50
    step := s.steps[idx]
51
    log.Printf("[Saga:%s] 补偿步骤 %d: %s", s.name, idx+1, step.Name)
52

53
    // 补偿操作必须幂等，重试直到成功或上下文取消
54
    if err := step.Compensate(ctx); err != nil {
55
      log.Printf("[Saga:%s] 补偿步骤 %d 失败: %v，需要人工介入", s.name, idx+1, err)
56
    }
57
  }
58
}

1
package main
2

3
import (
4
  "context"
5
  "fmt"
6
  "log"
7
  "saga"
8
)
9

10
func main() {
11
  orderSaga := saga.NewSaga("创建订单", []saga.Step{
12
    {
13
      Name: "创建订单",
14
      Execute: func(ctx context.Context) error {
15
        log.Println("创建订单：写入订单表")
16
        return nil
17
      },
18
      Compensate: func(ctx context.Context) error {
19
        log.Println("取消订单：标记订单为已取消")
20
        return nil
21
      },
22
    },
23
    {
24
      Name: "扣减库存",
25
      Execute: func(ctx context.Context) error {
26
        log.Println("扣减库存：库存 -1")
27
        return nil
28
      },
29
      Compensate: func(ctx context.Context) error {
30
        log.Println("恢复库存：库存 +1")
31
        return nil
32
      },
33
    },
34
    {
35
      Name: "扣款支付",
36
      Execute: func(ctx context.Context) error {
37
        log.Println("扣款支付：余额不足，扣款失败")
38
        return fmt.Errorf("余额不足")
39
      },
40
      Compensate: func(ctx context.Context) error {
41
        log.Println("退款支付：无需退款，支付未成功")
42
        return nil
43
      },
44
    },
45
  })
46

47
  if err := orderSaga.Execute(context.Background()); err != nil {
48
    log.Printf("Saga 执行失败: %v", err)
49
  }
50
}

1
// 编排模式（Choreography）：事件驱动的 Saga
2

3
type EventHandler = (data: unknown) => Promise<void>;
4

5
// 简单的事件总线
6
class EventBus {
7
  private handlers = new Map<string, EventHandler[]>();
8

9
  // 注册事件处理器
10
  on(event: string, handler: EventHandler): void {
11
    const list = this.handlers.get(event) ?? [];
12
    list.push(handler);
13
    this.handlers.set(event, list);
14
  }
15

16
  // 发布事件，触发所有注册的处理器
17
  async emit(event: string, data: unknown): Promise<void> {
18
    const handlers = this.handlers.get(event) ?? [];
19
    for (const handler of handlers) {
20
      await handler(data);
21
    }
22
  }
23
}
24

25
// 订单服务
26
class OrderService {
27
  constructor(private bus: EventBus) {
28
    // 监听库存恢复事件，执行补偿
29
    this.bus.on("InventoryRestored", async (data: unknown) => {
30
      const { orderId } = data as { orderId: string };
31
      console.log(`取消订单 ${orderId}：标记为已取消`);
32
    });
33
  }
34

35
  async createOrder(orderId: string): Promise<void> {
36
    console.log(`创建订单 ${orderId}：写入订单表`);
37
    // 发布订单创建事件，触发下游
38
    await this.bus.emit("OrderCreated", { orderId });
39
  }
40
}
41

42
// 库存服务
43
class InventoryService {
44
  constructor(private bus: EventBus) {
45
    // 监听订单创建事件
46
    this.bus.on("OrderCreated", async (data: unknown) => {
47
      const { orderId } = data as { orderId: string };
48
      console.log(`扣减库存：订单 ${orderId} 库存 -1`);
49
      await this.bus.emit("InventoryDeducted", { orderId });
50
    });
51

52
    // 监听支付失败事件，执行补偿
53
    this.bus.on("PaymentFailed", async (data: unknown) => {
54
      const { orderId } = data as { orderId: string };
55
      console.log(`恢复库存：订单 ${orderId} 库存 +1`);
56
      await this.bus.emit("InventoryRestored", { orderId });
57
    });
58
  }
59
}
60

61
// 支付服务
62
class PaymentService {
63
  constructor(private bus: EventBus) {
64
    // 监听库存扣减事件
65
    this.bus.on("InventoryDeducted", async (data: unknown) => {
66
      const { orderId } = data as { orderId: string };
67
      console.log(`扣款支付：订单 ${orderId} 余额不足，扣款失败`);
68
      // 支付失败，发布失败事件触发补偿链
69
      await this.bus.emit("PaymentFailed", { orderId });
70
    });
71
  }
72
}
73

74
// 使用示例
75
async function main(): Promise<void> {
76
  const bus = new EventBus();
77
  const orderService = new OrderService(bus);
78
  new InventoryService(bus); // 注册事件处理器
79
  new PaymentService(bus);
80

81
  // 从订单服务开始，事件链自动驱动
82
  await orderService.createOrder("ORD-001");
83
}