分布式锁笔记

文章目录

单体应用锁的局限性
分布式锁
- 基于数据库实现分布式锁
- - 实现步骤：
  - 优缺点
- 基于Redis实现分布式锁
- 基于Zookeeper实现分布式锁
- 分布式锁的对比

单体应用锁的局限性 JDK提供的锁都是在一个JVM下起作用的，也就是在一个tomcat下是没问题的。当存在两个或两个以上的tomcat时，大量的并发请求分散到不同的tomcat上，每一个tomcat中是可以反正并发的产生，但是多个tomcat之间，每个tomcat中获得锁的这个请求，又产生了并发，会出现并发问题。单体应用锁的局限性是只能在一个JVM内加锁，而不能从这个应用层面加锁。
分布式锁分布式锁就是可以跨越多个JVM、跨越多个进程的锁。分布式锁都是通过第三方组件来实现的。目前比较流行的分布式解决方案有：

数据库，通过数据库可以实现分布式锁，但在高并发的情况下对数据库的压力比较大，所以很少使用。
Redis，借助Redis也可以实现分布式锁。Redis的Java客户端种类很多，使用的方法也不同。
Zookeeper：借助zookeeper瞬时节点。

基于数据库实现分布式锁实现步骤： 【分布式锁笔记】多个进程、多个线程访问共同的组件数据库。
通过select … for update 访问同一条数据库。
for update 锁定数据，其他线程只能等待。

SELECT @@autocommit;// 手动提交事务 SET @@autocommit=0;SELECT * FROM distribute_lock WHERE lock_key = 'demo' for update;COMMIT;

优缺点优点：简单方便、易于理解、易于操作
缺点：并发量大时，对数据库的压力较大。可以将锁的数据库与业务数据库分开
基于Redis实现分布式锁基于redis Setnx 实现
实现原理：利用NX的原子性，多个线程并发时，只有一个线程可以设置成功。
1、获取锁的redis命令
SET resource_name my_random_value NX PX 300000 resource_name：资源名称，可根据不同的业务区分不同的锁。
my_random_value：随机值，每个线程的随机值都不同，用于释放锁时的校验。
NX：key不存在时设置成功。
PX/EX：过期时间，出现异常时，锁可以过期失效。
2、释放锁采用Redis的delete命令，释放时需要校验之前设置的随机数，相同才能释放。为了保证这一操作的原子性，使用LUA脚本，LUA脚本如下：
if redis.call("get",KEYS[1]) == ARGV[1] thenreturn redis.call("del",KEYS[1])elsereturn 0end 3、Redis的一个客户端Redisson：
看门狗机制：
如果拿到分布式锁的节点宕机，且这个锁正好处于锁住的状态时，会出现锁死的状态，为了避免这种情况的发生，锁都会设置一个过期时间。
这样也存在一个问题，加入一个线程拿到了锁设置了30s超时，在30s后这个线程还没有执行完毕，锁超时释放了，就会导致问题。Redisson给出了自己的答案，就是 watch dog 自动延期机制。
Redisson提供了一个监控锁的看门狗，它的作用是在Redisson实例被关闭前，不断的延长锁的有效期，也就是说，如果一个拿到锁的线程一直没有完成逻辑，那么看门狗会帮助线程不断的延长锁超时时间，锁不会因为超时而被释放。默认情况下，看门狗的续期时间是30s，也可以通过修改Config.lockWatchdogTimeout来另行指定。另外Redisson 还提供了可以指定leaseTime参数的加锁方法来指定加锁的时间。超过这个时间后锁便自动解开了，不会延长锁的有效期。
看门狗机制相关博客链接：https://www.cnblogs.com/jelly12345/p/14699492.html
加锁的代码：

 RFuture tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {return evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,// 如果锁不存在，则设置值和过期时间"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +// 如果锁已存在，并且锁的是当前线程，则通过hincrby给数值递增1"if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +"redis.call('hincrby', KEYS[1], ARGV[2], 1); " +"redis.call('pexpire', KEYS[1], ARGV[1]); " +"return nil; " +"end; " +// 如果锁已存在，但非本线程，则返回过期时间ttl"return redis.call('pttl', KEYS[1]);",Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));}

解锁的代码：

protected RFuture unlockInnerAsync(long threadId) {return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,// 如果锁的线程和已存在锁的线程不是同一个线程，返回null"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +"return nil;" +"end; " +// 通过hincrby递减1的方式，释放一次锁//若剩余次数大于0，则刷新过期时间"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +"if (counter > 0) then " +"redis.call('pexpire', KEYS[1], ARGV[2]); " +"return 0; " +"else " +//否则证明锁已经释放，删除key并发布锁释放的消息"redis.call('del', KEYS[1]); " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end; " +"return nil;",Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, internalLockLeaseTime, getLockName(threadId));}

基于Zookeeper实现分布式锁 Zookeeper的观察器：可以监测某个节点的变化，3个方法：getData()，getChildren()，exists()
实现原理：

利用Zookeeper的瞬时有序节点的特性
多线程并发创建瞬时节点时，得到有序的序列
序号最小的线程获得锁
其他的线程则监听自己序号的前一个序号
前一个线程执行完成，删除自己的序号的节点
下个序号的线程得到通知，继续执行

package com.lank;import lombok.extern.slf4j.Slf4j;import org.apache.zookeeper.*;import org.apache.zookeeper.data.Stat;import java.io.IOException;import java.nio.charset.StandardCharsets;import java.util.Collections;import java.util.List;import java.util.Objects;/** * @author lank * @since 2022-03-07 20:04 */public class ZkLock implements Watcher {private final ZooKeeper zooKeeper;private String zNode;public ZkLock() throws IOException {this.zooKeeper = new ZooKeeper("localhost:2181", 50000000, this);}public boolean getLock(String lockCode) throws Exception {try {// 判断根节点是否存在Stat stat = zooKeeper.exists("/" + lockCode, false);if (Objects.isNull(stat)) {zooKeeper.create("/" + lockCode,lockCode.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);}// 创建瞬时有序节点zNode = zooKeeper.create("/" + lockCode + "/" + lockCode + "_",lockCode.getBytes(),ZooDefs.Ids.OPEN_ACL_UNSAFE,// 瞬时有序节点CreateMode.EPHEMERAL_SEQUENTIAL);System.out.println("zNode:" + zNode);// 判断该节点是不是序号最小的节点List nodeChildren = zooKeeper.getChildren("/" + lockCode, false);// 排序-升序Collections.sort(nodeChildren);String firstNode = nodeChildren.get(0);if (zNode.endsWith(firstNode)) {// 成功获取锁return true;}// 没获取锁则监听前一个节点String lastNode = firstNode;for (String node : nodeChildren) {if (zNode.endsWith(node)) {// 这里有监听，节点消失进入process()方法Stat exists = zooKeeper.exists("/" + lockCode + "/" + lastNode, true);break;} else {lastNode = node;}}synchronized (this) {wait();}} catch (Exception e) {e.printStackTrace();} finally {close();}return false;}@Overridepublic void process(WatchedEvent watchedEvent) {if (watchedEvent.getType().equals(Event.EventType.NodeDeleted)) {synchronized (this) {notify();}}}public void close() throws Exception {zooKeeper.delete(zNode, -1);zooKeeper.close();System.out.println("我已经释放了锁");}}

分布式锁的对比方式优点缺点数据库实现简单、易于理解对数据压力大Redis易于理解自己实现，不支持阻塞，不能续租Zookeeper支持阻塞需理解Zookeeper、程序复杂Curator提供锁的方法依赖Zookeeper，强一致性Redission提供锁的方法，可阻塞