分布式 | 分布式锁的实现

双鬼带单

发布于 2021-03-18 21:22:44

7300

发布于 2021-03-18 21:22:44

文章被收录于专栏：CodingToDieCodingToDie

分布式锁的实现

在常见的分布式锁中有以下三种实现：

Redis 实现
Zookeeper 实现
数据库实现

1. 基于Redis 的实现1.1 实现原理1.2 实现方式1.2.1 原生代码1.2.2 Spring Redis Lock 实现1.2.3 Redission 实现1.3 优缺点2. 基于 Zookeeper 的实现2.1 实现原理2.2 使用2.2.1 使用 spring-integration-zookeeper 实现2.2.2 使用 Apache Curator2.3 优缺点3. 基于数据库的实现3.1 实现原理3.2 优缺点4. 对比

1. 基于Redis 的实现

在 Redis 中有个3个重要命令，通过这三个命令可以实现分布式锁

setnx key val：当且仅当key不存在时，set一个key为val的字符串，返回1；若key存在，则什么都不做，返回0。
expire key timeout：为key设置一个超时时间，单位为second，超过这个时间 key 会自动删除。
delete key：删除key

1.1 实现原理

获取锁的时候，使用 setnx 命令设置一个 kv，其中 k 为锁的名字，v 为一个随机数字，如果成功设置则获取锁，如果未设置成功则失败。如果设置了尝试获取锁的最大的时间，则需要在最大时间内，不停的重复该步骤，直到获取锁或者超过最大时间才能结束。
使用 expire 命令为刚才创建的 key 设置超时一个合理的超时时间，防止在无法正确释放锁的时候也能通过超时时间进行释放，这个超时时间需要根据项目请求情况进行设置；
释放锁的时候，通过 v 判断是不是还是原来的锁，若是该锁，则执行 delete 进行锁释放。

1.2 实现方式

1.2.1 原生代码

 1public class DistributedLock implements Lock {
 2
 3  private static JedisPool JEDIS_POOL = null;
 4  private static int EXPIRE_SECONDS = 60;
 5
 6  public static void setJedisPool(JedisPool jedisPool, int expireSecond) {
 7    JEDIS_POOL = jedisPool;
 8    EXPIRE_SECONDS = expireSecond;
 9  }
10
11  private String lockKey;
12  private String lockValue;
13
14  private DistributedLock(String lockKey) {
15    this.lockKey = lockKey;
16  }
17
18  public static DistributedLock newLock(String lockKey) {
19    return new DistributedLock(lockKey);
20  }
21
22  @Override
23  public void lock() {
24    if (!tryLock()) {
25      throw new IllegalStateException("未获取到锁");
26    }
27  }
28
29  @Override
30  public void lockInterruptibly() throws InterruptedException {
31  }
32
33  @Override
34  public boolean tryLock() {
35    return tryLock(0, null);
36  }
37
38  @Override
39  public boolean tryLock(long time, TimeUnit unit) {
40    Jedis conn = null;
41    String retIdentifier = null;
42    try {
43      conn = JEDIS_POOL.getResource();
44      lockKey = UUID.randomUUID().toString();
45
46      // 获取锁的超时时间，超过这个时间则放弃获取锁
47      long end = 0;
48      if (time != 0) {
49        end = System.currentTimeMillis() + unit.toMillis(time);
50      }
51
52      do {
53        if (conn.setnx(lockKey, lockValue) == 1) {
54          conn.expire(lockKey, EXPIRE_SECONDS);
55          return true;
56        }
57
58        try {
59          Thread.sleep(10);
60        } catch (InterruptedException e) {
61          Thread.currentThread().interrupt();
62        }
63      } while (System.currentTimeMillis() < end);
64    } catch (JedisException e) {
65      if (lockValue.equals(conn.get(lockKey))) {
66        conn.del(lockKey);
67      }
68      e.printStackTrace();
69    } finally {
70      if (conn != null) {
71        conn.close();
72      }
73    }
74    return false;
75  }
76
77  @Override
78  public void unlock() {
79    Jedis conn = null;
80    try {
81      conn = JEDIS_POOL.getResource();
82      if (lockValue.equals(conn.get(lockKey))) {
83        conn.del(lockKey);
84      }
85    } catch (JedisException e) {
86      e.printStackTrace();
87    } finally {
88      if (conn != null) {
89        conn.close();
90      }
91    }
92  }
93
94  @Override
95  public Condition newCondition() {
96    return null;
97  }
98}

