如何使用mysql实现可重入的分布式锁
目录
- 什么是分布式锁？
- 如何实现分布式锁？
- 定义分布式表结构
- 定义锁统一接口
- 使用mysql来实现分布式锁
① 生成线程标记ID
② 加锁
③ 解锁
④ 重置锁
- 写在最后
1. 什么是分布式锁？
百度百科：分布式锁是控制分布式系统之间同步访问共享资源的一种方式。
ㅤ如引用所述，分布式锁是一种用于在分布式系统中对资源进行同步访问的机制。在分布式系统中，多个节点同时访问某个共享资源时，需要确保资源的一致性和正确性。分布式锁可以通过协调多个节点之间的操作，保证在同一时间内只有一个节点能够访问该资源，从而避免竞态条件和数据不一致的问题。
2. 如何实现分布式锁？
ㅤ基于数据库的分布式锁：使用数据库的事务机制来实现分布式锁，通过在数据库中创建一个唯一约束或者加锁表来保证同一时间只有一个节点能够获得锁。
ㅤ基于共享存储的分布式锁：使用共享存储如Redis、ZooKeeper等来实现分布式锁，通过在共享存储中创建占位节点或者临时节点来表示锁的状态。
ㅤ基于乐观锁的分布式锁：通过使用版本号或者时间戳等方式，在修改资源时判断是否被其他节点修改，从而实现资源的同步访问。
ㅤ分布式锁的实现需要考虑各种复杂条件，如锁的可重入性、死锁的处理、锁的过期时间等。因此，使用分布式锁时需要谨慎设计和合理选择实现方式，以保证系统的性能和可靠性。
3. 定义分布式表结构
create table if not exists t_distributed_lock
(
id int auto_increment primary key,
lock_key varchar(255) default '' not null comment '锁key',
thread_ident varchar(255) default '' not null comment '线程标识',
timeout mediumtext not null comment '超时时间',
reentrant_count int default 0 not null comment '可重入次数',
version int default 1 not null comment '版本号（用于乐观锁）'
)
comment '分布式锁';
4. 定义锁统一接口
public interface ILock<T> {
/**
* 加锁（支持可重入）
*
* @param lockKey 锁key
* @param lockTimeoutMillisecond 锁超时时间（1.锁不是无限存在的 2.为可能存在的死锁做兜底）
* @param getLockTimeoutMillisecond 获取锁的超时时间（获取锁的时间应该也是有限的）
*/
boolean lock(String lockKey, long lockTimeoutMillisecond, long getLockTimeoutMillisecond);
/**
* 解锁
*
* @param lockKey 锁key
*/
boolean unLock(String lockKey);
/**
* 重置锁
*
* @param t 锁对象
*/
boolean resetLock(T t);
}
ㅤ定义关于锁的统一接口，对参数类型进行泛化。
5. 使用mysql来实现分布式锁
5.1 生成线程标记ID
import cn.hutool.core.lang.UUID;
import cn.wxroot.learn.distributedLock.mysql.entity.DistributedLock;
import cn.wxroot.learn.distributedLock.mysql.mapper.DistributedLockMapper;
import cn.wxroot.learn.distributedLock.ILock;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import java.util.Objects;
@Service
@Slf4j
public class MysqlLock extends ServiceImpl<DistributedLockMapper, DistributedLock> implements ILock<DistributedLock> {
ThreadLocal<String> threadIdent = new ThreadLocal<>();
/**
* 获取线程标记ID
*
* @return 线程标记ID
*/
private String getThreadIdentId() {
if (Objects.isNull(threadIdent.get())) {
threadIdent.set(UUID.randomUUID().toString());
}
return threadIdent.get();
}
// 加锁、解锁、重置锁
}
ㅤ创建mysql方式的锁实现类。方法getThreadIdentId()来生成线程的唯一标识。
5.1 加锁
@Override
public boolean lock(String lockKey, long lockTimeoutMillisecond, long getLockTimeoutMillisecond) {
boolean result = false;
long startTime = System.currentTimeMillis();
String threadIdentId = this.getThreadIdentId();
while (true) {
if (startTime + getLockTimeoutMillisecond < System.currentTimeMillis()) {
log.info("获取锁超时！");
break;
}
QueryWrapper<DistributedLock> distributedLockQueryWrapper = new QueryWrapper<>();
distributedLockQueryWrapper.lambda().eq(true, DistributedLock::getLockKey, lockKey);
DistributedLock distributedLock = this.getOne(distributedLockQueryWrapper);
if (Objects.nonNull(distributedLock)) {
if (StringUtils.isNotEmpty(distributedLock.getThreadIdent())) {
if (threadIdentId.equals(distributedLock.getThreadIdent())) {
log.info("重入锁+1");
distributedLock.setReentrantCount(distributedLock.getReentrantCount() + 1);
distributedLock.setTimeout(System.currentTimeMillis() + lockTimeoutMillisecond);
this.updateById(distributedLock);
result = true;
break;
} else {
// 其他线程锁定了该lockKey，需要等待
if (distributedLock.getTimeout() < System.currentTimeMillis()) {
this.resetLock(distributedLock);
} else {
try {
log.info("休眠");
Thread.sleep(100);
} catch (InterruptedException ignored) {
}
}
}
} else {
log.info("占用锁");
distributedLock.setThreadIdent(threadIdentId);
distributedLock.setReentrantCount(1);
distributedLock.setTimeout(System.currentTimeMillis() + lockTimeoutMillisecond);
this.updateById(distributedLock);
result = true;
break;
}
} else {
log.info("创建新记录");
DistributedLock addDistributedLock = DistributedLock.builder()
.lockKey(lockKey)
.threadIdent(threadIdentId)
.timeout(System.currentTimeMillis() + lockTimeoutMillisecond)
.reentrantCount(1)
.version(1)
.build();
this.save(addDistributedLock);
result = true;
break;
}
}
return result;
}
ㅤ实现加锁的方法boolean lock(···)应考虑以下几点：
ㅤ1. 使用锁的时间应该有限的，即不能永久占有锁。
ㅤ2. 获取锁的过程应该是有限的，即指定时间内获取不到锁应该返回。
5.1 解锁
@Override
public boolean unLock(String lockKey) {
boolean result = false;
String threadIdentId = this.getThreadIdentId();
log.info("解锁");
QueryWrapper<DistributedLock> distributedLockQueryWrapper = new QueryWrapper<>();
distributedLockQueryWrapper.lambda().eq(true, DistributedLock::getLockKey, lockKey);
DistributedLock distributedLock = this.getOne(distributedLockQueryWrapper);
if (Objects.nonNull(distributedLock)) {
if (distributedLock.getThreadIdent().equals(threadIdentId)) {
if (distributedLock.getReentrantCount() > 1) {
distributedLock.setReentrantCount(distributedLock.getReentrantCount() - 1);
this.updateById(distributedLock);
result = true;
} else {
result = this.resetLock(distributedLock);
}
}
}
return result;
}
ㅤ基于可重入锁的特性，在进行解锁操作时，应该有所判断。
5.1 重置锁
@Override
public boolean resetLock(DistributedLock distributedLock) {
log.info("重置锁");
distributedLock.setThreadIdent("");
distributedLock.setReentrantCount(0);
this.updateById(distributedLock);
return true;
}
转载请注明出处