3个银行营销活动方案手写实现坑，面试原理一问就露馅

面试被问“手写实现一个银行营销活动方案”，你脑子里是不是只有 if-else 堆砌？别慌，这题考的不是业务逻辑，而是高并发下的数据一致性和状态机管理。我见过太多人，方案写得花里胡哨，代码一跑就超发优惠券。今天拆解 3 个最常见的坑，全是真实生产环境血泪教训。

坑一：优惠券超发，库存扣减不原子

现象用户点击“领取优惠券”，后端返回成功，但数据库里库存变成负数。第二天对账，发现发出去的券比库存多，财务直接炸锅。这是银行营销活动最典型的事故，尤其是春节、双 11 这种高并发场景。

根本原因 很多人习惯用“查库存 -> 判断是否大于 0 -> 扣减库存”这三步走。在单线程下没问题，但高并发下，线程 A 查到库存 1，线程 B 也查到库存 1，两者同时执行扣减，结果库存变成 -1。这就是经典的竞态条件（Race Condition）。

正确写法对比

❌ 错误写法（非原子操作）

def claim_coupon_wrong(user_id, coupon_id):# 1. 查询当前库存stock = db.query("SELECT stock FROM coupons WHERE id=?", coupon_id)# 2. 判断库存if stock > 0:# 3. 扣减库存（这里存在时间窗口）db.execute("UPDATE coupons SET stock = stock - 1 WHERE id=?", coupon_id)# 4. 发放优惠券给用户db.execute("INSERT INTO user_coupons (user_id, coupon_id) VALUES (?, ?)", user_id, coupon_id)return Trueelse:return False

✅ 正确写法（原子操作 + 乐观锁）

def claim_coupon_right(user_id, coupon_id):# 1. 原子扣减：利用 SQL 的原子性，直接更新# 只有当 stock > 0 时，才会执行扣减，返回受影响的行数affected_rows = db.execute("UPDATE coupons SET stock = stock - 1 WHERE id = ? AND stock > 0", coupon_id)# 2. 判断扣减是否成功if affected_rows == 1:# 3. 发放优惠券给用户db.execute("INSERT INTO user_coupons (user_id, coupon_id) VALUES (?, ?)", user_id, coupon_id)return Trueelse:return False

关键点：UPDATE ... WHERE stock > 0 是数据库层面的原子操作，无论多少个线程同时执行，数据库引擎会保证每次扣减都是独立的，不会出现负数。

复现与修复代码 你可以写一个简单的压力测试，用 100 个线程同时调用 claim_coupon_wrong，你会发现库存很容易变成负数。换成 claim_coupon_right，库存只会扣减到 0 为止。

规避建议

永远不要用应用层代码做“检查后修改”（Check-Then-Act），要用数据库的原子操作。
如果业务复杂，可以考虑引入 Redis 做预扣减，再异步同步到数据库，但要注意 Redis 和 DB 的数据一致性。

坑二：用户重复领取，缺乏幂等性控制

现象用户网络抖动，点击“领取”按钮后页面卡住，用户以为没成功，又点了一次。结果领到了两张优惠券。或者，用户快速双击按钮，导致重复发放。

根本原因 前端没做防抖，后端没做幂等性校验。银行营销活动对重复领取极其敏感，因为每张券都有成本。

正确写法对比

❌ 错误写法（无幂等性）

def claim_coupon_no_idempotent(user_id, coupon_id):# 直接检查库存并扣减，不关心用户是否已经领取过stock = db.query("SELECT stock FROM coupons WHERE id=?", coupon_id)if stock > 0:db.execute("UPDATE coupons SET stock = stock - 1 WHERE id=?", coupon_id)db.execute("INSERT INTO user_coupons (user_id, coupon_id) VALUES (?, ?)", user_id, coupon_id)return Truereturn False

✅ 正确写法（基于唯一索引的幂等性）

# 数据库表结构：user_coupons (user_id, coupon_id, created_at)
# 关键：在 (user_id, coupon_id) 上建立唯一索引def claim_coupon_idempotent(user_id, coupon_id):try:# 1. 尝试插入记录，利用唯一索引保证幂等# 如果用户已经领取过，插入会失败db.execute("INSERT INTO user_coupons (user_id, coupon_id) VALUES (?, ?)", user_id, coupon_id)# 2. 原子扣减库存affected_rows = db.execute("UPDATE coupons SET stock = stock - 1 WHERE id = ? AND stock > 0", coupon_id)if affected_rows == 1:return Trueelse:# 库存不足，回滚用户领取记录db.execute("DELETE FROM user_coupons WHERE user_id = ? AND coupon_id = ?", user_id, coupon_id)return Falseexcept IntegrityError:# 3. 唯一索引冲突，说明用户已经领取过return "ALREADY_CLAIMED"

关键点：利用数据库的唯一索引作为幂等性的基石。只要 user_id 和 coupon_id 的组合是唯一的，重复请求就会被拦截。

复现与修复代码 模拟用户快速连续点击，你会发现错误写法会插入多条记录，而正确写法只会插入一条，后续请求返回“已领取”。

规避建议

后端必须做幂等性校验，不能依赖前端的防抖。
使用唯一索引是最简单可靠的方式，比在代码里加锁更高效。
如果业务需要更复杂的幂等性，可以考虑引入请求 ID（Request ID）作为幂等键。

坑三：活动状态切换，并发下状态不一致

现象活动配置了“10 点开始，11 点结束”。但在 10 点 59 分 59 秒时，有些用户还能领取，10 点 00 分 01 秒时，有些用户却不能领取。甚至出现活动已经结束了，但还有用户在领取的情况。

根本原因 活动状态（未开始、进行中、已结束）在内存中维护，但没有与数据库状态同步。高并发下，不同线程读取的状态可能不一致。

正确写法对比

❌ 错误写法（内存状态）

class MarketingActivity:def __init__(self, activity_id, start_time, end_time):self.activity_id = activity_idself.start_time = start_timeself.end_time = end_timeself.status = "NOT_STARTED"  # 内存状态def check_status(self):current_time = time.time()if current_time < self.start_time:self.status = "NOT_STARTED"elif current_time < self.end_time:self.status = "IN_PROGRESS"else:self.status = "ENDED"return self.statusdef claim_coupon(self, user_id, coupon_id):if self.check_status() != "IN_PROGRESS":return False# ... 扣减库存逻辑 ...return True

✅ 正确写法（数据库状态 + 实时校验）

def claim_coupon_with_status_check(user_id, coupon_id, activity_id):# 1. 从数据库查询活动状态，而不是内存activity = db.query("SELECT start_time, end_time, status FROM activities WHERE id = ?", activity_id)if not activity:return False# 2. 实时校验时间，不依赖状态字段current_time = datetime.now()if current_time < activity['start_time'] or current_time >= activity['end_time']:return False# 3. 原子扣减库存affected_rows = db.execute("UPDATE coupons SET stock = stock - 1 WHERE id = ? AND stock > 0", coupon_id)if affected_rows == 1:db.execute("INSERT INTO user_coupons (user_id, coupon_id) VALUES (?, ?)", user_id, coupon_id)return Trueelse:return False

关键点：