上面的代码中也有一个问题，setnx 和 expire 是分为两步进行了，虽然在 catch 中处理异常并尝试将可能出现锁删除，但这种方式并不友好，一个好的方案是通过执行 lua 脚本来实现。在 Spring Redis Lock 和 Redission 都是通过 lua 脚本实现的

1local lockClientId = redis.call('GET', KEYS[1])
2if lockClientId == ARGV[1] then
3    redis.call('PEXPIRE', KEYS[1], ARGV[2])
4    return true
5elseif not lockClientId then
6    redis.call('SET', KEYS[1], ARGV[1], 'PX', ARGV[2])
7    return true
8end
9return false

1.2.2 Spring Redis Lock 实现

1. 引入库

在 Spring Boot 项目会根据 Spring Boot 依赖管理自动配置版本号

Maven

 1<dependency>
 2    <groupId>org.springframework.boot</groupId>
 3    <artifactId>spring-boot-starter-integration</artifactId>
 4</dependency>
 5
 6<dependency>
 7    <groupId>org.springframework.integration</groupId>
 8    <artifactId>spring-integration-redis</artifactId>
 9</dependency>
10
11<dependency>
12    <groupId>org.springframework.boot</groupId>
13    <artifactId>spring-boot-starter-data-redis</artifactId>
14</dependency>

2. 配置 redis

在 application-xxx.yml 中配置

1spring:
2  redis:
3    host: 127.0.0.1
4    port: 6379
5    timeout: 2500
6    password: xxxxx

3. 增加配置

RedisLockConfig.java

 1import java.util.concurrent.TimeUnit;
 2import org.springframework.context.annotation.Bean;
 3import org.springframework.context.annotation.Configuration;
 4import org.springframework.data.redis.connection.RedisConnectionFactory;
 5import org.springframework.integration.redis.util.RedisLockRegistry;
 6
 7@Configuration
 8public class RedisLockConfig {
 9
10  @Bean
11  public RedisLockRegistry redisLockRegistry(RedisConnectionFactory redisConnectionFactory) {
12    return new RedisLockRegistry(redisConnectionFactory, "redis-lock",
13        TimeUnit.MINUTES.toMillis(10));
14  }
15}

4. 使用

 1@Autowired
 2private RedisLockRegistry lockRegistry;
 3
 4Lock lock = lockRegistry.obtain(key);
 5boolean locked = false;
 6try {
 7  locked = lock.tryLock();
 8  if (!locked) {
 9    // 没有获取到锁的逻辑    
10  }
11
12  // 获取锁的逻辑
13} finally {
14  // 一定要解锁
15  if (locked) {
16    lock.unlock();
17  }
18}

1.2.3 Redission 实现

 1Config config = new Config();
 2config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("xxxxxx").setDatabase(0);
 3RedissonClient redissonClient = Redisson.create(config);
 4RLock rLock = redissonClient.getLock("lockKey");
 5boolean locked = false;
 6try {
 7  /* 
 8   * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败
 9   * leaseTime   锁的持有时间,超过这个时间锁会自动失效
10   */
11  locked = rLock.tryLock((long) waitTimeout, (long) leaseTime, TimeUnit.SECONDS);
12  if (!locked) {
13    // 没有获取锁的逻辑
14
15
16  }
17
18  // 获取锁的逻辑
19} catch (Exception e) {
20  throw new RuntimeException("aquire lock fail");
21} finally {
22  if(locked)
23    rLock.unlock();
24}

