分布式锁的实现
使用数据库乐观锁
乐观锁通常实现基于数据版本(version)的记录机制实现的，比如有一张红包表（t_bonus），有一个字段(left_count)记录礼物的剩余个数，用户每领取一个奖品，对应的left_count减1，在并发的情况下如何要保证left_count不为负数，乐观锁的实现方式为在红包表上添加一个版本号字段（version），默认为0。
异常实现流程
-- 可能会发生的异常情况
-- 线程1查询，当前left_count为1，则有记录
select * from t_bonus where id = 10001 and left_count > 0
-- 线程2查询，当前left_count为1，也有记录
select * from t_bonus where id = 10001 and left_count > 0
-- 线程1完成领取记录，修改left_count为0,
update t_bonus set left_count = left_count - 1 where id = 10001
-- 线程2完成领取记录，修改left_count为-1，产生脏数据
update t_bonus set left_count = left_count - 1 where id = 10001
通过乐观锁实现
-- 添加版本号控制字段
ALTER TABLE table ADD COLUMN version INT DEFAULT '0' NOT NULL AFTER t_bonus;
-- 线程1查询，当前left_count为1，则有记录，当前版本号为1234
select left_count, version from t_bonus where id = 10001 and left_count > 0
-- 线程2查询，当前left_count为1，有记录，当前版本号为1234
select left_count, version from t_bonus where id = 10001 and left_count > 0
-- 线程1,更新完成后当前的version为1235，update状态为1，更新成功
update t_bonus set version = 1235, left_count = left_count-1 where id = 10001 and version = 1234
-- 线程2,更新由于当前的version为1235，udpate状态为0，更新失败，再针对相关业务做异常处理
update t_bonus set version = 1235, left_count = left_count-1 where id = 10001 and version = 1234
Redis Lua 脚本 实现分布式锁
获取锁,采用的是
lua脚本,这样可以保证加锁 和 设置失效时间的原子性。
避免获取锁成功后,异常退出,造成锁无法释放的问题。
1.首先在application.properties中配置
lua.lockScript=if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then return redis.call('expire',KEYS[1],ARGV[2]) else return 0 end
lua.releaseLockScript=if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end
- 关于如何读取自定义的配置属性
咱们可以参考读取
application.properties的三种方式：https://blog.csdn.net/qq_37171353/article/details/78005845
@Autowired
private Environment env;
//然后读取
env.getProperty("lua.lockScript");
2. 加锁和释放锁的实现
package com.nezha.learn.demo.util;
import org.springframework.core.env.Environment;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import java.util.Collections;
public class RedisTool {
public static boolean getLock(StringRedisTemplate stringRedisTemplate, Environment env, String key, String requestId, Long expiresTime) {
DefaultRedisScript<Long> longDefaultRedisScript = new DefaultRedisScript<>(env.getProperty("lua.lockScript"), Long.class);
Long result = stringRedisTemplate.execute(longDefaultRedisScript, Collections.singletonList(key), requestId,String.valueOf(expiresTime));
return result == 1;
}
public static boolean releaseLock(StringRedisTemplate stringRedisTemplate, Environment env, String key, String requestId) {
DefaultRedisScript<Long> longDefaultRedisScript = new DefaultRedisScript<>(env.getProperty("lua.releaseLockScript"), Long.class);
Long result = stringRedisTemplate.execute(longDefaultRedisScript, Collections.singletonList(key), requestId);
return result == 1;
}
}
3. 使用该工具类
public void getLock(String name){
String key = "demo:"+name;
boolean lock = RedisTool.getLock(redisTemplate,env,key,name,20L);
if (lock){
try{
logger.info("执行睡眠...");
Thread.sleep(10000);
}catch (InterruptedException e){
logger.info("发生错误{}",e);
}finally {
RedisTool.releaseLock(redisTemplate,env,key,name);
}
}
}
Zookeeper实现分布式锁
参考文献:
分布式利器Zookeeper（二）:分布式锁
!!!!curator笔记-分布式锁的实现与原理 ---这篇源码分析的还是不错的。
zookeeper的分布式锁实现原理
ZooKeeper机制规定同一个目录下只能有一个唯一的文件名，zookeeper上的一个znode看作是一把锁，通过createznode的方式来实现。所有客户端都去创建/lock/${lock_name}_lock节点，最终成功创建的那个客户端也即拥有了这把锁，创建失败的可以选择监听继续等待，还是放弃抛出异常实现独占锁。
InterProcessMutex原理总结
InterProcessMutex通过在zookeeper的某路径节点下创建临时序列节点来实现分布式锁，即每个线程（跨进程的线程）获取同一把锁前，都需要在同样的路径下创建一个节点，节点名字由uuid + 递增序列组成。而通过对比自身的序列数是否在所有子节点的第一位，来判断是否成功获取到了锁。当获取锁失败时，它会添加watcher来监听前一个节点的变动情况，然后进行等待状态。直到watcher的事件生效将自己唤醒，或者超时时间异常返回。
基于Curator实现
上代码：
首先引入pom依赖：curator-recipes
@Service
public class LockServiceZK {
private final static Logger logger = LoggerFactory.getLogger(LockServiceZK.class);
private CuratorFramework zk;
private InterProcessMutex lock;
@Value("${zookeeper.connect}")
private String hosts;
@Value("${zookeeper.retry}")
private Integer retry = 3;
@Value("${zookeeper.timeout.connect}")
private Integer connectTimeout = 30;
@Value(("${zookeeper.timeout.session}"))
private Integer sessionTimeout = 300;
@Value("${zookeeper.root}")
private String root;
@PostConstruct
private void start() throws Exception {
ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, retry);
zk = CuratorFrameworkFactory.builder()
.connectString(hosts)
.connectionTimeoutMs(connectTimeout)
.sessionTimeoutMs(sessionTimeout)
.retryPolicy(retryPolicy)
.build();
zk.start();
lock = new InterProcessMutex(zk, Paths.get(root, "/lock").toString());
}
@PreDestroy
private void stop() throws Exception {
if(zk != null) {
zk.close();
}
}
public boolean acquire() throws Exception {
return lock.acquire(1,TimeUnit.SECONDS);
}
public void release() throws Exception {
lock.release();
}
}
然后是测试类
@RestController
public class RestApi {
private final static Logger logger = LoggerFactory.getLogger(RestApi.class);
@Autowired
private LockServiceZK zkLockService;
@RequestMapping(value = "/lock1")
public String hello(){
try {
boolean getLock = zkLockService.acquire();
if (getLock){
//Do something
logger.info("{},拿到zk锁并且睡眠了20秒",this.getClass().getName());
}else {
logger.info("获取锁失败");
}
}catch (Exception e){
logger.info("1,并发竞争锁失败");
}finally {
try {
zkLockService.release();
logger.info("1,释放锁成功了。。。");
}catch (Exception e){
logger.info("1,释放锁失败");
}
}
return "Hello World";
}
}
参考文献
参考文献：Java分布式锁三种实现方案---https://www.jianshu.com/p/535efcab356d
实践基于Redis的分布式锁：https://www.jianshu.com/p/e72baf5e5617
Redis Lua脚本 实现分布式锁：https://www.jianshu.com/p/1add6858f687
7 张图讲清楚ZooKeeper分布式锁实现原理:https://mp.weixin.qq.com/s/smuIGalT7Qvjy8evVA2mBg ----这篇文章讲的很好强烈推荐
网友评论