Go语言的分布式事务处理

1. 分布式事务简介

分布式系统里，事务处理的门槛一下子就被拉高了。过去单机环境下，数据库的ACID特性就能把一致性安排得明明白白；可一旦上了分布式，网络延迟、节点故障这些变量一掺和，保证多个服务之间的数据一致性就成了一个不小的挑战。

分布式事务的特点

• 跨服务：一个事务往往要牵扯多个服务或者数据库
• 网络依赖：通讯靠网络，延迟和故障就成了家常便饭
• 数据一致性：目标是要让多个服务的数据最终能对齐
• 复杂性：实现难度不低，各种异常场景都得提前想到

分布式事务的挑战

• 网络分区：网络一断，部分节点直接失联
• 节点故障：参与方的机器说不准什么时候就宕了
• 消息丢失：网络传输中信息可能半路失踪
• 并发冲突：多个事务同时操作同一份数据，容易“打架”
• 性能问题：分布式事务的协调成本，往往会影响系统吞吐

2. 常见的分布式事务解决方案

2PC（两阶段提交）

2PC算得上是分布式事务里的元老级方案了。核心思路是把提交过程拆成两个阶段：先准备，再提交。

原理：

1. 准备阶段：协调者向所有参与者发“准备”指令，大家各自执行操作但不提交
2. 提交阶段：如果所有参与者都反馈“准备成功”，协调者就发“提交”指令；否则发“回滚”指令

特点：

• 强一致性
• 阻塞式协议，性能瓶颈明显
• 单点故障问题——协调者一旦挂掉，全部卡住

3PC（三阶段提交）

3PC是对2PC的优化，核心改动是增加了一个“准备提交”阶段，减少阻塞时间。

原理：

1. CanCommit：协调者先问问参与者：你能执行这个事务吗？
2. PreCommit：确认大家都“能”之后，发预提交指令，参与者执行但不提交
3. DoCommit：最后发提交指令，参与者真正提交

特点：

• 阻塞时间缩短了
• 可用性比2PC高一些
• 但性能问题依然存在

TCC（Try-Confirm-Cancel）

TCC走的是业务逻辑的路线。它把事务拆成三个动作：尝试、确认、取消。

原理：

1. Try：尝试执行业务操作，先把资源预留好
2. Confirm：确认执行，正式使用预留的资源
3. Cancel：取消操作，释放预留的资源

特点：

• 业务侵入性强——每个服务都得实现Try/Confirm/Cancel
• 好处是可以根据具体业务场景做精细控制
• 性能表现较好

Saga模式

Saga是一种基于事件的思路，把一个大事务拆成一连串的小事务，每个小事务都有对应的补偿操作。

原理：

1. 定义一系列本地事务，每个事务配一个“反操作”
2. 按顺序执行本地事务，一旦某个失败，就反向执行补偿操作

特点：

• 追求的是最终一致性
• 性能不错
• 实现复杂度偏高

本地消息表

这个方案把业务操作和消息写入放在同一个本地事务里，然后依赖消息队列来保证最终一致性。

原理：

1. 业务操作和消息写入在同一个数据库事务中完成
2. 消息队列消费消息，触发远程操作
3. 如果远程操作失败，通过重试确保它最终被执行

特点：

• 最终一致性
• 实现简单，门槛低
• 依赖消息队列的可靠性

基于消息队列的最终一致性

更直接一点——干脆直接用消息队列来实现最终一致性，适合那些对一致性要求不高的场景。

原理：

1. 生产者把消息扔到队列里
2. 消费者消费并执行操作
3. 通过重试机制确保消息被处理

特点：

• 最终一致性
• 实现简单
• 但要注意消息重复消费的问题

3. 2PC协议的Go语言实现

基本架构

package main

import (
	"errors"
	"fmt"
	"sync"
)

// Participant 事务参与者
type Participant interface {
	Prepare() error
	Commit() error
	Rollback() error
}

// Coordinator 事务协调者
type Coordinator struct {
	participants []Participant
}

