在分布式系统开发中，为了保证同一时刻只有一个节点能执行某个关键操作，我们通常会使用分布式锁，同时操作数据库时又会用到数据库事务来保证数据一致性，二者结合时如果处理不当，很容易出现操作原子性被破坏的问题。
冲突产生的核心原因
分布式锁和数据库事务的冲突本质是两个机制的生命周期不匹配。分布式锁的持有周期和数据库事务的执行周期如果没有对齐，就会出现锁已经释放但事务还没提交，或者其他节点在事务回滚前拿到锁执行操作的情况。
常见的冲突场景有两种：
- 先加分布式锁，再开启数据库事务，事务提交前就释放锁，此时其他节点拿到锁后读取到的还是事务未提交前的旧数据，后续事务提交后数据被覆盖，出现不一致。
- 先开启数据库事务，再加分布式锁，事务回滚时锁已经释放，其他节点拿到锁执行操作，回滚后之前的操作结果被丢弃，同样破坏原子性。
保证原子性的核心原则
要让分布式锁和数据库事务配合保证操作原子性，需要遵循两个核心原则：
- 分布式锁的持有周期必须完全覆盖数据库事务的整个生命周期，包括事务的提交或者回滚过程。
- 事务执行过程中如果出现异常，必须先完成事务的回滚，再释放分布式锁，避免锁提前释放导致其他节点操作异常数据。
具体实现方案
方案一：先加锁，事务完成后释放锁
这是最常用的方案，流程为：获取分布式锁 -> 开启数据库事务 -> 执行数据库操作 -> 提交/回滚事务 -> 释放分布式锁。无论事务执行成功还是失败，锁都会在事务完全结束后才释放。
以下是基于Spring Boot和Redis分布式锁的示例代码：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@Service
public class OrderService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private OrderMapper orderMapper;
// 分布式锁key前缀
private static final String LOCK_KEY_PREFIX = "order_lock_";
// 锁过期时间，避免死锁
private static final long LOCK_EXPIRE_TIME = 30L;
@Transactional(rollbackFor = Exception.class)
public void createOrder(String orderId) {
String lockKey = LOCK_KEY_PREFIX + orderId;
String lockValue = UUID.randomUUID().toString();
boolean lockSuccess = false;
try {
// 1. 获取分布式锁，使用setnx命令，过期时间30秒
lockSuccess = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, LOCK_EXPIRE_TIME, TimeUnit.SECONDS);
if (!lockSuccess) {
throw new RuntimeException("获取分布式锁失败，请稍后重试");
}
// 2. 执行数据库事务内的操作
// 查询订单是否存在
Order order = orderMapper.selectById(orderId);
if (order != null) {
throw new RuntimeException("订单已存在");
}
// 插入新订单
Order newOrder = new Order();
newOrder.setId(orderId);
newOrder.setStatus(1);
orderMapper.insert(newOrder);
// 事务会在方法执行结束后自动提交，如果出现异常会自动回滚
} catch (Exception e) {
// 异常抛出后事务会回滚
throw e;
} finally {
// 3. 无论事务成功还是失败，最后释放分布式锁
if (lockSuccess) {
// 释放锁时需要校验value，避免释放其他节点的锁
String currentValue = redisTemplate.opsForValue().get(lockKey);
if (lockValue.equals(currentValue)) {
redisTemplate.delete(lockKey);
}
}
}
}
}
方案二：事务内加锁，结合事务同步回调
如果需要在事务内部动态决定是否加锁，或者加锁逻辑依赖事务内的查询结果，可以使用Spring的事务同步管理器，在事务提交或者回滚完成后再执行释放锁的操作。
示例代码如下：
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronization;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
@Service
public class StockService {
@Autowired
private RedisTemplate<String, String> redisTemplate;
@Autowired
private StockMapper stockMapper;
private static final String STOCK_LOCK_KEY = "stock_lock";
@Transactional(rollbackFor = Exception.class)
public void deductStock(Long productId, Integer count) {
// 1. 先执行事务内的查询操作
Stock stock = stockMapper.selectByProductId(productId);
if (stock == null || stock.getNum() < count) {
throw new RuntimeException("库存不足");
}
// 2. 事务内获取分布式锁
String lockValue = UUID.randomUUID().toString();
Boolean lockSuccess = redisTemplate.opsForValue().setIfAbsent(STOCK_LOCK_KEY, lockValue, 30L, TimeUnit.SECONDS);
if (!lockSuccess) {
throw new RuntimeException("获取库存锁失败");
}
// 3. 注册事务同步回调，事务完成后释放锁
TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() {
@Override
public void afterCommit() {
// 事务提交后释放锁
String currentValue = redisTemplate.opsForValue().get(STOCK_LOCK_KEY);
if (lockValue.equals(currentValue)) {
redisTemplate.delete(STOCK_LOCK_KEY);
}
}
@Override
public void afterCompletion(int status) {
// 事务回滚后也释放锁
if (status == STATUS_ROLLED_BACK) {
String currentValue = redisTemplate.opsForValue().get(STOCK_LOCK_KEY);
if (lockValue.equals(currentValue)) {
redisTemplate.delete(STOCK_LOCK_KEY);
}
}
}
});
// 4. 执行库存扣减操作
stock.setNum(stock.getNum() - count);
stockMapper.updateById(stock);
}
}
注意事项
在实际使用中还需要注意以下几点：
- 分布式锁必须设置合理的过期时间，避免持有锁的节点宕机后出现死锁，同时过期时间要大于事务的最大执行时间，防止锁提前过期。
- 释放锁的时候必须校验锁的持有者，避免释放其他节点获取的锁，可以使用唯一标识作为锁的value，释放时对比value是否一致。
- 如果使用的是MySQL的乐观锁（比如版本号机制），和分布式锁结合时同样要遵循锁周期覆盖事务周期的原则，避免版本号校验失效。
- 尽量避免在长事务中持有分布式锁，长事务会延长锁的持有时间，降低系统的并发能力，同时增加事务超时的风险。
总结
MySQL中处理分布式锁和数据库事务的冲突，核心是保证分布式锁的生命周期完全覆盖事务的整个执行过程，无论事务是提交还是回滚，锁都在事务完全结束后才释放。开发者可以根据实际的业务场景选择先加锁再执行事务，或者结合事务同步回调在事务内加锁的方案，同时注意锁的过期时间和释放校验，就能有效保证操作的原子性，避免数据不一致问题。