面试翻车实录：青桔单车后端手写实现避坑指南

上周陪一个刚入职青桔单车的兄弟复盘面试，他卡在了一道看似简单的并发控制题上。面试官问：“如果同时有一百万个用户抢同一辆车的锁，你的代码怎么保证不超卖？请现场手写实现。”他脑子一热，直接写了个 if-else 判断库存，结果被追问到哑口无言。那一刻，你我都可能经历过这种尴尬：面试被问原理答不上来，明明业务逻辑都会，但一让手写实现底层逻辑，脑子就空白。

青桔单车作为头部共享出行平台，其后端系统对高并发、数据一致性的要求极高。很多候选人只盯着业务代码，忽略了底层的资源竞争处理。今天我们就以青桔单车的“车辆状态机”和“订单创建”为例，拆解那些让你在现场手写实现时容易翻车的坑。

坑的现象：锁了个寂寞，数据还是脏的

最典型的坑就是“伪原子操作”。在共享电动车场景中，一辆车从“空闲”变为“已借出”是一个状态迁移。很多开发者习惯用“先查后改”的模式：先查询车辆状态是否为空闲，如果是，则更新为已借出。

# 错误写法：典型的非原子操作，高并发下必出Bug
def unlock_bike_wrong(bike_id, user_id):# 1. 查询车辆状态bike = db.query("SELECT status FROM bikes WHERE id = ?", bike_id)if bike['status'] == 'available':# 2. 更新车辆状态db.execute("UPDATE bikes SET status = 'borrowed', user_id = ? WHERE id = ?", user_id, bike_id)return Trueelse:return False

这段代码在单线程测试时毫无问题，但在青桔单车这种高并发场景下，当两个请求同时到达，都会查询到 status = 'available'，然后都执行更新操作。结果是：一辆车被两个用户同时“借走”，或者数据库主键冲突报错。这就是典型的竞态条件（Race Condition）。面试官问的不是你能不能跑通，而是你能不能保证数据一致性。

根本原因：缺乏对ACID与分布式一致性的敬畏

为什么会出现这种问题？根本原因在于对数据库事务隔离级别和分布式系统一致性的理解不够深入。在单机数据库中，我们可以依赖事务的原子性；但在分布式微服务架构中，车辆服务、订单服务、用户服务往往分布在不同机器上。

根据 RFC 1945（HTTP/1.0 规范）中关于无状态协议的定义，每个请求都是独立的，服务端不保留客户端状态。但在业务层，我们必须手动维护状态的最终一致性。青桔单车的车辆状态机是一个典型的状态机模型，状态迁移必须满足互斥性和原子性。

错误的写法违背了原子性原则。SELECT 和 UPDATE 是两个独立的操作，中间存在时间窗口。在高并发下，这个时间窗口就是灾难。正确的做法是利用数据库的行级锁机制，或者使用 Redis 的原子操作来抢占资源。

正确写法对比：乐观锁与悲观锁的选择

针对车辆状态的变更，业界主流有两种方案：乐观锁和悲观锁。在青桔单车这类读多写少、但写冲突激烈的场景下，乐观锁（基于版本号或 CAS 机制）通常是首选。

方案一：数据库层乐观锁（推荐）

在数据库表中增加一个 version 字段。每次更新时，带上版本号条件。如果版本号不匹配，说明数据被其他事务修改，更新失败，重试或报错。

-- 建表时增加 version 字段
ALTER TABLE bikes ADD COLUMN version INT DEFAULT 0;

# 正确写法：利用 version 字段实现乐观锁
def unlock_bike_optimistic(bike_id, user_id):max_retries = 3for i in range(max_retries):# 1. 查询当前版本号和状态bike = db.query("SELECT status, version FROM bikes WHERE id = ?", bike_id)if not bike:raise Exception("Bike not found")if bike['status'] != 'available':return Falsecurrent_version = bike['version']# 2. 原子更新：只有版本号匹配且状态为空闲时才更新affected_rows = db.execute("UPDATE bikes SET status = 'borrowed', user_id = ?, version = version + 1 WHERE id = ? AND version = ? AND status = 'available'",user_id, bike_id, current_version)# 3. 判断是否更新成功if affected_rows == 1:return Trueelse:# 更新失败，说明被抢了，重试或返回失败continuereturn False

