当前位置：首页 > news >正文

Go语言分布式事务与一致性保障

news 2026/5/24 1:15:07

Go语言分布式事务与一致性保障

引言

在分布式系统中，事务管理和数据一致性是核心挑战。本文将深入探讨Go语言中分布式事务的实现方案，包括两阶段提交、最终一致性、Saga模式等。

一、分布式事务概述

1.1 分布式事务的特性

特性	说明
ACID	原子性、一致性、隔离性、持久性
CAP定理	一致性、可用性、分区容错性三选二
BASE理论	基本可用、软状态、最终一致性

1.2 一致性级别

级别	说明
强一致性	数据更新后立即同步到所有副本
弱一致性	数据更新后不保证立即同步
最终一致性	数据更新后最终会同步到所有副本

二、两阶段提交（2PC）

2.1 2PC流程

协调者 参与者1 参与者2 | | | |--- Prepare ---| | | | |--- Prepare --| | | | |--- Prepare ---| | | | | | | | |<-- Ready ----| | | |<-- Ready --------| | | | | | |<-- Ready ----| | | | | |--- Commit ---| | | | |--- Commit --| | | | |--- Commit ---| | | | | | | | |<-- Done ----| | | |<-- Done --------| | | | | | |<-- Done ----|

2.2 实现示例