// NewCoordinator 创建新的协调者
func NewCoordinator(participants []Participant) *Coordinator {
	return &Coordinator{
		participants: participants,
	}
}

// Execute 执行两阶段提交
func (c *Coordinator) Execute() error {
	// 第一阶段：准备
	if err := c.prepare(); err != nil {
		// 准备失败，执行回滚
		c.rollback()
		return err
	}

	// 第二阶段：提交
	if err := c.commit(); err != nil {
		// 提交失败，执行回滚
		c.rollback()
		return err
	}

	return nil
}

// prepare 准备阶段
func (c *Coordinator) prepare() error {
	for _, p := range c.participants {
		if err := p.Prepare(); err != nil {
			return err
		}
	}
	return nil
}

// commit 提交阶段
func (c *Coordinator) commit() error {
	for _, p := range c.participants {
		if err := p.Commit(); err != nil {
			return err
		}
	}
	return nil
}

// rollback 回滚阶段
func (c *Coordinator) rollback() error {
	var wg sync.WaitGroup
	var errMutex sync.Mutex
	var rollbackErr error

	for _, p := range c.participants {
		wg.Add(1)
		go func(participant Participant) {
			defer wg.Done()
			if err := participant.Rollback(); err != nil {
				errMutex.Lock()
				if rollbackErr == nil {
					rollbackErr = err
				}
				errMutex.Unlock()
			}
		}(p)
	}

	wg.Wait()
	return rollbackErr
}

// 示例参与者实现
type ExampleParticipant struct {
	name string
	prepared bool
}

func NewExampleParticipant(name string) *ExampleParticipant {
	return &ExampleParticipant{
		name: name,
	}
}

func (p *ExampleParticipant) Prepare() error {
	fmt.Printf("Participant %s: preparing\n", p.name)
	// 模拟准备操作
	p.prepared = true
	return nil
}

func (p *ExampleParticipant) Commit() error {
	fmt.Printf("Participant %s: committing\n", p.name)
	// 模拟提交操作
	return nil
}

func (p *ExampleParticipant) Rollback() error {
	fmt.Printf("Participant %s: rolling back\n", p.name)
	// 模拟回滚操作
	p.prepared = false
	return nil
}

func main() {
	// 创建参与者
	p1 := NewExampleParticipant("Service A")
	p2 := NewExampleParticipant("Service B")
	p3 := NewExampleParticipant("Service C")

	// 创建协调者
	coordinator := NewCoordinator([]Participant{p1, p2, p3})

	// 执行事务
	if err := coordinator.Execute(); err != nil {
		fmt.Printf("Transaction failed: %v\n", err)
	} else {
		fmt.Println("Transaction succeeded")
	}
}

4. TCC模式的Go语言实现

基本架构

package main

import (
	"errors"
	"fmt"
)

// TCCService TCC服务接口
type TCCService interface {
	Try() error
	Confirm() error
	Cancel() error
}

// TCCManager TCC事务管理器
type TCCManager struct {
	services []TCCService
}

// NewTCCManager 创建新的TCC管理器
func NewTCCManager(services []TCCService) *TCCManager {
	return &TCCManager{
		services: services,
	}
}

// Execute 执行TCC事务
func (m *TCCManager) Execute() error {
	// 执行Try阶段
	if err := m.try(); err != nil {
		// Try失败，执行Cancel
		m.cancel()
		return err
	}

	// 执行Confirm阶段
	if err := m.confirm(); err != nil {
		// Confirm失败，执行Cancel
		m.cancel()
		return err
	}

	return nil
}

// try 执行Try阶段
func (m *TCCManager) try() error {
	for _, service := range m.services {
		if err := service.Try(); err != nil {
			return err
		}
	}
	return nil
}

// confirm 执行Confirm阶段
func (m *TCCManager) confirm() error {
	for _, service := range m.services {
		if err := service.Confirm(); err != nil {
			return err
		}
	}
	return nil
}

// cancel 执行Cancel阶段
func (m *TCCManager) cancel() error {
	for _, service := range m.services {
		if err := service.Cancel(); err != nil {
			// 记录错误但继续执行
			fmt.Printf("Cancel failed for service: %v\n", err)
		}
	}
	return nil
}