1.3 优缺点

优点：redis 本身的性能比较高，即使存在大量的 setnx 命令也不会有所下降

缺点：

如果 key 设置的超时时间过短可能导致业务流程还没处理完锁就释放了，导致其他请求也能获取到锁
如果 key 设置的超时时间过大，且未释放锁，会导致一些请求长时间在等待锁
在锁不断尝试的过程中，会浪费 CPU 资源

针对第 2 个缺点，在 Redission 通过续约机制，每隔一段时间去检测锁是否还在进行，如果还在运行就将对应的 key 增加一定的时间，保证在锁运行的情况下不会发生 key 到了过期时间自动删除的情况

2. 基于 Zookeeper 的实现

2.1 实现原理

基于zookeeper临时有序节点可以实现的分布式锁。

大致步骤：客户端对某个方法加锁时，在 zookeeper 上的与该方法对应的指定节点的目录下，生成一个唯一的临时有序节点。判断是否获取锁的方式很简单，只需要判断有序节点中序号最小的一个。当释放锁的时候，只需将这个瞬时节点删除即可。同时，其可以避免服务宕机导致的锁无法释放，而产生的死锁问题。

当第一个节点申请锁 xxxlock 时如下: 在 xxxlock 持久节点下，创建一个 lock 的临时有序节点，此时因为 lock 为有序节点中序号最小的一个，则此时获取到锁

当第一个节点还在处理业务逻辑未释放锁时，第二节点申请 xxxlock 锁，创建一个 lock 的临时有序节点，此时因为 lock 不是有序节点中序号最小的一个，则此时不能获取到锁，需要一直等到 lock:1 节点删除后才能获取到锁，此时 lock:2 会 watch 它的上一个节点（即 lock:1）等到 lock:1 删除后在获取锁

当第一个节点还在处理业务逻辑未释放锁时，第二节点还在排队，第三个节点申请锁时，创建一个 lock 的临时有序节点，此时因为 lock 不是有序节点中序号最小的一个，则此时不能获取到锁，需要一直等到上面的节点（ lock:1 和 lock:2 ）节点删除后才能获取到锁，此时 lock:3 会 watch 它的上一个节点（即 lock:2）等到 lock:2 删除后在获取锁

2.2 使用

2.2.1 使用 spring-integration-zookeeper 实现

Maven.

1<dependency>
2    <!-- spring integration -->
3    <groupId>org.springframework.boot</groupId>
4    <artifactId>spring-boot-starter-integration</artifactId>
5</dependency>
6<dependency>
7    <groupId>org.springframework.integration</groupId>
8    <artifactId>spring-integration-zookeeper</artifactId>
9</dependency>

Gradle.

1compile "org.springframework.integration:spring-integration-zookeeper:5.1.2.RELEASE"

增加配置

 1@Configuration
 2public class ZookeeperLockConfig {
 3
 4  @Value("${zookeeper.host}")
 5  private String zkUrl;
 6
 7  @Bean
 8  public CuratorFrameworkFactoryBean curatorFrameworkFactoryBean() {
 9    return new CuratorFrameworkFactoryBean(zkUrl);
10  }
11
12  @Bean
13  public ZookeeperLockRegistry zookeeperLockRegistry(CuratorFramework curatorFramework) {
14    return new ZookeeperLockRegistry(curatorFramework, "/lock");
15  }
16}

使用

 1@Autowired
 2private ZookeeperLockRegistry lockRegistry;
 3
 4Lock lock = lockRegistry.obtain(key);
 5boolean locked = false;
 6try {
 7  locked = lock.tryLock();
 8  if (!locked) {
 9    // 没有获取到锁的逻辑    
10  }
11
12  // 获取锁的逻辑
13} finally {
14  // 一定要解锁
15  if (locked) {
16    lock.unlock();
17  }
18}

2.2.2 使用 Apache Curator

Maven

1<dependency>
2    <groupId>org.apache.curator</groupId>
3    <artifactId>curator-framework</artifactId>
4    <version>5.1.0</version>
5</dependency>