type TransactionCoordinator struct { participants []Participant logger *zap.Logger } type Participant interface { Prepare() error Commit() error Rollback() error } func (tc *TransactionCoordinator) Execute(ctx context.Context) error { tc.logger.Info("Starting 2PC transaction") // Phase 1: Prepare tc.logger.Info("Phase 1: Prepare") for _, participant := range tc.participants { if err := participant.Prepare(); err != nil { tc.logger.Error("Prepare failed, rolling back", zap.Error(err)) tc.rollbackAll() return err } } // Phase 2: Commit tc.logger.Info("Phase 2: Commit") for _, participant := range tc.participants { if err := participant.Commit(); err != nil { tc.logger.Error("Commit failed", zap.Error(err)) // 部分提交失败，需要人工介入 return err } } tc.logger.Info("Transaction completed successfully") return nil } func (tc *TransactionCoordinator) rollbackAll() { for _, participant := range tc.participants { if err := participant.Rollback(); err != nil { tc.logger.Error("Rollback failed", zap.Error(err)) } } }

2.3 数据库参与者实现

type DBParticipant struct { db *sql.DB } func (p *DBParticipant) Prepare() error { _, err := p.db.Exec("BEGIN TRANSACTION") return err } func (p *DBParticipant) Commit() error { _, err := p.db.Exec("COMMIT") return err } func (p *DBParticipant) Rollback() error { _, err := p.db.Exec("ROLLBACK") return err }

三、三阶段提交（3PC）

3.1 3PC流程

协调者 参与者1 参与者2 | | | |--- CanCommit ---| | | | |--- CanCommit --| | | | |--- CanCommit ---| | | | | | |<-- Yes --------| | |<-- Yes ----------| | | | | |<-- Yes ----| | | | | |--- PreCommit ---| | | | |--- PreCommit --| | | | |--- PreCommit ---| | | | | | |<-- Ready --------| | |<-- Ready ----------| | | | | |<-- Ready ----| | | | | |--- DoCommit ---| | | | |--- DoCommit --| | | | |--- DoCommit ---|

3.2 3PC优化点

阶段	作用	优化说明
CanCommit	询问参与者是否可以提交	轻量级检查，不锁定资源
PreCommit	准备提交	锁定资源，写入redo log
DoCommit	执行提交	参与者超时自动提交

四、最终一致性与事件驱动

4.1 事件溯源模式

type EventStore interface { Append(event *DomainEvent) error GetStream(aggregateID string) ([]*DomainEvent, error) } type DomainEvent struct { EventID string AggregateID string EventType string Data []byte Version int Timestamp int64 } type UserAggregate struct { ID string Name string Email string Version int } func (u *UserAggregate) Apply(event *DomainEvent) { switch event.EventType { case "UserCreated": var data UserCreatedData json.Unmarshal(event.Data, &data) u.ID = data.UserID u.Name = data.Name u.Email = data.Email case "UserUpdated": var data UserUpdatedData json.Unmarshal(event.Data, &data) if data.Name != "" { u.Name = data.Name } if data.Email != "" { u.Email = data.Email } } u.Version = event.Version } func (u *UserAggregate) Create(name, email string) (*DomainEvent, error) { if u.ID != "" { return nil, errors.New("user already exists") } event := &DomainEvent{ EventID: uuid.New().String(), AggregateID: uuid.New().String(), EventType: "UserCreated", Version: 1, Timestamp: time.Now().Unix(), } data, _ := json.Marshal(UserCreatedData{ UserID: event.AggregateID, Name: name, Email: email, }) event.Data = data u.Apply(event) return event, nil }

4.2 CQRS模式

type CommandHandler interface { Handle(cmd Command) error } type QueryHandler interface { Handle(query Query) (interface{}, error) } type Command interface { GetCommandID() string } type Query interface { GetQueryID() string } type CreateUserCommand struct { CommandID string Name string Email string } func (c *CreateUserCommand) GetCommandID() string { return c.CommandID } type GetUserQuery struct { QueryID string UserID string } func (q *GetUserQuery) GetQueryID() string { return q.QueryID } type UserCommandHandler struct { eventStore EventStore eventBus EventBus } func (h *UserCommandHandler) Handle(cmd Command) error { switch c := cmd.(type) { case *CreateUserCommand: return h.handleCreateUser(c) } return nil } func (h *UserCommandHandler) handleCreateUser(cmd *CreateUserCommand) error { aggregate := &UserAggregate{} event, err := aggregate.Create(cmd.Name, cmd.Email) if err != nil { return err } if err := h.eventStore.Append(event); err != nil { return err } return h.eventBus.Publish(event) }

五、Saga模式

5.1 Saga协调器

type Saga struct { ID string Steps []SagaStep CurrentStep int Status SagaStatus CreatedAt time.Time UpdatedAt time.Time } type SagaStatus string const ( SagaStatusPending SagaStatus = "pending" SagaStatusRunning SagaStatus = "running" SagaStatusCompleted SagaStatus = "completed" SagaStatusFailed SagaStatus = "failed" SagaStatusCompensating SagaStatus = "compensating" ) type SagaStep struct { ID string Action func() error CompensatingAction func() error Status StepStatus } type StepStatus string const ( StepStatusPending StepStatus = "pending" StepStatusSuccess StepStatus = "success" StepStatusFailed StepStatus = "failed" ) func (s *Saga) Execute(ctx context.Context) error { s.Status = SagaStatusRunning for i := s.CurrentStep; i < len(s.Steps); i++ { step := &s.Steps[i] step.Status = StepStatusPending if err := step.Action(); err != nil { step.Status = StepStatusFailed s.Status = SagaStatusFailed return s.compensate(i) } step.Status = StepStatusSuccess s.CurrentStep = i + 1 } s.Status = SagaStatusCompleted return nil } func (s *Saga) compensate(failedStepIndex int) error { s.Status = SagaStatusCompensating for i := failedStepIndex - 1; i >= 0; i-- { step := &s.Steps[i] if step.Status == StepStatusSuccess { if err := step.CompensatingAction(); err != nil { // 补偿失败，记录日志并报警 return err } } } return nil }

5.2 Saga示例：订单创建流程

func CreateOrderSaga(orderID, userID string, items []OrderItem) *Saga { steps := []SagaStep{ { ID: "1", Action: func() error { return reserveInventory(items) }, CompensatingAction: func() error { return releaseInventory(items) }, }, { ID: "2", Action: func() error { return deductBalance(userID, calculateTotal(items)) }, CompensatingAction: func() error { return refundBalance(userID, calculateTotal(items)) }, }, { ID: "3", Action: func() error { return createOrder(orderID, userID, items) }, CompensatingAction: func() error { return cancelOrder(orderID) }, }, { ID: "4", Action: func() error { return sendNotification(userID, orderID) }, CompensatingAction: func() error { return nil // 通知无需补偿 }, }, } return &Saga{ ID: uuid.New().String(), Steps: steps, Status: SagaStatusPending, CreatedAt: time.Now(), } }

六、分布式锁

6.1 Redis分布式锁

type RedisLock struct { client *redis.Client key string value string ttl time.Duration } func NewRedisLock(client *redis.Client, key string, ttl time.Duration) *RedisLock { return &RedisLock{ client: client, key: key, value: uuid.New().String(), ttl: ttl, } } func (l *RedisLock) Acquire(ctx context.Context) (bool, error) { result, err := l.client.SetNX(ctx, l.key, l.value, l.ttl).Result() if err != nil { return false, err } return result, nil } func (l *RedisLock) Release(ctx context.Context) error { // 使用Lua脚本保证原子性 script := ` if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("DEL", KEYS[1]) else return 0 end ` _, err := l.client.Eval(ctx, script, []string{l.key}, l.value).Result() return err } func (l *RedisLock) Refresh(ctx context.Context) error { script := ` if redis.call("GET", KEYS[1]) == ARGV[1] then return redis.call("EXPIRE", KEYS[1], ARGV[2]) else return 0 end ` _, err := l.client.Eval(ctx, script, []string{l.key}, l.value, int(l.ttl.Seconds())).Result() return err }

6.2 ZooKeeper分布式锁

type ZKLock struct { conn *zk.Conn path string lockNode string sessionID int64 } func NewZKLock(conn *zk.Conn, path string) *ZKLock { return &ZKLock{ conn: conn, path: path, } } func (l *ZKLock) Acquire(ctx context.Context) error { // 创建临时有序节点 nodePath := l.path + "/lock-" lockNode, err := l.conn.Create( nodePath, []byte{}, zk.FlagEphemeral|zk.FlagSequence, zk.WorldACL(zk.PermAll), ) if err != nil { return err } l.lockNode = lockNode // 获取所有子节点并排序 children, _, err := l.conn.Children(l.path) if err != nil { return err } sort.Strings(children) // 检查是否是最小节点 for i, child := range children { if child == filepath.Base(lockNode) { if i == 0 { return nil // 获取锁成功 } // 监听前一个节点 prevNode := l.path + "/" + children[i-1] _, _, ch, err := l.conn.GetW(ctx, prevNode) if err != nil { return err } select { case <-ch: return l.Acquire(ctx) // 前一个节点删除，重试获取锁 case <-ctx.Done(): return ctx.Err() } } } return errors.New("lock not acquired") } func (l *ZKLock) Release(ctx context.Context) error { return l.conn.Delete(l.lockNode, -1) }

七、幂等性保障

7.1 唯一请求ID

func GenerateRequestID() string { return uuid.New().String() } type IdempotentService struct { cache *redis.Client ttl time.Duration } func (s *IdempotentService) CheckAndSet(requestID string) (bool, error) { result, err := s.cache.SetNX(context.Background(), requestID, "processing", s.ttl).Result() if err != nil { return false, err } return result, nil } func (s *IdempotentService) MarkCompleted(requestID string) error { return s.cache.Set(context.Background(), requestID, "completed", s.ttl).Err() } func (s *IdempotentService) GetStatus(requestID string) (string, error) { result, err := s.cache.Get(context.Background(), requestID).Result() if err == redis.Nil { return "", nil } if err != nil { return "", err } return result, nil }

7.2 业务唯一键

func (s *OrderService) CreateOrder(req *CreateOrderRequest) (*Order, error) { // 检查业务唯一键 businessKey := fmt.Sprintf("order:%s:%s", req.UserID, req.OrderNo) exists, err := s.idempotentService.CheckAndSet(businessKey) if err != nil { return nil, err } if !exists { // 重复请求，返回之前的结果 return s.getCachedOrder(req.OrderNo) } defer func() { if err == nil { s.idempotentService.MarkCompleted(businessKey) } }() // 执行业务逻辑 order, err := s.executeCreateOrder(req) if err != nil { return nil, err } // 缓存结果 s.cacheOrder(order) return order, nil }

八、分布式事务最佳实践

8.1 避免分布式事务

// 反例：跨服务事务 func TransferMoney(from, to string, amount float64) error { tx1, _ := fromDB.Begin() tx2, _ := toDB.Begin() err := deductBalance(tx1, from, amount) if err != nil { tx1.Rollback() return err } err = addBalance(tx2, to, amount) if err != nil { tx1.Rollback() tx2.Rollback() return err } tx1.Commit() tx2.Commit() return nil } // 正例：使用消息队列实现最终一致性 func TransferMoneyAsync(from, to string, amount float64) error { // 本地事务：扣除余额并记录转账记录 err := fromDB.Transaction(func(tx *gorm.DB) error { if err := deductBalance(tx, from, amount); err != nil { return err } if err := createTransferRecord(tx, from, to, amount); err != nil { return err } return nil }) if err != nil { return err } // 发送消息通知接收方 return eventBus.Publish(&TransferEvent{ From: from, To: to, Amount: amount, }) }