前言

java有synchronize和Lock，mysql 修改类的sql也带有锁。锁定数据状态，让数据状态在并发场景，按我们预想逻辑进行状态转移，然而在分布式，集群的情况下，怎么去锁定数据状态呢

• 数据库的分布式锁方案
• 基于redis实现分布式锁
• 基于zookeeper实现分布式锁

关注公众号，一起交流，微信搜一搜: 潜行前行

github地址，感谢star

数据库的分布式锁方案

数据库分布锁的难点

• 单点故障？ 数据库可以多搞个数据库备份
• 没有失效时间？ 每次加锁时，插入一个期待的有效时间；A：定时任务，隔一段时间清理时间失效锁。B：下次加锁时则先判断当前时间是否大于锁的有效时间，以此判断锁是否失效
• 不可重入？ 在数据加锁时加入一个幂等唯一值字段，下次获取时，先判断这个字段是否一致，一致则说明是当前操作重入操作

基于redis实现分布式锁

• redis 是一个快速访问的高性能服务，相比数据库，在redis实现锁比直接在数据库的数据加锁，性能好。同时也为数据库减压，减少事务执行因为锁的问题阻塞
• 引入jedis

<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.9.0</version>
</dependency>

setnx + expire

• setnx + expire 存在死锁的问题。setnx()方法作用就是SET IF NOT EXIST，expire()方法就是给锁加一个过期时间。由于这是两条Redis命令，不具有原子性

Long result = jedis.setnx(lockKey, requestId);
if (result == 1) {
    // 这里程序突然崩溃，则无法设置过期时间，将发生死锁
    jedis.expire(lockKey, expireTime);
}

lua脚本(正确方式)

• lua脚本在Redis的执行过程是原子性，要么成功，要么失败。

// setnx + expire 放在lua脚本执行
String script = "if redis.call('setnx',KEYS[1],ARGV[1]) == 1 then" +
            " redis.call('expire',KEYS[1],ARGV[2]) return 1 else return 0 end";  
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));
if(result.equals(1)){
    .... //加锁成功的操作
}

set {key} {value} nx ex {second} (正确方式)

• 这是Redis的SET指令扩展参数，具有原子性

String lockKey = "锁的KEY值";//固定的
String requestId = "当次加锁操作的唯一标识";
int  expireTime = 1000;//失效时间
String result = jedis.set(lockKey, requestId, "NX", "PX", expireTime);

删除redis分布锁

//-------- 错误方式 ------------
// 判断加锁与解锁是不是同一个客户端
if (requestId.equals(jedis.get(lockKey))) {
    // 若在此时，这把锁突然不是这个客户端的，则会误解锁
    jedis.del(lockKey);
}
//-------- 正确的方式 使用 lua ------------
String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
Object result = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(requestId));

基于Redlock算法实现分布式锁

• 以上redis分布锁的缺点就是它加锁时只作用在一个Redis节点上，即使redis通过sentinel保证高可用，如果这个master节点由于某些原因发生了主从切换，那么就会出现锁丢失的情况。redis主从同步不能保证一致性，master会优先返回结果，在同步数据到slave
• 例如：在redis的master节点上拿到了锁 -> 这个加锁的key还没有同步到slave节点 -> master故障，发生故障转移，slave节点升级为master节点 -> 导致锁丢失
• RedLock算法的实现步骤

1: 获取当前时间，以毫秒为单位

2: 按顺序向5个master节点请求加锁。客户端设置网络连接和响应超时时间，并且超时时间要小于锁的失效时间。（假设锁自动失效时间为10秒，则超时时间一般在5-50毫秒之间,我们就假设超时时间是50ms吧）。如果超时，跳过该master节点，尽快去尝试下一个master节点。

3: 客户端使用当前时间减去开始获取锁时间（即步骤1记录的时间），得到获取锁使用的时间。当且仅当超过一半（N/2+1，这里是5/2+1=3个节点）的Redis master节点都获得锁，并且使用的时间小于锁失效时间时，锁才算获取成功。（如上图，10s> 30ms+40ms+50ms+4m0s+50ms）

4: 如果取到了锁，key的真正有效时间就变啦，需要减去获取锁所使用的时间。

5: 如果获取锁失败（没有在至少N/2+1个master实例取到锁，有或者获取锁时间已经超过了有效时间），客户端要在所有的master节点上解锁（即便有些master节点根本就没有加锁成功，也需要解锁，以防止有些漏网之鱼）

