Appearance
监听器与表单提交并发一致性分析
目录
问题场景
用户描述的场景
并发操作：
- 监听器：流程状态变更，更新
status字段 - 用户提交表单：读取
status字段，根据状态执行不同逻辑
期望：
- 无论谁先谁后执行，都要等待另一个完成后，读取到最新数据
- 避免因读取到旧数据导致逻辑错误
风险：
- 监听器正在更新
status=1 - 用户提交表单读取到
status=0（旧数据） - 导致逻辑走向错误的分支
具体示例
业务场景：报销申请
java
// 监听器：流程完成时更新状态
@Override
public void upProcessResult(BpmProcessInstanceResultEvent event) {
synchronized (this) {
if (result.equals(WfNotice.Type.FINISH.getCode())) {
// 更新状态为"已完成"
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, StatusConstonts.FINISH_STATUS) // status = 2
.eq(HjlApplyExpenseEntity::getId, id));
}
}
}
// 用户提交表单：处理审批节点的额外数据
@Override
public void disposeProcessFrom(WfProcess process) {
synchronized (this) {
Long id = process.getBusinessKey();
// ⚠️ 读取当前状态
HjlApplyExpenseEntity entity = getById(id);
// ⚠️ 根据状态判断
if (entity.getStatus() == StatusConstonts.AUDIT_STATUS) { // status = 1（审批中）
// 分支A：允许填写报销信息
entity.setPaymentFrom(...);
updateById(entity);
} else if (entity.getStatus() == StatusConstonts.FINISH_STATUS) { // status = 2（已完成）
// 分支B：不允许修改，抛出异常
throw new BusinessException("流程已完成，无法修改");
}
}
}并发问题：
如果监听器和用户提交表单同时执行：
场景A：监听器先更新
status=2，用户后读取- 期望：读取到
status=2，抛出异常 - 实际：可能读到
status=1（旧数据），执行分支A ❌
- 期望：读取到
场景B：用户先读取
status=1，监听器后更新- 期望：用户读到
status=1，执行分支A，然后监听器更新status=2 - 实际：可能出现数据覆盖问题 ❌
- 期望：用户读到
风险分析
风险1：事务隔离级别导致的幻读
MySQL 默认隔离级别
MySQL InnoDB 默认隔离级别是 REPEATABLE READ（可重复读）。
特点：
- 同一事务中，多次读取同一数据，结果一致
- 通过 MVCC（多版本并发控制） 实现
- 第一次读取时建立快照，后续读取使用快照
问题：
时间轴：
T1 (监听器事务):
├─ 开始事务
├─ UPDATE status=2 WHERE id=1 (获取排他锁)
├─ 提交事务
└─ 释放锁
│
▼
T2 (用户提交表单事务):
├─ 开始事务
├─ SELECT * FROM table WHERE id=1 ← ⚠️ 读到的是什么？
│
│ 如果 T1 已提交：
│ - REPEATABLE READ: 可能读到 status=1（事务开始时的快照）❌
│ - READ COMMITTED: 读到 status=2（已提交的最新值）✅
│
├─ 业务逻辑（基于读取的 status）
└─ UPDATE ...结论：在 REPEATABLE READ 隔离级别下，即使 T1 已提交，T2 仍可能读到旧数据。
示例验证
终端1（监听器事务）：
sql
-- 开始事务
START TRANSACTION;
-- 更新状态
UPDATE hjl_apply_expense SET status = 2 WHERE id = 1;
-- 提交事务
COMMIT;终端2（用户提交表单事务）：
sql
-- 开始事务（在 T1 提交之前开始）
START TRANSACTION;
-- 第一次查询（可能在 T1 提交之前）
SELECT status FROM hjl_apply_expense WHERE id = 1;
-- 结果：status = 1（旧值）
-- （此时 T1 已提交）
-- 第二次查询（在 T1 提交之后）
SELECT status FROM hjl_apply_expense WHERE id = 1;
-- 结果：status = 1（仍然是旧值！因为 REPEATABLE READ）❌
-- 提交事务
COMMIT;原因：REPEATABLE READ 通过快照读实现，事务中的读取会使用事务开始时的数据快照。
风险2：synchronized 锁的作用范围有限
synchronized 的问题
java
synchronized (this) {
@Transactional
public void upProcessResult(...) {
update(Wrappers.lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, 2)
.eq(HjlApplyExpenseEntity::getId, id));
}
} // ← synchronized 释放
// 但事务可能还未提交！
// Spring 的事务提交在方法返回后执行时序分析：
线程1 (监听器):
├─ synchronized(this) 获取锁 🔒
├─ @Transactional 开始
├─ UPDATE status=2 WHERE id=1
├─ synchronized(this) 释放锁 🔓 ← ⚠️ 锁释放了
│
│ （此时线程2可以获取锁）
│
└─ @Transactional 提交 ← ⚠️ 但事务还未提交！
线程2 (用户提交表单):
├─ synchronized(this) 获取锁 🔒 ← ✅ 获取成功
├─ @Transactional 开始
├─ SELECT status FROM ... WHERE id=1
│ └─> 读到 status=1（线程1的事务还未提交）❌
├─ 业务逻辑（基于 status=1）
├─ UPDATE ...
└─> synchronized(this) 释放锁 🔓问题：
- synchronized 的释放时机在方法返回时
- 事务的提交时机在方法返回后（AOP 代理）
- 两者不同步，导致锁已释放但事务未提交
代码示例
java
@Service
public class HjlApplyExpenseServiceImpl {
@Override
@Transactional(rollbackFor = Exception.class)
public void upProcessResult(BpmProcessInstanceResultEvent event) {
synchronized (this) {
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, 2)
.eq(HjlApplyExpenseEntity::getId, 1));
} // ← synchronized 在这里释放
// 但 @Transactional 的提交在这里 ↓
} // ← 事务在方法返回后才提交（Spring AOP）
}执行顺序：
- 进入方法 → Spring AOP 开始事务
- synchronized 获取锁
- 执行 UPDATE
- synchronized 释放锁 ← 第一个释放点
- 方法返回
- Spring AOP 提交事务 ← 第二个释放点
时间差：第4步到第6步之间有时间差，其他线程可以获取锁但读到旧数据。
风险3：MyBatis-Plus 的 UPDATE 行为
问题场景
java
// 线程1：监听器更新状态
synchronized (this) {
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, 2)
.set(HjlApplyExpenseEntity::getCurrentNode, "结束")
.eq(HjlApplyExpenseEntity::getId, 1));
}
// 线程2：用户提交表单
synchronized (this) {
HjlApplyExpenseEntity entity = getById(1); // 查询
entity.setPaymentFrom("...");
entity.setExpenseType("...");
updateById(entity); // 更新
}MyBatis-Plus 的 updateById 行为：
java
HjlApplyExpenseEntity entity = getById(1);
// 查询结果：id=1, status=1, paymentFrom=null, ...
entity.setPaymentFrom("xxx");
updateById(entity);
// 生成的 SQL：
UPDATE hjl_apply_expense
SET payment_from = 'xxx',
status = 1, -- ❌ 会把 status 也更新！
expense_type = NULL, -- ❌ 会把所有字段都更新！
...
WHERE id = 1问题：
updateById会更新实体的所有字段- 即使字段没有修改，也会被更新为查询时的值
- 如果查询时
status=1，更新时也会设置为status=1 - 覆盖了监听器设置的
status=2❌
示例验证
时序：
T1 (监听器):
└─> UPDATE status=2 WHERE id=1 (提交)
T2 (用户提交表单):
├─> SELECT * FROM table WHERE id=1 (读到 status=1，旧快照)
├─> entity.setPaymentFrom("xxx")
└─> updateById(entity)
└─> UPDATE ... SET status=1, payment_from='xxx' WHERE id=1
└─> ❌ status 从 2 被覆盖为 1！结果：数据不一致，状态回退。
风险4：分布式场景下 synchronized 失效
单 JVM 有效，多 JVM 失效
单实例部署：
JVM-1:
线程1: synchronized(this@123456) 🔒
线程2: synchronized(this@123456) ← 等待线程1释放多实例部署：
JVM-1: JVM-2:
线程1: synchronized(this@123456) 🔒 线程2: synchronized(this@789012) 🔒
❌ 两个 this 是不同的对象！
❌ 两个线程可以同时执行！结果：分布式环境下，synchronized 无法保证并发安全。
hjl_oa 方案的问题
当前实现
HjlApplyExpenseServiceImpl.java：
java
@Service
public class HjlApplyExpenseServiceImpl {
@Override
@Transactional(rollbackFor = Exception.class)
public void disposeProcessFrom(WfProcess process) {
JSONObject jsonObject = process.getProcessFrom();
Long id = jsonObject.getLong("id");
synchronized (this) { // ❌ 锁粒度过粗
HjlApplyExpenseEntity entity = getById(id); // ⚠️ 可能读到旧数据
entity.setPaymentFrom(...);
entity.setExpenseType(...);
updateById(entity); // ❌ 可能覆盖监听器的更新
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public void upProcessResult(BpmProcessInstanceResultEvent event) {
Long id = Long.parseLong(event.getBusinessKey());
synchronized (this) { // ❌ 锁粒度过粗
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, 2)
.eq(HjlApplyExpenseEntity::getId, id));
} // ← synchronized 释放
// 但事务还未提交！
}
}问题总结
| 问题 | 描述 | 风险等级 |
|---|---|---|
| 事务隔离级别 | REPEATABLE READ 导致读到旧数据 | ⭐⭐⭐⭐⭐ 高 |
| synchronized 时机 | 锁释放但事务未提交 | ⭐⭐⭐⭐ 中高 |
| updateById 覆盖 | 更新所有字段，覆盖其他线程的修改 | ⭐⭐⭐⭐⭐ 高 |
| 分布式失效 | 多实例部署时 synchronized 失效 | ⭐⭐⭐⭐⭐ 高 |
结论
是的，存在严重的并发一致性风险！
并发场景测试
场景1：监听器先执行，用户后提交
时序图
时间轴：
0ms ─┬─ 线程1 (监听器) 开始
│ ├─ synchronized(this) 获取锁 🔒
│ ├─ @Transactional 开始
│ ├─ UPDATE status=2 WHERE id=1 (获取排他锁)
│
50ms ─┼─ 线程2 (用户提交表单) 开始
│ ├─ synchronized(this) ← 等待锁
│
100ms─┼─ 线程1 UPDATE 完成
│ └─ synchronized(this) 释放锁 🔓
│
├─ 线程2 synchronized(this) 获取锁 🔒
│ ├─ @Transactional 开始
│ ├─ SELECT * FROM table WHERE id=1
│ │ └─> ⚠️ 读到什么？
│
120ms─┼─ 线程1 事务提交
│ └─ 释放数据库排他锁
│
├─ 线程2 继续
│ ├─ 读到 status=? (取决于隔离级别和时机)
│ │
│ │ 情况A：如果 T2 在 T1 提交前开始事务
│ │ └─> 读到 status=1 (快照) ❌
│ │
│ │ 情况B：如果 T2 在 T1 提交后开始事务
│ │ └─> 读到 status=2 (最新值) ✅
│
├─ entity.setPaymentFrom("xxx")
└─> updateById(entity)
└─> UPDATE ... SET status=1, payment_from='xxx' WHERE id=1
└─> ❌ 如果读到 status=1，会覆盖为 1！测试代码
java
@Test
public void testConcurrentUpdate() throws Exception {
Long expenseId = 1L;
// 线程1：模拟监听器更新状态
Thread thread1 = new Thread(() -> {
service.upProcessResult(BpmProcessInstanceResultEvent.builder()
.businessKey(expenseId.toString())
.result(WfNotice.Type.FINISH.getCode())
.currentNode("结束")
.build());
System.out.println("线程1：更新状态为 FINISH");
});
// 线程2：模拟用户提交表单
Thread thread2 = new Thread(() -> {
try {
Thread.sleep(50); // 延迟50ms，确保线程1先执行
} catch (InterruptedException e) {
e.printStackTrace();
}
WfProcess process = new WfProcess();
process.setBusinessKey(expenseId);
JSONObject formData = new JSONObject();
formData.put("id", expenseId);
formData.put("paymentFrom", "xxx");
formData.put("expenseType", "yyy");
process.setProcessFrom(formData);
service.disposeProcessFrom(process);
System.out.println("线程2：提交表单数据");
});
// 启动线程
thread1.start();
thread2.start();
// 等待完成
thread1.join();
thread2.join();
// 验证结果
HjlApplyExpenseEntity entity = service.getById(expenseId);
System.out.println("最终状态：" + entity.getStatus());
System.out.println("最终数据：" + entity.getPaymentFrom());
// 期望：status=FINISH_STATUS, paymentFrom="xxx"
// 实际：status=? (可能被覆盖为 AUDIT_STATUS) ❌
}可能的结果
情况A：读到旧快照，覆盖状态
线程1：更新状态为 FINISH (status=2)
线程2：读到 status=1，提交表单
线程2：updateById 覆盖 status=1
最终状态：status=1 ❌ 错误！
最终数据：paymentFrom="xxx"情况B：读到新值，但仍可能覆盖
线程1：更新状态为 FINISH (status=2)
线程2：读到 status=2
线程2：业务逻辑抛出异常"流程已完成，无法修改" ✅ 正确！场景2：用户先提交，监听器后执行
时序图
时间轴：
0ms ─┬─ 线程1 (用户提交表单) 开始
│ ├─ synchronized(this) 获取锁 🔒
│ ├─ @Transactional 开始
│ ├─ SELECT * FROM table WHERE id=1 (status=1)
│
50ms ─┼─ 线程2 (监听器) 开始
│ ├─ synchronized(this) ← 等待锁
│
100ms─┼─ 线程1 业务逻辑
│ ├─ entity.setPaymentFrom("xxx")
│ ├─ UPDATE ... SET status=1, payment_from='xxx' WHERE id=1
│ └─ synchronized(this) 释放锁 🔓
│
├─ 线程2 synchronized(this) 获取锁 🔒
│ ├─ @Transactional 开始
│ ├─ UPDATE status=2 WHERE id=1 (等待线程1的排他锁)
│
120ms─┼─ 线程1 事务提交
│ └─ 释放数据库排他锁
│
├─ 线程2 获取排他锁
│ └─> UPDATE status=2 WHERE id=1 ✅ 成功
│
150ms─┼─ 线程2 事务提交
最终结果：
- status=2 ✅ 正确
- payment_from='xxx' ✅ 正确结论
这种情况下结果是正确的，因为：
- 线程1 先完成所有操作并提交
- 线程2 后更新 status，使用
lambdaUpdate只更新特定字段 - 不会覆盖
payment_from
场景3：极端并发（同时开始）
时序图
时间轴：
0ms ─┬─ 线程1 (监听器) 和 线程2 (用户提交) 同时开始
│
├─ 线程1: synchronized(this) 🔒
│ └─ 线程2: synchronized(this) ← 等待
│
├─ 线程1: @Transactional 开始
├─ 线程1: UPDATE status=2
│
50ms ─┼─ 线程1: synchronized 释放 🔓
│ └─ 线程2: synchronized 获取 🔒
│
60ms ─┼─ 线程1: 事务提交
│
├─ 线程2: @Transactional 开始
├─ 线程2: SELECT * FROM table WHERE id=1
│ └─> ⚠️ 如果事务在线程1提交前开始，读到 status=1
│ └─> ⚠️ 如果事务在线程1提交后开始，读到 status=2
│
└─ 线程2: updateById(entity)
└─> ❌ 可能覆盖 status问题
关键在于线程2的事务开始时机：
- 如果在线程1事务提交前开始 → 读到旧数据
- 如果在线程1事务提交后开始 → 读到新数据
synchronized 无法控制事务的开始时机。
解决方案
方案1：使用数据库锁（FOR UPDATE）⭐⭐⭐⭐⭐
核心思想
使用数据库的悲观锁（SELECT ... FOR UPDATE），确保读取时获取排他锁。
实现代码
java
@Service
public class HjlApplyExpenseServiceImpl {
@Resource
private HjlApplyExpenseMapper expenseMapper;
/**
* 使用悲观锁查询
*/
@Select("SELECT * FROM hjl_apply_expense WHERE id = #{id} FOR UPDATE")
HjlApplyExpenseEntity selectByIdForUpdate(Long id);
@Override
@Transactional(rollbackFor = Exception.class)
public void disposeProcessFrom(WfProcess process) {
Long id = process.getBusinessKey();
// ✅ 使用 FOR UPDATE 查询（获取排他锁）
HjlApplyExpenseEntity entity = selectByIdForUpdate(id);
// ✅ 此时已获取数据库行锁，其他事务必须等待
// ✅ 读取到的一定是最新数据
if (entity.getStatus() == StatusConstonts.AUDIT_STATUS) {
// 允许修改
entity.setPaymentFrom(...);
entity.setExpenseType(...);
// ✅ 使用 lambdaUpdate 只更新需要的字段
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getPaymentFrom, entity.getPaymentFrom())
.set(HjlApplyExpenseEntity::getExpenseType, entity.getExpenseType())
.eq(HjlApplyExpenseEntity::getId, id));
} else {
throw new BusinessException("当前状态不允许修改");
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public void upProcessResult(BpmProcessInstanceResultEvent event) {
Long id = Long.parseLong(event.getBusinessKey());
// ✅ 使用 lambdaUpdate，只更新特定字段
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, StatusConstonts.FINISH_STATUS)
.set(HjlApplyExpenseEntity::getCurrentNode, event.getCurrentNode())
.eq(HjlApplyExpenseEntity::getId, id));
}
}时序分析
线程1 (监听器):
├─ @Transactional 开始
├─ UPDATE status=2 WHERE id=1 (获取排他锁) 🔒
│
│ 线程2 (用户提交表单):
│ ├─ @Transactional 开始
│ ├─ SELECT * FROM ... WHERE id=1 FOR UPDATE
│ │ └─> ⚠️ 等待线程1释放排他锁
│
├─ 事务提交
└─ 释放排他锁 🔓
│
└─> 线程2 获取排他锁 🔒
├─ 读到 status=2 (最新值) ✅
├─ 判断状态，抛出异常 ✅
└─ 事务提交/回滚优点
- ✅ 强一致性：读取时加锁，保证读到最新数据
- ✅ 数据库级别保证：不依赖应用层锁
- ✅ 支持分布式：数据库锁在所有实例中有效
- ✅ 无需 synchronized：去掉 Java 锁
缺点
- ❌ 性能开销：会阻塞其他事务
- ❌ 可能死锁：需要注意锁的顺序
方案2：使用乐观锁（Version）⭐⭐⭐⭐
核心思想
使用版本号字段，更新时校验版本号，如果版本号不匹配则失败。
实现代码
java
@Data
@TableName("hjl_apply_expense")
public class HjlApplyExpenseEntity {
@TableId(type = IdType.AUTO)
private Long id;
private Integer status;
@Version // ✅ 乐观锁版本号
private Integer version;
private String paymentFrom;
private String expenseType;
}
@Service
public class HjlApplyExpenseServiceImpl {
@Override
@Transactional(rollbackFor = Exception.class)
public void disposeProcessFrom(WfProcess process) {
Long id = process.getBusinessKey();
// 查询（记录版本号）
HjlApplyExpenseEntity entity = getById(id);
Integer originalVersion = entity.getVersion(); // version=5
// 业务逻辑
if (entity.getStatus() == StatusConstonts.AUDIT_STATUS) {
entity.setPaymentFrom(...);
entity.setExpenseType(...);
// ✅ 使用 lambdaUpdate，只更新需要的字段
boolean success = update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getPaymentFrom, entity.getPaymentFrom())
.set(HjlApplyExpenseEntity::getExpenseType, entity.getExpenseType())
.eq(HjlApplyExpenseEntity::getId, id)
.eq(HjlApplyExpenseEntity::getVersion, originalVersion)); // ✅ 校验版本号
if (!success) {
// 版本号不匹配，说明数据被其他线程修改了
throw new BusinessException("数据已被修改，请刷新后重试");
}
} else {
throw new BusinessException("当前状态不允许修改");
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public void upProcessResult(BpmProcessInstanceResultEvent event) {
Long id = Long.parseLong(event.getBusinessKey());
// ✅ 乐观锁更新
boolean success = update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, StatusConstonts.FINISH_STATUS)
.set(HjlApplyExpenseEntity::getCurrentNode, event.getCurrentNode())
.eq(HjlApplyExpenseEntity::getId, id));
if (!success) {
throw new BusinessException("更新失败");
}
}
}SQL 示例
sql
-- 查询时记录版本号
SELECT * FROM hjl_apply_expense WHERE id = 1;
-- 结果: id=1, status=1, version=5
-- 更新时校验版本号
UPDATE hjl_apply_expense
SET payment_from = 'xxx',
version = 6 -- 自动 +1
WHERE id = 1 AND version = 5; -- ✅ 校验版本号
-- 如果其他事务已经更新，version != 5
-- 则 UPDATE 影响行数 = 0，返回 false时序分析
线程1 (监听器):
├─ SELECT * FROM ... WHERE id=1 (version=5)
├─ UPDATE ... SET status=2, version=6 WHERE id=1 AND version=5 ✅
└─ 提交 (version 变为 6)
线程2 (用户提交表单):
├─ SELECT * FROM ... WHERE id=1 (version=5，旧快照)
├─ 业务逻辑
└─> UPDATE ... SET payment_from='xxx', version=6 WHERE id=1 AND version=5
└─> ❌ 失败！因为 version 已经是 6 了
└─> 抛出异常，提示用户刷新优点
- ✅ 无锁：不阻塞其他事务
- ✅ 性能好：无等待
- ✅ 支持分布式：不依赖 JVM 锁
- ✅ 冲突提示：用户知道数据被修改
缺点
- ❌ 需要重试：失败后需要用户刷新
- ❌ 需要修改表结构：添加 version 字段
- ❌ ABA 问题：理论上存在（实际很少见）
方案3：使用分布式锁（Redisson）⭐⭐⭐⭐⭐
核心思想
使用 Redis 分布式锁，锁定具体的业务数据（而不是整个 Service）。
实现代码
java
@Service
public class HjlApplyExpenseServiceImpl {
@Resource
private RedissonClient redissonClient;
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void disposeProcessFrom(WfProcess process) {
Long id = process.getBusinessKey();
// ✅ 使用分布式锁，锁定具体的业务数据
String lockKey = "workflow:expense:form:" + id;
RLock lock = redissonClient.getLock(lockKey);
try {
// ✅ 尝试获取锁，最多等待10秒
if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
try {
// ✅ 此时已获取锁，其他线程必须等待
HjlApplyExpenseEntity entity = getById(id);
if (entity.getStatus() == StatusConstonts.AUDIT_STATUS) {
entity.setPaymentFrom(...);
entity.setExpenseType(...);
// ✅ 使用 lambdaUpdate，只更新需要的字段
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getPaymentFrom, entity.getPaymentFrom())
.set(HjlApplyExpenseEntity::getExpenseType, entity.getExpenseType())
.eq(HjlApplyExpenseEntity::getId, id));
} else {
throw new BusinessException("当前状态不允许修改");
}
} finally {
// ✅ 确保释放锁
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
} else {
throw new BusinessException("系统繁忙，请稍后重试");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new BusinessException("操作被中断");
}
}
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void upProcessResult(BpmProcessInstanceResultEvent event) {
Long id = Long.parseLong(event.getBusinessKey());
// ✅ 使用分布式锁
String lockKey = "workflow:expense:status:" + id;
RLock lock = redissonClient.getLock(lockKey);
try {
if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
try {
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, StatusConstonts.FINISH_STATUS)
.set(HjlApplyExpenseEntity::getCurrentNode, event.getCurrentNode())
.eq(HjlApplyExpenseEntity::getId, id));
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}时序分析
线程1 (监听器):
├─ 获取 Redis 锁: workflow:expense:status:1 🔒
├─ @Transactional 开始
├─ UPDATE status=2 WHERE id=1
├─ 事务提交
└─ 释放 Redis 锁 🔓
线程2 (用户提交表单):
├─ 尝试获取 Redis 锁: workflow:expense:form:1
│ └─> ⚠️ 如果线程1还持有锁，等待
│ └─> ✅ 线程1释放后，获取锁 🔒
├─ @Transactional 开始
├─ SELECT * FROM ... WHERE id=1
│ └─> ✅ 读到 status=2 (线程1已提交)
├─ 判断状态，抛出异常
└─ 释放 Redis 锁 🔓优点
- ✅ 支持分布式：多实例部署有效
- ✅ 锁粒度细：锁定具体业务数据
- ✅ 自动过期：避免死锁
- ✅ 跨事务保护：锁的范围包含整个事务
缺点
- ❌ 依赖 Redis：需要额外组件
- ❌ 网络开销：获取/释放锁需要网络请求
- ❌ 复杂度增加：需要处理锁异常
方案4：避免 updateById，使用 lambdaUpdate（⭐⭐⭐⭐⭐ 必须）
核心思想
永远不要使用 updateById(entity)，而是使用 lambdaUpdate 只更新需要的字段。
问题代码
java
// ❌ 错误：会更新所有字段
HjlApplyExpenseEntity entity = getById(1); // status=1（旧快照）
entity.setPaymentFrom("xxx");
updateById(entity);
// 生成的 SQL: UPDATE ... SET status=1, payment_from='xxx', ... WHERE id=1
// ❌ status 从 2 被覆盖为 1！正确代码
java
// ✅ 正确：只更新需要的字段
HjlApplyExpenseEntity entity = getById(1);
entity.setPaymentFrom("xxx");
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getPaymentFrom, entity.getPaymentFrom())
.set(HjlApplyExpenseEntity::getExpenseType, entity.getExpenseType())
.eq(HjlApplyExpenseEntity::getId, 1));
// 生成的 SQL: UPDATE ... SET payment_from='xxx', expense_type='yyy' WHERE id=1
// ✅ 不会覆盖 status！优点
- ✅ 简单有效：无需额外机制
- ✅ 避免覆盖：不会覆盖其他字段
- ✅ 性能好：减少数据传输
缺点
- ❌ 需要修改代码：所有
updateById都要改 - ❌ 代码冗长：需要手动指定每个字段
推荐方案
综合方案：分布式锁 + lambdaUpdate + 新事务（⭐⭐⭐⭐⭐）
实现代码
java
@Service
public class HjlApplyExpenseServiceImpl {
@Resource
private RedissonClient redissonClient;
/**
* 处理表单数据
*/
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void disposeProcessFrom(WfProcess process) {
Long id = process.getBusinessKey();
// ✅ 1. 使用分布式锁，锁定具体业务数据
String lockKey = "workflow:expense:" + id;
RLock lock = redissonClient.getLock(lockKey);
try {
if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
try {
// ✅ 2. 使用新事务（REQUIRES_NEW）
// ✅ 3. 读取最新数据
HjlApplyExpenseEntity entity = getById(id);
// ✅ 4. 状态校验
if (entity.getStatus().equals(StatusConstonts.FINISH_STATUS)) {
throw new BusinessException("流程已完成，无法修改");
}
// ✅ 5. 业务逻辑
String paymentFrom = process.getFormData("paymentFrom");
String expenseType = process.getFormData("expenseType");
// ✅ 6. 使用 lambdaUpdate，只更新需要的字段
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getPaymentFrom, paymentFrom)
.set(HjlApplyExpenseEntity::getExpenseType, expenseType)
.set(HjlApplyExpenseEntity::getIsExpense, "1")
.eq(HjlApplyExpenseEntity::getId, id));
log.info("处理表单数据成功: id={}", id);
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
} else {
throw new BusinessException("系统繁忙，请稍后重试");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new BusinessException("操作被中断");
}
}
/**
* 更新流程状态
*/
@Override
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void upProcessResult(BpmProcessInstanceResultEvent event) {
Long id = Long.parseLong(event.getBusinessKey());
Integer result = event.getResult();
String currentNode = event.getCurrentNode();
// ✅ 1. 使用分布式锁
String lockKey = "workflow:expense:" + id;
RLock lock = redissonClient.getLock(lockKey);
try {
if (lock.tryLock(10, 30, TimeUnit.SECONDS)) {
try {
// ✅ 2. 使用 lambdaUpdate，只更新特定字段
Integer status;
if (result.equals(WfNotice.Type.REJECT.getCode())) {
status = StatusConstonts.REJECT_STATUS;
} else if (result.equals(WfNotice.Type.FINISH.getCode())) {
status = StatusConstonts.FINISH_STATUS;
} else {
status = StatusConstonts.AUDIT_STATUS;
}
update(Wrappers.<HjlApplyExpenseEntity>lambdaUpdate()
.set(HjlApplyExpenseEntity::getStatus, status)
.set(HjlApplyExpenseEntity::getCurrentNode, currentNode)
.eq(HjlApplyExpenseEntity::getId, id));
log.info("更新流程状态成功: id={}, status={}", id, status);
} finally {
if (lock.isHeldByCurrentThread()) {
lock.unlock();
}
}
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("更新流程状态失败: id={}", id, e);
}
}
}关键点
- ✅ 分布式锁：使用 Redisson，支持多实例
- ✅ 新事务：
REQUIRES_NEW，避免与外层事务冲突 - ✅ 锁粒度细：锁定具体业务数据
workflow:expense:{id} - ✅ 只更新需要的字段：使用
lambdaUpdate - ✅ 状态校验：读取最新状态后再决定是否执行
- ✅ 异常处理：锁超时、中断等
时序保证
线程1 (监听器):
├─ 获取 Redis 锁: workflow:expense:1 🔒
├─ REQUIRES_NEW 事务开始
├─ UPDATE status=2, current_node='结束' WHERE id=1
├─ 事务提交
└─ 释放 Redis 锁 🔓
│
▼
线程2 (用户提交表单):
├─ 获取 Redis 锁: workflow:expense:1 🔒 ← ✅ 等待线程1释放
├─ REQUIRES_NEW 事务开始
├─ SELECT * FROM ... WHERE id=1
│ └─> ✅ 读到 status=2（最新值，线程1已提交）
├─ if (status == FINISH_STATUS)
│ └─> ✅ 抛出异常"流程已完成，无法修改"
└─ 事务回滚，释放锁 🔓优点
- ✅ 强一致性：保证读到最新数据
- ✅ 支持分布式：多实例部署有效
- ✅ 避免覆盖：只更新需要的字段
- ✅ 性能可控：锁粒度细，等待时间短
- ✅ 易于维护：代码清晰，逻辑明确
缺点
- ❌ 依赖 Redis：需要部署 Redis
- ❌ 改动较大：需要改造所有业务代码
总结
风险确认
是的，hjl_oa 的方案存在严重的并发一致性风险！
| 风险类型 | 描述 | 严重程度 |
|---|---|---|
| 事务隔离级别 | REPEATABLE READ 导致读到旧快照 | ⭐⭐⭐⭐⭐ 高 |
| synchronized 时机 | 锁释放但事务未提交 | ⭐⭐⭐⭐ 中高 |
| updateById 覆盖 | 更新所有字段，覆盖其他线程的修改 | ⭐⭐⭐⭐⭐ 高 |
| 分布式失效 | 多实例部署时 synchronized 失效 | ⭐⭐⭐⭐⭐ 高 |
推荐方案对比
| 方案 | 一致性 | 性能 | 复杂度 | 分布式支持 | 推荐度 |
|---|---|---|---|---|---|
| FOR UPDATE | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ | ⭐⭐ 低 | ✅ | ⭐⭐⭐⭐⭐ |
| 乐观锁 | ⭐⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐⭐⭐ 中 | ✅ | ⭐⭐⭐⭐ |
| 分布式锁 | ⭐⭐⭐⭐⭐ | ⭐⭐⭐⭐ | ⭐⭐⭐⭐ 高 | ✅ | ⭐⭐⭐⭐⭐ |
| lambdaUpdate | ⭐⭐⭐ | ⭐⭐⭐⭐⭐ | ⭐ 低 | ✅ | ⭐⭐⭐⭐⭐ 必须 |
最终建议
立即实施：
- ✅ 所有
updateById改为lambdaUpdate（必须） - ✅ 使用
REQUIRES_NEW事务传播（推荐）
短期实施： 3. ✅ 引入分布式锁（Redisson）（推荐） 4. ✅ 或使用 FOR UPDATE 悲观锁（备选）
长期优化： 5. ✅ 考虑乐观锁方案（如果性能要求高）
核心原则：
- 永远不要用
updateById(entity) - 使用分布式锁或数据库锁保证一致性
- 使用新事务避免与外层事务冲突