// 示例TCC服务实现
type OrderService struct {
	orderID string
	reserved bool
}

func NewOrderService(orderID string) *OrderService {
	return &OrderService{
		orderID: orderID,
	}
}

func (s *OrderService) Try() error {
	fmt.Printf("OrderService: trying to reserve order %s\n", s.orderID)
	// 模拟预留资源
	s.reserved = true
	return nil
}

func (s *OrderService) Confirm() error {
	fmt.Printf("OrderService: confirming order %s\n", s.orderID)
	// 模拟确认操作
	return nil
}

func (s *OrderService) Cancel() error {
	fmt.Printf("OrderService: cancelling order %s\n", s.orderID)
	// 模拟取消操作
	s.reserved = false
	return nil
}

type InventoryService struct {
	productID string
	quantity  int
	reserved  int
}

func NewInventoryService(productID string, quantity int) *InventoryService {
	return &InventoryService{
		productID: productID,
		quantity:  quantity,
	}
}

func (s *InventoryService) Try() error {
	reserveQuantity := 1
	if s.quantity < reserveQuantity {
		return errors.New("insufficient inventory")
	}
	fmt.Printf("InventoryService: reserving %d units of product %s\n", reserveQuantity, s.productID)
	s.reserved = reserveQuantity
	s.quantity -= reserveQuantity
	return nil
}

func (s *InventoryService) Confirm() error {
	fmt.Printf("InventoryService: confirming reservation for product %s\n", s.productID)
	// 模拟确认操作
	return nil
}

func (s *InventoryService) Cancel() error {
	fmt.Printf("InventoryService: cancelling reservation for product %s\n", s.productID)
	// 模拟取消操作
	s.quantity += s.reserved
	s.reserved = 0
	return nil
}

type PaymentService struct {
	userID   string
	amount   float64
	reserved bool
}

func NewPaymentService(userID string, amount float64) *PaymentService {
	return &PaymentService{
		userID: userID,
		amount: amount,
	}
}

func (s *PaymentService) Try() error {
	fmt.Printf("PaymentService: reserving $%.2f for user %s\n", s.amount, s.userID)
	// 模拟预留资金
	s.reserved = true
	return nil
}

func (s *PaymentService) Confirm() error {
	fmt.Printf("PaymentService: confirming payment of $%.2f for user %s\n", s.amount, s.userID)
	// 模拟确认支付
	return nil
}

func (s *PaymentService) Cancel() error {
	fmt.Printf("PaymentService: cancelling payment for user %s\n", s.userID)
	// 模拟取消支付
	s.reserved = false
	return nil
}

func main() {
	// 创建TCC服务
	orderService := NewOrderService("order-123")
	inventoryService := NewInventoryService("product-456", 10)
	paymentService := NewPaymentService("user-789", 100.0)

	// 创建TCC管理器
	manager := NewTCCManager([]TCCService{orderService, inventoryService, paymentService})

	// 执行TCC事务
	if err := manager.Execute(); err != nil {
		fmt.Printf("TCC transaction failed: %v\n", err)
	} else {
		fmt.Println("TCC transaction succeeded")
	}
}

5. Saga模式的Go语言实现

基本架构

package main

import (
	"errors"
	"fmt"
)

// SagaStep Saga步骤
type SagaStep struct {
	Execute   func() error
	Compensate func() error
}

// Saga Saga事务
type Saga struct {
	steps []SagaStep
}

// NewSaga 创建新的Saga
func NewSaga() *Saga {
	return &Saga{
		steps: make([]SagaStep, 0),
	}
}

// AddStep 添加步骤
func (s *Saga) AddStep(execute, compensate func() error) {
	s.steps = append(s.steps, SagaStep{
		Execute:   execute,
		Compensate: compensate,
	})
}

// Execute 执行Saga
func (s *Saga) Execute() error {
	// 执行步骤
	for i, step := range s.steps {
		if err := step.Execute(); err != nil {
			// 执行失败，补偿已执行的步骤
			s.compensate(i)
			return err
		}
	}
	return nil
}