使用

 1CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(
 2                connectString,
 3                sessionTimeoutMs,
 4                connectionTimeoutMs,
 5                new RetryNTimes(retryCount, elapsedTimeMs));
 6
 7InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "lock name");
 8
 9mutex.acquire(); // 获取锁
10mutex.acquire(long time, TimeUnit unit) // 获取锁并设置最大等待时间
11mutex.release(); // 释放锁

2.3 优缺点

优点：

解决了单点问题，通过集群部署 zookeeper；
因为用的临时节点，在项目出现意外的情况下可以保证锁可以释放，当 session 异常断开时，临时节点会自动删除；
不用在设置存储过期时间，避免了 Redis 锁过期引发的问题；

缺点：

性能不如 Redis 实现；

3. 基于数据库的实现

3.1 实现原理

1create table distributed_lock (
2  id int(11) unsigned NOT NULL auto_increment primary key,
3  key_name varchar(30) unique NOT NULL comment '锁名',
4  update_time datetime default current_timestamp on update current_timestamp comment '更新时间'
5)ENGINE=InnoDB comment '数据库锁';

方式一：通过 insert 和 delete 实现

使用数据库唯一索引，当我们想获取一个锁的时候，就 insert 一条数据，如果 insert 成功则获取到锁，获取锁之后，通过 delete 语句来删除锁

这种方式实现，锁不会等待，如果想设置获取锁的最大时间，需要自己实现

方式二：通过for update 实现

以下操作需要在事务中进行

1select * from distributed_lock where key_name = 'lock' for update;

在查询语句后面增加 for update，数据库会在查询过程中给数据库表增加排他锁。当某条记录被加上排他锁之后，其他线程无法再在该行记录上增加排他锁。for update 的另一个特性就是会阻塞，这样也间接实现了一个阻塞队列，但是 for update 的阻塞时间是由数据库决定的，而不是程序决定的。

在 MySQL 8 中，for update 语句可以加上 nowait 来实现非阻塞用法

1select * from distributed_lock where key_name = 'lock' for update nowait;

在 InnoDB 引擎在加锁的时候，只有通过索引查询时才会使用行级锁，否则为表锁，而且如果查询不到数据的时候也会升级为表锁。这种方式需要在数据库中实现已经存在数据的情况下使用。

3.2 优缺点

优点：

如果项目中已经使用了数据库在不引入其他中间件的情况下，可以直接使用数据库，减少依赖直接借助数据库，容易理解。

缺点：

操作数据库需要一定的开销，性能问题需要考虑；
使用数据库的行级锁并不一定靠谱，尤其是当我们的锁表并不大的时候；
没有锁超时机制，导致必须自己删除，故障后如何删除锁成为一个问题
for update 方式必须在事务内部，如果业务操作不能在事务里面执行又是一个问题
各种各样的问题，在解决问题的过程中会使整个方案变得越来越复杂。

4. 对比

从性能角度（从高到低）缓存 > Zookeeper >= 数据库

从可靠性角度（从高到低）Zookeeper > 缓存 > 数据库

问题、实现	Redis	Zookeeper	数据库
性能	高	中	低
可靠性	中	高	低
过期删除	有，设置过期时间，或者手动删除	执行业务逻辑后手动删除	1. for update 事务完成后，数据库自动释放 2. insert 方式执行业务逻辑后手动删除
阻塞队列	无，需要客户端自旋解决	通过监听上一个 lock 解决，watch 机制	1. for update 数据库自己解决 2. insert 方式需要客户端自旋解决
超时时间内业务未完成问题	需要自己写续约机制完成，Redission 内部自己实现了	无这问题	1. for update 执行时间过长，可能导致事务本身超时 2. insert 方式无此问题
项目异常导致锁未手动删除的情况	redis 有过期时间，过期时间后自动删除	session 断开后，临时节点自动删除	1. for update 机制数据库会自动清除 2. insert 方式就得自己想解决方案了