<dependency>
	<groupId>org.springframework.boot</groupId>
	<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
      <groupId>org.redisson</groupId>
      <artifactId>redisson</artifactId>
      <version>3.4.3</version>
</dependency>

• 代码示例

Config config = new Config().useSingleServer().setAddress("127.0.0.1:6380").setDatabase(0);
RedissonClient rLock1 = Redisson.create(config);
config = new Config().useSingleServer().setAddress("127.0.0.1:6381").setDatabase(0);
RedissonClient rLock2 = Redisson.create(config);
config = new Config().useSingleServer().setAddress("127.0.0.1:6382").setDatabase(0);
RedissonClient rLock3 = Redisson.create(config);
//初始化
String lockKey = "XXX";
RLock rLock1 = redissonRed1.getLock(lockKey);
RLock rLock2 = redissonRed2.getLock(lockKey);
RLock rLock3 = redissonRed2.getLock(lockKey);
RedissonRedLock rLock = new RedissonRedLock(rLock1,rLock2,rLock3);
//加锁
rLock.lock();
//释放
rLock.unlock();

基于 zookeeper 实现分布式锁

• maven引入

<dependency>
   <groupId>org.apache.curator</groupId>
   <artifactId>curator-recipes</artifactId>
   <version>2.4.1</version>
</dependency>

• Redlock算法往往需要多个redis集群才能实现，东西越多，就越容易出错。但是如何实现一个高效高可用的分布式锁呢 ? zookeeper
• zookeeper特点
  • 最终一致性：客户端的操作状态会在 zookeepr 集群保持一致
  • 可靠性：zookeeper 集群具有简单、健壮、良好的性能
  • 原子性：操作只能成功或者失败，没有中间状态
  • 时间顺序性：如果消息 A 在消息 B 发布，则 A 则排在 B 前面
• zookeeper 临时顺序节点：临时节点的生命周期和客户端会话绑定。也就是说，如果客户端会话失效，那么这个节点就会自动被清除掉（可解决分布式锁的自动失效）。另外，在临时节点下面不能创建子节点，集群zk环境下，同一个路径的临时节点只能成功创建一个
• zookeeper 监视器：zookeeper创建一个节点时，会注册一个该节点的监视器，当节点状态发生改变时，watch会被触发，zooKeeper将会向客户端发送一条通知
• zookeeper 分布式锁原理

创建临时有序节点，每个线程均能创建节点成功，但是其序号不同，只有序号最小的可以拥有锁，其它线程只需要监听比自己序号小的节点状态即可

1: 在指定的节点下创建一个锁目录lock

2: 线程X进来获取锁在lock目录下，并创建临时有序节点

3: 线程X获取lock目录下所有子节点，并获取比自己小的兄弟节点，如果不存在比自己小的节点，说明当前线程序号最小，顺利获取锁

4: 此时线程Y进来创建临时节点并获取兄弟节点，判断自己是否为最小序号节点，发现不是，于是设置监听（watch）比自己小的节点（这里是为了发生上面说的羊群效应）

5: 线程X执行完逻辑，删除自己的节点，线程Y监听到节点有变化，进一步判断自己是已经是最小节点，顺利获取锁

• 代码实例

//初始化
CuratorFramework curatorFramework= CuratorFrameworkFactory.newClient("zookeeper1.tq.master.cn:2181",new ExponentialBackoffRetry(1000,3));
curatorFramework.start();
//创建临时节点锁
String lockPath = "/distributed/lock/";//根节点
//可重入排它锁
String lockName = "xxxx";
InterProcessMutex interProcessMutex = new InterProcessMutex(curatorFramework, lockPath + lockName);
//加锁
interProcessMutex.acquire(2, TimeUnit.SECONDS)
//释放锁
if(interProcessMutex.isAcquiredInThisProcess()){
    interProcessMutex.release();
    curatorFramework.delete().inBackground().forPath(lockPath + lockName);
}

欢迎指正文中错误

参数文章

• 记一次分布式锁-基于数据库
• Redis分布式锁的正确实现方式
• redis实现分布式锁,单机-集群-红锁
• 如何能通俗的讲解Zookeeper分布式锁的应用场景？
• 基于Zookeeper开源客户端Curator实现分布式锁

最后

以上就是眯眯眼唇膏为你收集整理的框架篇：分布式锁前言参数文章的全部内容，希望文章能够帮你解决框架篇：分布式锁前言参数文章所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。