// compensate 补偿已执行的步骤
func (s *Saga) compensate(upToIndex int) {
	// 从后往前补偿
	for i := upToIndex; i >= 0; i-- {
		if err := s.steps[i].Compensate(); err != nil {
			// 记录补偿失败
			fmt.Printf("Compensation failed for step %d: %v\n", i, err)
		}
	}
}

func main() {
	// 创建Saga
	saga := NewSaga()

	// 添加步骤1：创建订单
	saga.AddStep(
		func() error {
			fmt.Println("Step 1: Creating order")
			// 模拟创建订单
			return nil
		},
		func() error {
			fmt.Println("Compensating Step 1: Cancelling order")
			// 模拟取消订单
			return nil
		},
	)

	// 添加步骤2：扣减库存
	saga.AddStep(
		func() error {
			fmt.Println("Step 2: Deducting inventory")
			// 模拟扣减库存
			return nil
		},
		func() error {
			fmt.Println("Compensating Step 2: Restoring inventory")
			// 模拟恢复库存
			return nil
		},
	)

	// 添加步骤3：处理支付
	saga.AddStep(
		func() error {
			fmt.Println("Step 3: Processing payment")
			// 模拟支付失败
			return errors.New("payment failed")
		},
		func() error {
			fmt.Println("Compensating Step 3: Refunding payment")
			// 模拟退款
			return nil
		},
	)

	// 执行Saga
	if err := saga.Execute(); err != nil {
		fmt.Printf("Saga failed: %v\n", err)
	} else {
		fmt.Println("Saga succeeded")
	}
}

6. 本地消息表模式的Go语言实现

基本架构

package main

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"time"

	_ "github.com/go-sql-driver/mysql"
)

// Message 消息结构
type Message struct {
	ID        int       `json:"id"`
	Type      string    `json:"type"`
	Data      string    `json:"data"`
	Status    string    `json:"status"`
	CreatedAt time.Time `json:"created_at"`
	UpdatedAt time.Time `json:"updated_at"`
}

// Order 订单结构
type Order struct {
	ID     int     `json:"id"`
	UserID int     `json:"user_id"`
	Amount float64 `json:"amount"`
	Status string  `json:"status"`
}

// createTables 创建表结构
func createTables(db *sql.DB) error {
	// 创建订单表
	_, err := db.Exec(`
	CREATE TABLE IF NOT EXISTS orders (
		id INT AUTO_INCREMENT PRIMARY KEY,
		user_id INT NOT NULL,
		amount DECIMAL(10,2) NOT NULL,
		status VARCHAR(20) NOT NULL,
		created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
		updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
	)
	`)
	if err != nil {
		return err
	}

	// 创建消息表
	_, err = db.Exec(`
	CREATE TABLE IF NOT EXISTS messages (
		id INT AUTO_INCREMENT PRIMARY KEY,
		type VARCHAR(50) NOT NULL,
		data TEXT NOT NULL,
		status VARCHAR(20) NOT NULL DEFAULT 'pending',
		created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
		updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
	)
	`)
	return err
}

// createOrder 创建订单并发送消息
func createOrder(db *sql.DB, userID int, amount float64) error {
	// 开始事务
	tx, err := db.Begin()
	if err != nil {
		return err
	}

	// 创建订单
	result, err := tx.Exec(
		"INSERT INTO orders (user_id, amount, status) VALUES (?, ?, ?)",
		userID, amount, "created",
	)
	if err != nil {
		tx.Rollback()
		return err
	}

	// 获取订单ID
	orderID, err := result.LastInsertId()
	if err != nil {
		tx.Rollback()
		return err
	}

	// 创建订单消息
	order := Order{
		ID:     int(orderID),
		UserID: userID,
		Amount: amount,
		Status: "created",
	}

	orderJSON, err := json.Marshal(order)
	if err != nil {
		tx.Rollback()
		return err
	}

	// 插入消息
	_, err = tx.Exec(
		"INSERT INTO messages (type, data, status) VALUES (?, ?, ?)",
		"order_created", string(orderJSON), "pending",
	)
	if err != nil {
		tx.Rollback()
		return err
	}

	// 提交事务
	return tx.Commit()
}

