谈谈并发控制
背景
在多用户协作的管理系统中，经常会遇到这样的场景：
场景举例：Bush 分类管理
Bush & Jungle 是为鲜切花相关商品服务的管理系统，在分类管理中：
- 用户 A 打开分类编辑弹窗，准备修改"玫瑰"分类的名称（改为"鲜切玫瑰"）
- 同时，用户 B 也打开同一个分类，准备修改它的排序编号
- 两人几乎同时点击保存
- 如果没有并发控制，后提交的用户会覆盖先提交用户的修改
这就是典型的并发场景，需要保证数据一致性。
问题分析
什么是并发场景？
只要满足以下任一条件，就属于并发场景：
- 多用户同时操作同一数据（如上述例子）
- 单用户多窗口操作（用户打开两个浏览器标签页，同时编辑）
- 用户 + 系统定时任务同时操作（如库存扣减时，用户下单 + 系统自动补货）
时间轴展示
T1: 用户 A 读取数据 { id: 1, name: "玫瑰", order: 10 }
T2: 用户 B 读取数据 { id: 1, name: "玫瑰", order: 10 }
T3: 用户 A 修改 name → "鲜切玫瑰"，提交 { id: 1, name: "鲜切玫瑰", order: 10 }
T4: 用户 B 修改 order → 20，提交 { id: 1, name: "玫瑰", order: 20 }
T5: 数据库最终状态 { id: 1, name: "玫瑰", order: 20 } ❌用户 A 的修改丢失了！ 这就是经典的 Lost Update（更新丢失） 问题。
常见的并发控制策略
1. 乐观锁（Optimistic Locking）✅ Bush/Jungle 采用
原理： 不加锁，提交时检查数据是否被修改
实现方式：
- 版本号（version）
- 时间戳（updatedAt）
- 数据快照（比较整行数据）
优点：
- ✅ 无锁等待，性能高
- ✅ 适合读多写少场景
- ✅ 实现简单
- ✅ 无死锁风险
缺点：
- ❌ 并发写入时会有冲突，需要重试
- ❌ 用户体验稍差（需要处理冲突提示）
适用场景：
typescript
✅ 管理后台（用户操作频率低）
✅ 博客编辑（单人编辑为主）
✅ 配置管理（修改不频繁）
✅ 我们的分类管理（典型场景）2. 悲观锁（Pessimistic Locking）
原理： 读取时就加锁，直到提交才释放
实现方式：
typescript
// 后端实现示例
async update(id: number, dto: UpdateDto): Promise<Category> {
return await this.connection.transaction(async (manager) => {
// SELECT ... FOR UPDATE 加行锁
const category = await manager
.createQueryBuilder(Category, 'category')
.setLock('pessimistic_write')
.where('id = :id', { id })
.getOne();
// ...
});
}优点：
- ✅ 完全避免冲突
- ✅ 适合写多读少场景
- ✅ 数据一致性强
缺点：
- ❌ 锁等待，性能差
- ❌ 可能死锁
- ❌ 并发度低
适用场景：
✅ 库存扣减（绝对不能超卖）
✅ 财务对账（金额必须精确）
✅ 抢购秒杀（先到先得）3. 分布式锁（Distributed Lock）
原理： 多实例部署时，通过 Redis/Zookeeper 实现全局锁
实现方式：
typescript
// Redis 分布式锁
import { RedisService } from '@nestjs-modules/ioredis';
async updateWithLock(id: number, dto: UpdateDto): Promise<Category> {
const lockKey = `category:lock:${id}`;
const lockValue = uuidv4(); // 唯一标识，防止误删别人的锁
// 1. 尝试获取锁（超时 10 秒）
// SET key value EX 10 NX
// - NX: 仅当 key 不存在时才设置（Not eXists）→ 获取锁
// - EX 10: 设置过期时间 10 秒（EXpire）→ 防止死锁
// - 返回 'OK' 表示加锁成功，null 表示锁已被占用
const acquired = await this.redis.set(
lockKey, // 锁的键名
lockValue, // 锁的值（用于验证是否是自己的锁）
'EX', 10, // 10 秒后自动过期
'NX' // 仅当不存在时才设置
);
if (!acquired) {
throw new ConflictException('操作频繁，请稍后重试');
}
try {
// 2. 执行业务逻辑
const category = await this.findOne(id);
Object.assign(category, dto);
return await this.categoryRepository.save(category);
} finally {
// 3. 释放锁（使用 Lua 脚本保证原子性）
// 为什么用 Lua？
// - 必须检查是否是自己的锁（lockValue 匹配）
// - 检查和删除必须是原子操作，防止误删别人的锁
await this.redis.eval(
`if redis.call("get",KEYS[1]) == ARGV[1] then
return redis.call("del",KEYS[1])
else
return 0
end`,
1, // 参数个数
lockKey, // KEYS[1]
lockValue // ARGV[1]
);
}
}Lua 是什么？如何保证原子性？
typescript
// Lua 是一种轻量级脚本语言
// Redis 支持在服务端执行 Lua 脚本，整个脚本作为一个原子操作
// ❌ 错误做法：分两步操作（非原子）
const value = await redis.get(lockKey); // 步骤1：读取
if (value === lockValue) {
await redis.del(lockKey); // 步骤2：删除
}
// 问题：步骤1和2之间，锁可能已经过期并被别人获取
// 结果：误删了别人的锁！
// ✅ 正确做法：Lua 脚本（原子操作）
await redis.eval(`
if redis.call("get", KEYS[1]) == ARGV[1] then
return redis.call("del", KEYS[1])
else
return 0
end
`, 1, lockKey, lockValue);
// Redis 保证：整个 Lua 脚本执行期间不会插入其他命令
// 相当于：检查 + 删除 = 一个不可分割的操作
// 类比：数据库事务
BEGIN;
SELECT value FROM locks WHERE key = 'lockKey';
IF value = 'myValue' THEN
DELETE FROM locks WHERE key = 'lockKey';
END IF;
COMMIT;
// Lua 脚本在 Redis 中起到了类似作用为什么获取锁用 SET 而不是专门的 LOCK 命令？
Redis 没有专门的 LOCK 命令，SET key value NX EX 是业界标准做法：
NX（Not eXists）：只有 key 不存在时才设置成功 → 实现互斥EX（EXpire）：设置过期时间 → 防止死锁（进程崩溃后自动释放）- 这是一个原子操作，避免了竞态条件
优点：
- ✅ 支持多实例部署
- ✅ 避免并发冲突
- ✅ 灵活（可设置超时、自动续期）
缺点：
- ❌ 实现复杂
- ❌ 依赖 Redis/Zookeeper
- ❌ 性能开销
适用场景：
✅ 微服务架构（多实例部署）
✅ 多实例定时任务（防止重复执行）
✅ 多实例抢购秒杀（高并发场景）为什么这些场景需要分布式锁？
定时任务防重复：
场景：每小时统计订单数据 问题：3 台服务器，每台都有定时任务，会统计 3 次！ 解决：加分布式锁，只有抢到锁的实例才执行 悲观锁不行：数据库锁只在单个事务内有效，定时任务可能跨多台服务器 乐观锁不行：无法阻止其他实例启动，只能事后检测冲突秒杀场景：
场景：1000 件商品，10000 人抢购 问题：多实例部署，每个实例都在扣库存，如何保证不超卖？ 解决：分布式锁 + Redis 队列 - 分布式锁：保证多实例只有一个在扣库存 - Redis 队列：先把请求存起来慢慢处理，避免瞬间压垮服务器 悲观锁不行：数据库锁性能差，无法应对高并发 乐观锁不行：大量请求会冲突重试，浪费资源
分布式锁和 Redis 队列的关系（重要！）
首先澄清：分布式锁和队列都是用 Redis 实现的，但它们解决不同的问题：
typescript
// 1. Redis 队列：解决"流量控制"问题
// 问题：10000 个请求同时到达，服务器扛不住
// 解决：先把请求放进队列，慢慢处理
// 用 Redis List 实现队列
await redis.lpush('秒杀队列', JSON.stringify({ userId, productId }));
// Worker 慢慢取出处理（比如每秒处理 100 个）
const request = await redis.brpop('秒杀队列', 0);
// 2. 分布式锁：解决"多实例并发"问题
// 问题：3 台服务器同时从队列取请求，同时扣库存，可能超卖
// 解决：加分布式锁，确保同一时刻只有一个实例在扣库存
// 用 Redis SET NX 实现分布式锁
const lockKey = `product:lock:${productId}`;
const acquired = await redis.set(lockKey, uuid, 'EX', 10, 'NX');
↑ 这就是用 Redis 加锁
if (acquired) {
// 只有抢到锁的 Worker 才能扣库存
await 扣减库存();
await redis.del(lockKey); // 释放锁
}
// 完整流程（都用 Redis）
┌─ 实例 A Worker ─┐
10000 请求 → 队列 ─┼─ 实例 B Worker ─┼→ 抢分布式锁 → 扣库存
(Redis) └─ 实例 C Worker ─┘ (Redis)
↑ ↑ ↑
解决流量高峰 从队列慢慢取 保证只有一个扣库存
// 总结：
// - Redis 队列：用 List 数据结构（lpush/brpop）
// - 分布式锁：用 String 数据结构（SET NX EX）
// - 都是 Redis，但用的功能不同4. MVCC（多版本并发控制）—— 了解即可
说明： MVCC 是数据库内置机制，不需要我们编码实现。这里简单了解概念，知道 PostgreSQL 默认就在用即可。
原理： 数据库自动保存数据的多个版本，读写互不阻塞
日常开发中的体现：
typescript
// 你可能已经在用 MVCC，只是不知道而已
// Jungle 使用 PostgreSQL，默认就是 MVCC
// 你不需要写任何特殊代码，数据库自动帮你做了：
场景：
- 事务 A 正在读取分类列表（耗时 2 秒）
- 事务 B 同时更新某个分类
- 结果：两个事务互不干扰，都能正常执行
// 如果没有 MVCC（老式数据库）：
// - 事务 A 读取时，事务 B 必须等待（写等读）
// - 或者事务 B 更新时，事务 A 必须等待（读等写）是否需要深入了解？
typescript
// 🟢 日常开发：不需要
// - PostgreSQL/MySQL 默认就开启了
// - 我们主要关注乐观锁（手动实现）
// - MVCC 是数据库帮我们做的优化
// 🟡 面试/进阶：可以了解
// - 理解"隔离级别"的概念
// - 知道 Read Committed、Repeatable Read 的区别
// - 了解脏读、幻读等术语
// 🔴 深入研究：数据库内核开发才需要
// - 实现细节（xmin/xmax）
// - VACUUM 机制
// - 性能调优Jungle 的隔离级别：
typescript
// PostgreSQL 默认：Read Committed
// 含义：只能读到已提交的数据（不会脏读）
// 我们的做法：Read Committed + 乐观锁总结：
- MVCC 让读写互不阻塞，我们的项目自动享受这个优化
- 日常开发关注乐观锁即可，MVCC 了解概念就够了
策略选择
"多实例"与"多数据库"
同一服务的多实例（共享数据库）
┌─────────┐ ┌─────────┐ ┌─────────┐
│ 实例 1 │ │ 实例 2 │ │ 实例 3 │ ← Jungle 后端 3 个容器
└────┬────┘ └────┬────┘ └────┬────┘
└────────┬────────┬─────────┘
↓ ↓
┌──────────────────┐
│ PostgreSQL │ ← 共享同一个数据库
└──────────────────┘
✅ 乐观锁可以工作（version 在数据库中）
✅ 悲观锁可以工作（数据库锁机制）微服务架构（各自独立数据库）
┌──────────────┐ ┌──────────────┐
│ 订单服务 │ │ 库存服务 │
│ (3个实例) │ │ (3个实例) │
└──────┬───────┘ └──────┬───────┘
↓ ↓
┌──────────────┐ ┌──────────────┐
│ 订单数据库 │ │ 库存数据库 │
└──────────────┘ └──────────────┘
❌ 单个服务内的乐观锁/悲观锁无法跨服务生效
✅ 需要分布式锁（Redis）或分布式事务业界实践：
- 小型项目（Bush/Jungle）：单体架构，一个数据库，多实例部署 → 乐观锁/悲观锁即可
- 中型项目：多个微服务，各自数据库 → 需要分布式锁/分布式事务
- 大型项目：分库分表（Sharding）→ 需要分布式事务中间件（如 Seata）
决策树
┌─ 能否接受"最终一致性"？
│
├─ 能接受（允许短暂不一致 + 重试）
│ └─ 乐观锁 ✅ (Bush/Jungle)
│ - 单实例/多实例共享数据库都可以
│ - 适合：分类管理、用户配置、系统设置等
│
└─ 不能接受（必须强一致性）
│
├─ 单数据库（单实例 or 多实例共享）
│ └─ 悲观锁 + 事务
│ - SELECT ... FOR UPDATE
│ - 适合：库存扣减、订单创建等
│
└─ 多数据库（微服务 or 分库分表）
└─ 分布式锁（Redis）或分布式事务
- SET NX EX + Lua 脚本
- 适合：跨服务操作、秒杀、定时任务等理解数据一致性
一致性的分类
1. 强一致性（Strong Consistency）
定义：任何时刻，所有节点看到的数据都是一致的
例子：银行转账
- A 账户扣款 100 元
- B 账户必须立即收到 100 元
- 不能出现：A 扣了，B 没收到的中间状态
实现：分布式锁 + 悲观锁 + 2PC（两阶段提交）2. 最终一致性（Eventual Consistency）
定义：短时间内数据可能不一致，但最终会达到一致
例子：社交媒体点赞
- 你点赞后，自己立即看到（本地状态）
- 别人可能延迟几秒才看到（异步同步）
- 最终大家看到的数据是一致的
实现：乐观锁 + 重试机制为什么乐观锁不是强一致性？
typescript
场景：分布式库存扣减
// 问题：乐观锁允许短暂的不一致
时间轴（多实例部署）：
T1: 实例 A 读取库存 = 100（version: 1）
T2: 实例 B 读取库存 = 100（version: 1）
T3: 实例 A 扣减成功，库存 = 99（version: 2）
T4: 实例 B 尝试扣减，version 不匹配，冲突！
T5: 实例 B 重试，读取最新库存 = 99（version: 2）
T6: 实例 B 扣减成功，库存 = 98（version: 3）
关键点：
- T3-T4 之间，实例 B 持有的是"过期数据"（库存 100）
- 虽然最终结果正确，但有"短暂不一致"的窗口期
- 如果 T4 时查询库存，可能看到不同的值
// 强一致性方案：分布式锁
时间轴：
T1: 实例 A 获取分布式锁
T2: 实例 B 尝试获取锁，等待...
T3: 实例 A 扣减库存，释放锁
T4: 实例 B 获取锁，读取最新库存，扣减，释放锁
关键点：
- 任何时刻只有一个实例在操作
- 不存在"过期数据"的窗口期
- 所有实例看到的数据都是最新的总结：
- 乐观锁 → 最终一致性 → 适合低频修改、允许重试的场景
- 分布式锁 + 悲观锁 → 强一致性 → 适合高频并发、必须实时准确的场景
Bush & Jungle 的解决方案
我们的项目采用 乐观锁（Optimistic Locking） 方案。
为什么选择最终一致性？
分类管理场景特点：
- ✅ 冲突概率低（两人同时编辑同一分类的概率很小）
- ✅ 允许短暂不一致（冲突后刷新即可，不影响业务）
- ✅ 用户体验好（无锁等待，响应快）
不适合最终一致性的场景：
- ❌ 库存扣减（不能超卖）
- ❌ 余额变更（金额必须精确）
- ❌ 秒杀抢购（必须严格排队）
实现方案
1. 数据库设计 & 实体定义
在 categories 表添加 version 字段。
typescript
// jungle/src/modules/category/entities/category.entity.ts
import { VersionColumn } from 'typeorm';
@Entity('categories')
export class Category {
@PrimaryGeneratedColumn()
id: number;
@Column()
name: string;
/**
* 乐观锁版本号
* TypeORM 会在每次 save() 时自动递增
* 用于防止并发更新时的数据覆盖
*/
@VersionColumn()
version: number;
}2. 后端校验逻辑
typescript
// jungle/src/modules/category/category.service.ts
async update(id: number, updateCategoryDto: UpdateCategoryDto): Promise<Category> {
const category = await this.findOne(id);
// 乐观锁：检查版本号
if (
updateCategoryDto.version !== undefined &&
updateCategoryDto.version !== category.version
) {
throw new ConflictException('分类已被其他用户修改，请刷新后重试');
}
// ...
// TypeORM 保存时会自动递增 version
}3. 前端处理
编辑弹窗提交
tsx
// bush/src/modules/category/views/CategoryListView.tsx
const handleFinish = async (values: Record<string, any>) => {
const submissionData = { ...values };
try {
if (isEditing) {
// 携带当前的 version 字段
submissionData.version = currentRow!.version;
await updateCategory(currentRow!.id, submissionData);
}
actionRef.current?.reload();
return true;
} catch (error: any) {
// 处理 409 冲突错误
if (error?.response?.status === 409) {
modal.confirm({
title: '数据冲突',
content: '分类已被其他用户修改，是否重新加载最新数据？',
onOk: async () => {
await actionRef.current?.reload();
setModalVisible(false);
},
});
return false;
}
throw error;
}
};状态开关切换
tsx
<Switch
checked={record.isActive}
onChange={async (checked) => {
try {
await updateCategory(record.id, {
isActive: checked,
version: record.version, // 携带 version 用于乐观锁
});
actionRef.current?.reload();
} catch (error: any) {
if (error?.response?.status === 409) {
modal.warning({
title: '数据冲突',
content: '分类已被其他用户修改，请刷新页面后重试',
onOk: () => {
actionRef.current?.reload();
},
});
}
}
}}
/>4. 完整流程演示
T1: 用户 A 读取 { id: 1, name: "玫瑰", version: 1 }
T2: 用户 B 读取 { id: 1, name: "玫瑰", version: 1 }
T3: 用户 A 提交 { id: 1, name: "鲜切玫瑰", version: 1 }
→ 后端校验: version(1) == DB.version(1) ✅
→ 保存成功，version 自动递增为 2
T4: 用户 B 提交 { id: 1, order: 20, version: 1 }
→ 后端校验: version(1) != DB.version(2) ❌
→ 抛出 409 冲突异常
T5: 前端提示用户 B："数据已被修改，请刷新后重试"
T6: 用户 B 重新加载，看到最新数据 { id: 1, name: "鲜切玫瑰", version: 2 }