从一次线上数据重复事故，复盘MySQL幂等插入的3种最佳实践（附Go/Python代码）
凌晨三点，告警铃声刺破了寂静——监控系统显示优惠券发放接口的重复调用率突然飙升300%。登录服务器查看日志，发现因网络抖动导致的重试机制，让部分用户收到了两张完全相同的优惠券。这不是简单的数据冗余问题，而是涉及资金成本的重大事故。经过紧急回滚和问题排查，我们不得不重新审视那个看似简单却暗藏玄机的问题：如何确保MySQL插入操作的绝对幂等性？
1. 唯一约束：数据库层的天然屏障
当我们在用户表发现两条相同手机号的记录时，才意识到数据库设计阶段忽略的唯一约束有多重要。唯一索引不仅是性能优化手段，更是数据一致性的第一道防线。
1.1 INSERT IGNORE的温柔陷阱
-- 用户基础表结构 CREATE TABLE `users` ( `id` BIGINT AUTO_INCREMENT, `mobile` VARCHAR(11) NOT NULL, `name` VARCHAR(64), PRIMARY KEY (`id`), UNIQUE KEY `uk_mobile` (`mobile`) ) ENGINE=InnoDB; -- 错误示范：普通INSERT可能引发异常 INSERT INTO users(mobile, name) VALUES('13800138000', '张三'); -- 正确用法：忽略重复 INSERT IGNORE INTO users(mobile, name) VALUES('13800138000', '李四');注意：
INSERT IGNORE会静默吞掉所有错误，包括数据类型转换等非唯一键冲突，使用时需确保业务能接受这种"宽容"处理。
1.2 ON DUPLICATE KEY UPDATE的智慧
当需要更新重复记录时，这个语法展现出惊人威力：
# Python示例：存在则更新最后登录时间 import pymysql conn = pymysql.connect(host='localhost', user='root', password='', db='test') try: with conn.cursor() as cursor: sql = """INSERT INTO users(mobile, name, last_login) VALUES(%s, %s, NOW()) ON DUPLICATE KEY UPDATE last_login = NOW()""" cursor.execute(sql, ('13800138000', '张三')) conn.commit() finally: conn.close()三种唯一约束方案的对比：
| 方案 | 冲突处理方式 | 返回值差异 | 适用场景 |
|---|---|---|---|
| 普通INSERT | 抛出异常 | 错误代码1062 | 需要严格中断流程 |
| INSERT IGNORE | 静默跳过 | AffectedRows=0 | 允许静默失败 |
| ON DUPLICATE UPDATE | 执行更新操作 | AffectedRows=1/2 | 需要更新重复记录 |
2. 应用层双检模式：精确制导的防御策略
在优惠券系统中，我们发现仅靠数据库约束无法应对所有场景。比如需要根据用户等级判断发放资格时，就需要更复杂的校验逻辑。
2.1 Go语言实现的事务型双检
// Go示例：发放限时优惠券 func GrantCoupon(userID int64, couponID string) error { tx, err := db.Begin() if err != nil { return err } defer tx.Rollback() // 第一重检查：是否已领取 var count int err = tx.QueryRow("SELECT COUNT(1) FROM user_coupons WHERE user_id=? AND coupon_id=?", userID, couponID).Scan(&count) if err != nil { return err } if count > 0 { return errors.New("coupon already granted") } // 第二重检查：库存是否充足 var stock int err = tx.QueryRow("SELECT stock FROM coupons WHERE id=?", couponID).Scan(&stock) if err != nil { return err } if stock <= 0 { return errors.New("coupon out of stock") } // 执行发放 _, err = tx.Exec("INSERT INTO user_coupons(user_id, coupon_id) VALUES(?, ?)", userID, couponID) if err != nil { return err } // 扣减库存 _, err = tx.Exec("UPDATE coupons SET stock=stock-1 WHERE id=?", couponID) return tx.Commit() }2.2 双检模式的致命缺陷
在分布式环境下，我们踩过一个深坑：两个并发的请求可能同时通过第一重检查，导致超发。解决方案是将检查逻辑放入数据库事务中，并合理设置事务隔离级别：
-- 设置事务隔离级别为REPEATABLE READ SET TRANSACTION ISOLATION LEVEL REPEATABLE READ; BEGIN; -- 检查与插入必须在同一个事务中 SELECT * FROM user_coupons WHERE user_id=123 AND coupon_id='NEWYEAR2023' FOR UPDATE; INSERT INTO user_coupons(user_id, coupon_id) VALUES(123, 'NEWYEAR2023'); COMMIT;3. 分布式锁与Token机制：集群环境下的终极防御
当系统扩展到数十个实例时，数据库约束和本地事务都力不从心。我们引入Redis实现分布式锁方案：
3.1 基于Redis的防重令牌
# Python实现防重令牌 import redis import uuid r = redis.Redis(host='localhost', port=6379) def generate_token(user_id): token = str(uuid.uuid4()) r.setex(f"req_token:{user_id}", 300, token) return token def verify_token(user_id, token): stored_token = r.get(f"req_token:{user_id}") if not stored_token or stored_token.decode() != token: return False r.delete(f"req_token:{user_id}") return True3.2 Go实现的分布式锁方案
// Go实现Redis分布式锁 func AcquireLock(lockKey string, timeout time.Duration) (string, bool) { token := uuid.New().String() result, err := redisClient.SetNX(lockKey, token, timeout).Result() if err != nil || !result { return "", false } return token, true } func ReleaseLock(lockKey, token string) { script := ` if redis.call("get", KEYS[1]) == ARGV[1] then return redis.call("del", KEYS[1]) else return 0 end` redisClient.Eval(script, []string{lockKey}, token).Result() } // 使用示例 func ProcessOrder(orderID string) { lockKey := "order_lock:" + orderID token, ok := AcquireLock(lockKey, 10*time.Second) if !ok { return errors.New("操作过于频繁") } defer ReleaseLock(lockKey, token) // 核心业务逻辑 }三种方案的性能对比测试数据：
环境：MySQL 8.0, Redis 6.2, 4核8G云服务器，100并发测试
| 方案 | QPS | 平均延迟 | 错误率 | 资源消耗 |
|---|---|---|---|---|
| 数据库唯一约束 | 1250 | 78ms | 0% | CPU 45% |
| 应用层双检 | 860 | 112ms | 1.2% | CPU 62% |
| Redis分布式锁 | 3200 | 30ms | 0% | MEM 70% |
4. 混合策略：电商订单系统的实战案例
在日订单量超百万的电商系统中，我们最终采用分层防御策略：
- 前端防御：提交按钮防重复点击，请求添加唯一ID
- 网关层：Nginx限流 + 请求去重缓存
- 应用层：
- 短时缓存标记（Redis 5秒过期）
- 数据库唯一索引（订单号+业务类型）
- 数据层：
- 最终一致性检查（定时任务补偿）
// 完整订单创建流程示例 func CreateOrder(userID int64, items []CartItem) (*Order, error) { // 1. 生成唯一请求ID requestID := uuid.New().String() // 2. Redis防重检查 if ok := redisClient.SetNX("order_req:"+requestID, 1, 5*time.Second); !ok { return nil, ErrDuplicateRequest } // 3. 获取分布式锁 lockToken, ok := AcquireLock(f"order_lock:{userID}", 3*time.Second) if !ok { return nil, ErrOperationTooFrequent } defer ReleaseLock(f"order_lock:{userID}", lockToken) // 4. 数据库事务 tx := db.Begin() defer tx.Rollback() // 5. 检查库存 for _, item := range items { var stock int if err := tx.Raw("SELECT stock FROM skus WHERE id=? FOR UPDATE", item.SkuID).Scan(&stock); err != nil { return nil, err } if stock < item.Quantity { return nil, ErrInsufficientStock } } // 6. 创建订单（包含唯一约束） orderNo := generateOrderNo() if err := tx.Exec("INSERT INTO orders(order_no, user_id) VALUES(?, ?)", orderNo, userID); err != nil { if isDuplicateEntry(err) { return nil, ErrOrderExists } return nil, err } // 7. 扣减库存 for _, item := range items { if err := tx.Exec("UPDATE skus SET stock=stock-? WHERE id=?", item.Quantity, item.SkuID); err != nil { return nil, err } } if err := tx.Commit().Error; err != nil { return nil, err } return &Order{OrderNo: orderNo}, nil }在经历这次事故后，我们建立了完整的幂等防御体系。有趣的是，当系统真正达到金融级数据一致性时，CPU使用率反而下降了15%——因为重试风暴导致的无效计算大幅减少。这或许就是良好设计带来的意外收获：最好的性能优化，有时就来自于避免不必要的操作。