// processMessages 处理待处理的消息
func processMessages(db *sql.DB) error {
	// 查询待处理的消息
	rows, err := db.Query(
		"SELECT id, type, data, status FROM messages WHERE status = 'pending' LIMIT 10",
	)
	if err != nil {
		return err
	}
	defer rows.Close()

	for rows.Next() {
		var msg Message
		err := rows.Scan(&msg.ID, &msg.Type, &msg.Data, &msg.Status)
		if err != nil {
			return err
		}

		// 处理消息
		err = processMessage(&msg)
		if err != nil {
			// 标记消息为失败
			_, updateErr := db.Exec(
				"UPDATE messages SET status = 'failed' WHERE id = ?",
				msg.ID,
			)
			if updateErr != nil {
				log.Printf("Failed to update message status: %v", updateErr)
			}
			continue
		}

		// 标记消息为成功
		_, err = db.Exec(
			"UPDATE messages SET status = 'success' WHERE id = ?",
			msg.ID,
		)
		if err != nil {
			log.Printf("Failed to update message status: %v", err)
		}
	}

	return rows.Err()
}

// processMessage 处理单个消息
func processMessage(msg *Message) error {
	fmt.Printf("Processing message %d: %s\n", msg.ID, msg.Type)

	// 根据消息类型处理
	switch msg.Type {
	case "order_created":
		// 处理订单创建消息
		var order Order
		if err := json.Unmarshal([]byte(msg.Data), &order); err != nil {
			return err
		}

		// 模拟处理订单，如通知库存服务、支付服务等
		fmt.Printf("Processing order %d for user %d, amount $%.2f\n", order.ID, order.UserID, order.Amount)

		// 模拟网络延迟
		time.Sleep(1 * time.Second)

		return nil

	default:
		return fmt.Errorf("unknown message type: %s", msg.Type)
	}
}

func main() {
	// 连接数据库
	db, err := sql.Open("mysql", "root:password@tcp(localhost:3306)/test")
	if err != nil {
		log.Fatalf("Failed to connect to database: %v", err)
	}
	defer db.Close()

	// 创建表结构
	if err := createTables(db); err != nil {
		log.Fatalf("Failed to create tables: %v", err)
	}

	// 创建订单
	if err := createOrder(db, 1, 100.0); err != nil {
		log.Fatalf("Failed to create order: %v", err)
	}

	// 处理消息
	if err := processMessages(db); err != nil {
		log.Fatalf("Failed to process messages: %v", err)
	}

	fmt.Println("Order created and message processed successfully")
}

7. 基于消息队列的最终一致性实现

使用RabbitMQ实现

package main

import (
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/streadway/amqp"
)

// Order 订单结构
type Order struct {
	ID     int     `json:"id"`
	UserID int     `json:"user_id"`
	Amount float64 `json:"amount"`
	Status string  `json:"status"`
}

// connectRabbitMQ 连接到RabbitMQ
func connectRabbitMQ() (*amqp.Connection, *amqp.Channel, error) {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		return nil, nil, err
	}

	ch, err := conn.Channel()
	if err != nil {
		conn.Close()
		return nil, nil, err
	}

	// 声明交换机
	err = ch.ExchangeDeclare(
		"orders",  // 交换机名称
		"fanout", // 交换机类型
		true,      // 持久化
		false,     // 自动删除
		false,     // 内部
		false,     // 不等待
		nil,       // 额外参数
	)
	if err != nil {
		ch.Close()
		conn.Close()
		return nil, nil, err
	}

	return conn, ch, nil
}

