隔离级别、锁与最终一致性实战
在掌握 InnoDB 锁机制后，需要结合隔离级别与业务代码理解锁的实际行为。单机场景下用悲观锁或乐观锁即可；跨服务（订单、库存分库）时，本地事务无法覆盖全局一致性，需要配合 Kafka + 本地重试 + 死信队列 兜底。
本文分两部分：单机锁实战（RC/RR、间隙锁、悲观/乐观锁、死锁）与 分布式最终一致性（下单扣库存链路）。
RC 与 RR 的生产差异
常见配置：MySQL 默认隔离级别为 RR（可重复读），不少互联网公司改为 RC（读已提交）+ binlog 格式 ROW。
| 隔离级别 | 间隙锁 | 并发 | 一致性 |
|---|---|---|---|
| RR | 有，锁住行及行间间隙 | 较低，易死锁 | 同一事务内可重复读，防幻读 |
| RC | 无，只锁实际命中行 | 较高 | 可能出现不可重复读、幻读 |
RR 间隙锁示例：WHERE id BETWEEN 1 AND 10 即使范围内无记录，也会锁住该区间，阻止其他事务插入 id = 5。
选型参考：
- 强一致性场景（金融对账）：倾向 RR，或 RC 下辅以分布式锁 / 乐观锁。
- 高并发场景（抢购、秒杀）：倾向 RC，在应用层补偿一致性。
间隙锁复现
以下用两个并发事务演示 RR 下普通索引 UPDATE 引发的间隙锁阻塞。
终端 A（先执行，暂不提交）：
UPDATE orders SET price = 100 WHERE status = 1;
-- 暂停 10 秒后再 COMMIT
终端 B（A 未提交时执行）：
INSERT INTO orders (status, price) VALUES (1, 200);
预期结果：B 被阻塞，直至 A 提交或触发 Lock wait timeout exceeded。说明 RR 下对 status = 1 的 UPDATE 在普通索引上产生了间隙锁，阻塞同间隙内的 INSERT。
悲观锁与乐观锁
悲观锁（SELECT ... FOR UPDATE）
在事务内锁定行，其他事务对同一行的写操作需排队等待。
TypeORM 示例：
await this.repo.findOne({
where: { id: 1 },
lock: { mode: 'pessimistic_write' },
});
适用：冲突频繁、必须保证互斥（库存扣减）。事务内逻辑应尽量短，避免 RPC 或长耗时计算占用连接。
乐观锁（Version 字段）
通过版本号检测冲突，更新时校验 version 是否变化。
TypeORM 示例：Entity 添加 @VersionColumn()，更新 SQL 形如：
UPDATE ... SET version = version + 1 WHERE id = 1 AND version = :oldVersion
影响行数为 0 表示版本冲突，应用层捕获后重试。
适用：冲突少、读多写少（如修改用户昵称）。并发 TPS 高，但冲突时会产生重试。
选型对比
| 方式 | 冲突频繁（抢库存） | 冲突极少（改资料） |
|---|---|---|
| 悲观锁 | 推荐 | 连接占用高，不推荐 |
| 乐观锁 | 大量重试，不推荐 | 推荐 |
死锁本地复现
开启两个并发请求调用同一扣库存接口，在第一个事务提交前让第二个事务持有冲突锁，可触发：
Deadlock found when trying to get lock
排查时执行 SHOW ENGINE INNODB STATUS，查看 LATEST DETECTED DEADLOCK 中的 WE ROLL BACK TRANSACTION、*** (1) TRANSACTION、*** (2) TRANSACTION，定位 owner 与 waiter 的锁等待关系。
从单机锁到分布式一致性
典型问题
订单服务与库存服务分属不同数据库时，@Transactional 无法跨服务回滚，会出现：
- 订单已创建，库存扣减失败 → 超卖风险需人工对账。
- 库存已扣，订单未落库 → 库存被占用但无对应订单。
核心原则：先保证本服务数据落地，再通过消息异步驱动下游；失败靠重试 + 死信队列（DLQ）+ 补偿任务，而不是指望分布式两阶段提交。
架构概览
与 Kafka 系列文档的对应关系：
| 环节 | 风险 | 详见 |
|---|---|---|
| 生产者发消息 | 消息丢失 | 消息丢失解决方案 |
| 消费者处理 | 重复消费 | 消息重复消费 |
| 分区再分配 | Rebalance 丢消息/重复 | Rebalance 机制 |
生产者：事务提交后再发消息
订单创建与 Kafka 发送不能放在同一本地事务里（MQ 不参与 DB 事务）。正确顺序：
- 本地事务写入订单，状态
PENDING。 - 事务提交成功后再发 Kafka 消息。
- 发送失败时记录日志，由补偿任务扫描
PENDING订单重发（见文末 Outbox 方案）。
// order.service.ts
@Injectable()
export class OrderService {
constructor(
@InjectRepository(Order) private orderRepo: Repository<Order>,
private dataSource: DataSource,
private kafkaProducer: KafkaProducerService,
) {}
async createOrder(dto: CreateOrderDto) {
const order = await this.dataSource.transaction(async (manager) => {
const entity = manager.create(Order, { ...dto, stockStatus: 'PENDING' });
return manager.save(entity);
});
// 事务已提交，再发消息
try {
await this.kafkaProducer.send('inventory.deduct', {
orderId: order.id,
productId: dto.productId,
quantity: dto.quantity,
timestamp: Date.now(),
});
} catch (err) {
// Kafka 不可用：订单已落库，依赖补偿任务重发
this.logger.error('Kafka send failed, order pending retry', { orderId: order.id, err });
}
return order;
}
}
生产环境更稳妥的做法是 Transactional Outbox：在同一事务内写业务表 + outbox 表，由独立进程轮询 outbox 发 Kafka，避免「DB 提交成功、消息未发出」的窗口期无迹可寻。
消费者：扣库存 + 重试 + DLQ
消费者侧要点：
- 先处理业务，再提交 offset（
enable.auto.commit=false），避免消息丢失。 - 乐观锁冲突等可恢复错误做有限次本地重试。
- 不可恢复错误（库存不足）转发 DLQ 后提交 offset，避免阻塞主队列。
- 消费逻辑必须幂等（按
orderId去重），应对重复投递。
// inventory.consumer.ts
@Injectable()
export class InventoryConsumer {
constructor(
private inventoryService: InventoryService,
private kafkaProducer: KafkaProducerService,
) {}
@KafkaConsumer('inventory.deduct')
async handleDeduct(message: DeductPayload, context: KafkaContext) {
const { orderId, productId, quantity } = message;
try {
await this.deductWithRetry(orderId, productId, quantity);
await this.kafkaProducer.send('order.update.status', { orderId, status: 'PAID' });
await context.commitOffset();
} catch (error) {
if (this.isRetryable(error)) {
// 不提交 offset，由 Kafka 重新投递
throw error;
}
await this.kafkaProducer.send('inventory.deduct.dlq', {
originalMessage: message,
error: error.message,
failedAt: new Date().toISOString(),
});
await context.commitOffset();
}
}
private async deductWithRetry(orderId: string, productId: string, quantity: number) {
const maxAttempts = 3;
for (let attempt = 1; attempt <= maxAttempts; attempt++) {
try {
const ok = await this.inventoryService.deductStock(productId, quantity, orderId);
if (!ok) throw new ConflictException('version conflict');
return;
} catch (e) {
if (!(e instanceof ConflictException) || attempt === maxAttempts) throw e;
await sleep(attempt * 200);
}
}
}
private isRetryable(error: unknown): boolean {
return error instanceof ConflictException;
}
}
deductStock 内部按前文选型使用悲观锁或乐观锁；orderId 作为幂等键，重复消费时直接返回成功。
库存扣减与锁的衔接
| 场景 | 锁策略 | 说明 |
|---|---|---|
| 秒杀、热点 SKU | 悲观锁 FOR UPDATE | 冲突高，避免乐观锁大量重试 |
| 普通商品、冲突低 | 乐观锁 @VersionColumn | 吞吐更高，冲突时消费者本地重试 |
| 扣减记录幂等 | 唯一索引 (order_id) | 防止重复消费双扣 |
-- 悲观锁扣减（事务内）
SELECT stock FROM inventory WHERE product_id = ? FOR UPDATE;
UPDATE inventory SET stock = stock - ? WHERE product_id = ? AND stock >= ?;
-- 乐观锁扣减
UPDATE inventory
SET stock = stock - ?, version = version + 1
WHERE product_id = ? AND version = ? AND stock >= ?;
故障演练清单
| 步骤 | 操作 | 预期 |
|---|---|---|
| 1. 正常链路 | 下单 → 扣库存 → 订单变 PAID | Consumer Lag 趋近 0 |
| 2. 乐观锁冲突 | 人为让首次扣减抛 ConflictException | 控制台可见重试日志，最终成功 |
| 3. 库存不足 | 将库存设为 0 后下单 | 消息进入 inventory.deduct.dlq，主队列不堵塞 |
| 4. Kafka 宕机 | 下单时 Broker 不可用 | 订单保持 PENDING，补偿任务恢复后重发 |
Outbox 补偿（生产级兜底）
仅靠 try-catch 会在服务重启或 Kafka 长时间不可用时留下大量 PENDING 订单。标准做法是：
- Outbox 表：与订单同事务写入
{ order_id, topic, payload, status, retry_count }。 - 定时任务（如每分钟）：扫描
status = PENDING且创建超过 5 分钟的记录，重发 Kafka 并递增retry_count。 - 上限与告警：
retry_count超过阈值转人工或 DLQ，避免无限重试。
@Cron('*/1 * * * *')
async scanPendingOutbox() {
const rows = await this.outboxRepo.find({
where: { status: 'PENDING', retryCount: LessThan(10) },
});
for (const row of rows) {
try {
await this.kafkaProducer.send(row.topic, row.payload);
await this.outboxRepo.update(row.id, { status: 'SENT' });
} catch {
await this.outboxRepo.increment({ id: row.id }, 'retryCount', 1);
}
}
}
幂等键（orderId）保证补偿重发与正常消费结果一致，与 消息重复消费 中的去重设计一脉相承。
面试速答
Q：订单服务重启或 Kafka 宕机，如何保证 PENDING 订单最终被处理？
A：Transactional Outbox + 定时补偿扫描；消费端按 orderId 幂等；生产者 acks=all + 幂等生产者；消费者手动提交 offset 且先处理后提交。彻底失败进 DLQ 人工介入。