逐行讲解：

SELECT 获取当前 version。
UPDATE 语句中增加了 AND version = ? 和 AND status = 'available' 条件。
如果并发发生，第一个请求更新成功，version 变为 1。第二个请求拿着 version=0 去更新，条件不满足，affected_rows 为 0。
通过检查 affected_rows 判断是否成功，实现了原子性。

方案二：Redis 原子抢占（高性能场景）

如果流量极大，数据库压力扛不住，可以先在 Redis 层做一层拦截。利用 Redis 的 SETNX 或 Lua 脚本保证原子性。

-- Redis Lua 脚本：保证原子性
local bike_status = redis.call('GET', 'bike:status:' .. KEYS[1])
if bike_status == 'available' thenredis.call('SET', 'bike:status:' .. KEYS[1], 'borrowed')redis.call('SET', 'bike:user:' .. KEYS[1], ARGV[1])return 1
elsereturn 0
end

# 正确写法：Redis Lua 脚本抢占
def unlock_bike_redis(bike_id, user_id):# 执行 Lua 脚本，保证原子性result = redis.eval(lua_script, 1, bike_id, user_id)if result == 1:# Redis 抢占成功，异步同步到数据库async_sync_to_db(bike_id, user_id)return Trueelse:return False

对比分析：

数据库乐观锁：数据一致性最强，直接落库，适合对数据准确性要求极高的核心链路。
Redis 抢占：性能极高，QPS 可达十万级，但需要处理 Redis 与 DB 的最终一致性问题（如 Redis 成功但 DB 失败的情况，需通过消息队列补偿）。

在青桔单车的架构中，通常采用 Redis 缓存车辆状态 + DB 乐观锁兜底 的组合拳。Redis 用于快速过滤无效请求，DB 用于保证最终数据一致性。

复现与修复代码：从报错到稳定

为了让大家更直观地看到问题，我们模拟一个高并发场景。假设有一辆车，100 个线程同时尝试借车。

复现 Bug

使用之前的“错误写法”：

import threadingdef simulate_wrong_code():bike_id = 1001db.update("UPDATE bikes SET status = 'available' WHERE id = ?", bike_id)results = []def worker():res = unlock_bike_wrong(bike_id, 'user_' + str(threading.get_ident()))results.append(res)threads = []for _ in range(100):t = threading.Thread(target=worker)threads.append(t)t.start()for t in threads:t.join()success_count = sum(1 for r in results if r)print(f"成功借车人数: {success_count}") # 预期 1，实际可能 > 1

运行结果往往是：成功借车人数 1-5 人不等。这就是超卖。

修复后的代码

使用“乐观锁”写法：

def simulate_correct_code():bike_id = 1001db.update("UPDATE bikes SET status = 'available', version = 0 WHERE id = ?", bike_id)results = []def worker():res = unlock_bike_optimistic(bike_id, 'user_' + str(threading.get_ident()))results.append(res)threads = []for _ in range(100):t = threading.Thread(target=worker)threads.append(t)t.start()for t in threads:t.join()success_count = sum(1 for r in results if r)print(f"成功借车人数: {success_count}") # 预期且稳定为 1

运行结果稳定为 1。这就是原子性带来的安全感。

规避建议：面试与实战的通用准则

不要相信 if-else 的原子性：在并发场景下，任何“先查后改”的逻辑都是危险的。必须使用 UPDATE ... WHERE version = ? 或 Redis 原子操作。
理解 RFC 规范中的幂等性要求：虽然 RFC 主要讲 HTTP，但引申到微服务接口设计，幂等性是必须的。借车接口必须保证多次调用效果一致，防止网络重试导致重复扣费或重复借车。
面试答题技巧：
- 先说结论：我会使用数据库乐观锁（版本号）来保证原子性。
- 再讲原理：解释 CAS（Compare And Swap）思想，避免死锁，提高并发吞吐量。
- 最后讲优化：如果流量更大，我会引入 Redis 做前置过滤，并通过消息队列保证 DB 最终一致性。
- 时间分配：手写代码控制在 10-15 分钟，留 5 分钟思考边界条件（如车坏了、用户黑名单等）。