// sendOrderMessage 发送订单消息
func sendOrderMessage(ch *amqp.Channel, order Order) error {
	// 序列化订单
	orderJSON, err := json.Marshal(order)
	if err != nil {
		return err
	}

	// 发送消息
	err = ch.Publish(
		"orders", // 交换机
		"",       // 路由键
		false,    // 强制
		false,    // 立即
		amqp.Publishing{
			ContentType: "application/json",
			Body:        orderJSON,
		},
	)

	return err
}

// consumeOrderMessages 消费订单消息
func consumeOrderMessages(ch *amqp.Channel) error {
	// 声明队列
	q, err := ch.QueueDeclare(
		"",    // 队列名称（自动生成）
		false, // 持久化
		true,  // 自动删除
		true,  // 排他性
		false, // 不等待
		nil,   // 额外参数
	)
	if err != nil {
		return err
	}

	// 绑定队列到交换机
	err = ch.QueueBind(
		q.Name,  // 队列名称
		"",      // 路由键
		"orders", // 交换机名称
		false,   // 不等待
		nil,     // 额外参数
	)
	if err != nil {
		return err
	}

	// 消费消息
	msgs, err := ch.Consume(
		q.Name, // 队列名称
		"",     // 消费者标签
		true,   // 自动确认
		false,  // 排他性
		false,  // 不本地
		false,  // 不等待
		nil,    // 额外参数
	)
	if err != nil {
		return err
	}

	// 处理消息
	go func() {
		for d := range msgs {
			var order Order
			if err := json.Unmarshal(d.Body, &order); err != nil {
				log.Printf("Failed to unmarshal order: %v", err)
				continue
			}

			fmt.Printf("Received order: %+v\n", order)

			// 处理订单，如更新库存、处理支付等
			processOrder(order)
		}
	}()

	return nil
}

// processOrder 处理订单
func processOrder(order Order) {
	fmt.Printf("Processing order %d...\n", order.ID)

	// 模拟处理时间
	time.Sleep(2 * time.Second)

	// 模拟处理结果
	fmt.Printf("Order %d processed successfully\n", order.ID)
}

func main() {
	// 连接到RabbitMQ
	conn, ch, err := connectRabbitMQ()
	if err != nil {
		log.Fatalf("Failed to connect to RabbitMQ: %v", err)
	}
	defer conn.Close()
	defer ch.Close()

	// 启动消费者
	if err := consumeOrderMessages(ch); err != nil {
		log.Fatalf("Failed to start consumer: %v", err)
	}

	// 创建订单
	order := Order{
		ID:     1,
		UserID: 123,
		Amount: 99.99,
		Status: "created",
	}

	// 发送订单消息
	if err := sendOrderMessage(ch, order); err != nil {
		log.Fatalf("Failed to send order message: %v", err)
	}

	fmt.Println("Order message sent successfully")

	// 等待消费者处理
	time.Sleep(5 * time.Second)
}

8. 实际应用案例

电商订单处理

场景：用户下单后，要创建订单、扣库存、处理支付，这些操作分布在不同的服务里。典型的分布式事务场景。

实现：用Saga模式来编排整个订单流程

package main

import (
	"errors"
	"fmt"
)

// OrderService 订单服务
type OrderService struct {
	orders map[int]string
}

func NewOrderService() *OrderService {
	return &OrderService{
		orders: make(map[int]string),
	}
}

func (s *OrderService) CreateOrder(orderID int) error {
	fmt.Printf("Creating order %d\n", orderID)
	s.orders[orderID] = "created"
	return nil
}

func (s *OrderService) CancelOrder(orderID int) error {
	fmt.Printf("Cancelling order %d\n", orderID)
	s.orders[orderID] = "cancelled"
	return nil
}

// InventoryService 库存服务
type InventoryService struct {
	inventory map[string]int
}

func NewInventoryService() *InventoryService {
	return &InventoryService{
		inventory: map[string]int{
			"product1": 10,
			"product2": 5,
		},
	}
}

func (s *InventoryService) DeductInventory(productID string, quantity int) error {
	fmt.Printf("Deducting %d units of %s\n", quantity, productID)
	if s.inventory[productID] < quantity {
		return errors.New("insufficient inventory")
	}
	s.inventory[productID] -= quantity
	return nil
}

