版本号机制在业务中的应用

CREATE TABLE `account` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `user_id` bigint NOT NULL COMMENT '用户id',
  `balance` decimal(18,2) NOT NULL COMMENT '当前余额，精确到分',
  `version` int NOT NULL DEFAULT '0' COMMENT '版本号',
  `status` tinyint NOT NULL DEFAULT '1' COMMENT '账户状态（1：正常、2：冻结）',
  `created_at` datetime NOT NULL,
  `updated_at` datetime NOT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `udx_uid` (`user_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

CREATE TABLE `account_flow` (
  `id` bigint NOT NULL AUTO_INCREMENT,
  `flow_no` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '流水号',
  `account_id` bigint NOT NULL COMMENT '关联的账户id',
  `amount` decimal(18,2) NOT NULL COMMENT '变动金额（正：进账，负：出账）',
  `balance_before` decimal(18,2) NOT NULL COMMENT '变动前余额',
  `balance_after` decimal(18,2) NOT NULL COMMENT '变动后余额',
  `type` tinyint NOT NULL COMMENT '流水类型（1：充值、2：消费、3：退款、4：提现）',
  `biz_no` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NOT NULL COMMENT '业务单号',
  `version_seq` int NOT NULL COMMENT '关联账户的版本号（用于追溯）',
  `created_at` datetime NOT NULL ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `udx_fno` (`flow_no`),
  KEY `idx_aid_cat` (`account_id`,`created_at`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

// 接口路由定义
func main() {
    account := biz.NewAccountBiz()

    // NOTE: 仅做演示，具体业务中不会直接将 uid 放到路由中暴露，一般会从登录信息中提取。
    r := gin.Default()
    r.GET("/accounts/:uid", account.GetUserAccount)
    r.POST("/accounts/:uid/actions/init", account.InitUserAccount)
    r.POST("/accounts/:uid/actions/update", account.UpdateUserAccountAndCreateFlow)

    r.Run(":3000")
}

// 更新账户并创建流水操作
func (s *AccountSvc) UpdateUserAccountAndCreateFlow(userId int64, amount float64, bizNo string, flowType model.AccountFlowType) (*model.Account, *model.AccountFlow, error) {
    // NOTE: 在主从架构下，需在主库执行事务，避免主从延迟导致的版本号不一致，导致大量失败的事务回滚
    tx, err := s.ds.Beginx()
    if err != nil {
        return nil, nil, err
    }
    defer tx.Commit()

    // NOTE: 在高频写入场景下，需要加 FOR UPDATE，否则会导致大量失败的事务回滚
    // NOTE: 在读多写少且低频写入的场景下，可以不加 FOR UPDATE，避免频繁的锁竞争（X 型行锁）
    account := new(model.Account)
    err = tx.Get(account, "SELECT * FROM account WHERE user_id = ? FOR UPDATE", userId) // NOTE: 可只查询必要字段，使用联合索引来避免回表操作
    if err != nil {
        tx.Rollback()
        return nil, nil, err
    }

    result, err := tx.NamedExec(`UPDATE account SET balance = balance + :amount, version = version + 1 WHERE id = :id AND version = :version`, map[string]any{"amount": amount, "id": account.ID, "version": account.Version})
    if err != nil {
        tx.Rollback()
        return nil, nil, err
    }

    // NOTE: 抛出错误，由客户端主动重试，此处也可以使用自动重试，但需要考虑重试次数和重试间隔
    if rowsAffected, _ := result.RowsAffected(); rowsAffected == 0 {
        tx.Rollback()
        return nil, nil, errors.New("version conflict, please retry")
    }

    accountFlow := &model.AccountFlow{
        FlowNo:        uuid.NewString(),
        AccountID:     account.ID,
        Amount:        amount,
        BalanceBefore: account.Balance,
        BalanceAfter:  account.Balance + amount,
        Type:          int32(flowType),
        BizNo:         bizNo,
        VersionSeq:    account.Version + 1,
        CreatedAt:     time.Now().Local(),
    }

    result, err = tx.NamedExec(`INSERT INTO account_flow (flow_no, account_id, amount, balance_before, balance_after, type, biz_no, version_seq, created_at) VALUES (:flow_no, :account_id, :amount, :balance_before, :balance_after, :type, :biz_no, :version_seq, :created_at)`, accountFlow)
    if err != nil {
        tx.Rollback()
        return nil, nil, err
    }

    accountFlow.ID, _ = result.LastInsertId()
    return account, accountFlow, nil
}