func (s *InventoryService) RestoreInventory(productID string, quantity int) error {
	fmt.Printf("Restoring %d units of %s\n", quantity, productID)
	s.inventory[productID] += quantity
	return nil
}

// PaymentService 支付服务
type PaymentService struct {
	userBalances map[int]float64
}

func NewPaymentService() *PaymentService {
	return &PaymentService{
		userBalances: map[int]float64{
			1: 1000.0,
			2: 500.0,
		},
	}
}

func (s *PaymentService) ProcessPayment(userID int, amount float64) error {
	fmt.Printf("Processing payment of $%.2f for user %d\n", amount, userID)
	if s.userBalances[userID] < amount {
		return errors.New("insufficient balance")
	}
	s.userBalances[userID] -= amount
	return nil
}

func (s *PaymentService) RefundPayment(userID int, amount float64) error {
	fmt.Printf("Refunding $%.2f to user %d\n", amount, userID)
	s.userBalances[userID] += amount
	return nil
}

// ShippingService 物流服务
type ShippingService struct {
	shipments map[int]string
}

func NewShippingService() *ShippingService {
	return &ShippingService{
		shipments: make(map[int]string),
	}
}

func (s *ShippingService) CreateShipment(orderID int) error {
	fmt.Printf("Creating shipment for order %d\n", orderID)
	s.shipments[orderID] = "created"
	return nil
}

func (s *ShippingService) CancelShipment(orderID int) error {
	fmt.Printf("Cancelling shipment for order %d\n", orderID)
	s.shipments[orderID] = "cancelled"
	return nil
}

func main() {
	// 初始化服务
	orderService := NewOrderService()
	inventoryService := NewInventoryService()
	paymentService := NewPaymentService()
	shippingService := NewShippingService()

	// 订单信息
	orderID := 1
	userID := 1
	productID := "product1"
	quantity := 2
	amount := 199.98

	// 创建Saga
	saga := NewSaga()

	// 步骤1：创建订单
	saga.AddStep(
		func() error {
			return orderService.CreateOrder(orderID)
		},
		func() error {
			return orderService.CancelOrder(orderID)
		},
	)

	// 步骤2：扣减库存
	saga.AddStep(
		func() error {
			return inventoryService.DeductInventory(productID, quantity)
		},
		func() error {
			return inventoryService.RestoreInventory(productID, quantity)
		},
	)

	// 步骤3：处理支付
	saga.AddStep(
		func() error {
			return paymentService.ProcessPayment(userID, amount)
		},
		func() error {
			return paymentService.RefundPayment(userID, amount)
		},
	)

	// 步骤4：创建物流
	saga.AddStep(
		func() error {
			return shippingService.CreateShipment(orderID)
		},
		func() error {
			return shippingService.CancelShipment(orderID)
		},
	)

	// 执行Saga
	if err := saga.Execute(); err != nil {
		fmt.Printf("Order processing failed: %v\n", err)
	} else {
		fmt.Println("Order processed successfully")
	}

	// 打印最终状态
	fmt.Printf("Order status: %s\n", orderService.orders[orderID])
	fmt.Printf("Inventory of %s: %d\n", productID, inventoryService.inventory[productID])
	fmt.Printf("User %d balance: $%.2f\n", userID, paymentService.userBalances[userID])
	fmt.Printf("Shipment status: %s\n", shippingService.shipments[orderID])
}

9. 性能优化和最佳实践

性能优化

1. 减少网络开销：

   • 能批量就别一个一个发请求
   • 异步通信走起来，别卡着等
   • 序列化/反序列化要轻量

2. 提高并发性能：

   • goroutine + channel 是Go的利器
   • 能用无锁设计就别加锁

3. 优化数据库操作：

   • 连接池是基本操作
   • 批量提交事务
   • SQL查询要写高效

4. 缓存策略：

   • 能缓存就别查库
   • 过期时间别设得太随意
   • 防止缓存穿透和雪崩

最佳实践

1. 选对方案：

   • 强一致性场景选2PC/3PC；业务逻辑复杂的选TCC；长事务选Saga；最终一致性选本地消息表或MQ
   • 要平衡一致性、性能和实现成本

2. 错误处理与重试：

   • 幂等性必须做
   • 重试用指数退避
   • 异常要有监控告警

3. 可观测性：

   • 追踪事务状态
   • 统计执行耗时
   • 日志要打全

4. 测试与演练：

   • 异常场景要覆盖全
   • 故障恢复流程要演练
   • 压力测试不能少

10. 代码优化建议

1. 错误处理优化

原始代码：

func (c *Coordinator) prepare() error {
	for _, p := range c.participants {
		if err := p.Prepare(); err != nil {
			return err
		}
	}
	return nil
}

优化建议：

func (c *Coordinator) prepare() error {
	for i, p := range c.participants {
		if err := p.Prepare(); err != nil {
			return fmt.Errorf("participant %d prepare failed: %w", i, err)
		}
	}
	return nil
}

2. 并发处理优化

原始代码：

func (c *Coordinator) rollback() error {
	var wg sync.WaitGroup
	var errMutex sync.Mutex
	var rollbackErr error

	for _, p := range c.participants {
		wg.Add(1)
		go func(participant Participant) {
			defer wg.Done()
			if err := participant.Rollback(); err != nil {
				errMutex.Lock()
				if rollbackErr == nil {
					rollbackErr = err
				}
				errMutex.Unlock()
			}
		}(p)
	}

	wg.Wait()
	return rollbackErr
}

优化建议：

func (c *Coordinator) rollback() error {
	var wg sync.WaitGroup
	errorsChan := make(chan error, len(c.participants))

	for _, p := range c.participants {
		wg.Add(1)
		go func(participant Participant) {
			defer wg.Done()
			if err := participant.Rollback(); err != nil {
				errorsChan <- err
			}
		}(p)
	}

	go func() {
		wg.Wait()
		close(errorsChan)
	}()

	var rollbackErr error
	for err := range errorsChan {
		if rollbackErr == nil {
			rollbackErr = err
		}
		log.Printf("Rollback error: %v", err)
	}

	return rollbackErr
}

3. 代码结构优化

原始代码：

// TCCService TCC服务接口
type TCCService interface {
	Try() error
	Confirm() error
	Cancel() error
}

优化建议：

// TCCService TCC服务接口
type TCCService interface {
	// Try 尝试执行业务操作，预留资源
	Try() error
	// Confirm 确认执行业务操作，使用预留的资源
	Confirm() error
	// Cancel 取消执行业务操作，释放预留的资源
	Cancel() error
}

// BaseTCCService 基础TCC服务实现
type BaseTCCService struct {
	ID string
}

// NewBaseTCCService 创建基础TCC服务
func NewBaseTCCService(id string) *BaseTCCService {
	return &BaseTCCService{ID: id}
}

// Try 默认实现
func (s *BaseTCCService) Try() error {
	return nil
}

// Confirm 默认实现
func (s *BaseTCCService) Confirm() error {
	return nil
}

// Cancel 默认实现
func (s *BaseTCCService) Cancel() error {
	return nil
}

11. 总结

分布式事务在分布式系统里从来都是一个复杂问题，不存在一个“万能方案”。选对路子，得看业务场景、一致性要求、性能需求，以及团队的维护成本。

通过本文，你应该对以下几点有了比较清晰的认识：

1. 分布式事务的基本概念和它面临的挑战
2. 几种主流方案（2PC、TCC、Saga、本地消息表等）各自的原理和适用场景
3. 用Go语言实现这些方案的具体代码路径
4. 电商订单处理的实战案例
5. 性能优化和最佳实践的思路

最后，给个选型参考：

• 强一致性优先考虑2PC或3PC
• 业务逻辑复杂的话，TCC更灵活
• 长事务用Saga，补偿机制清晰
• 最终一致性场景，本地消息表或消息队列就够用了

合理运用这些方案，才能让分布式系统的数据一致性有保障，同时兼顾系统的可靠性和性能。