分布式 | 分布式锁的实现

爱被打了一巴掌 2022-11-06 05:54 134阅读 0赞

# 分布式锁的实现 #

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

1.  Redis 实现
2.  Zookeeper 实现
3.  数据库实现

--------------------

### 文章目录 ###

*  分布式锁的实现
 *  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 Curator
     *  2.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 实现原理 ##

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

## 1.2 实现方式 ##

### 1.2.1 原生代码 ###

public class DistributedLock implements Lock { 
    
      private static JedisPool JEDIS_POOL = null;
      private static int EXPIRE_SECONDS = 60;
    
      public static void setJedisPool(JedisPool jedisPool, int expireSecond) { 
        JEDIS_POOL = jedisPool;
        EXPIRE_SECONDS = expireSecond;
      }
    
      private String lockKey;
      private String lockValue;
    
      private DistributedLock(String lockKey) { 
        this.lockKey = lockKey;
      }
    
      public static DistributedLock newLock(String lockKey) { 
        return new DistributedLock(lockKey);
      }
    
      @Override
      public void lock() { 
        if (!tryLock()) { 
          throw new IllegalStateException("未获取到锁");
        }
      }
    
      @Override
      public void lockInterruptibly() throws InterruptedException { 
      }
    
      @Override
      public boolean tryLock() { 
        return tryLock(0, null);
      }
    
      @Override
      public boolean tryLock(long time, TimeUnit unit) { 
        Jedis conn = null;
        String retIdentifier = null;
        try { 
          conn = JEDIS_POOL.getResource();
          lockKey = UUID.randomUUID().toString();
    
          // 获取锁的超时时间，超过这个时间则放弃获取锁
          long end = 0;
          if (time != 0) { 
            end = System.currentTimeMillis() + unit.toMillis(time);
          }
    
          do { 
            if (conn.setnx(lockKey, lockValue) == 1) { 
              conn.expire(lockKey, EXPIRE_SECONDS);
              return true;
            }
    
            try { 
              Thread.sleep(10);
            } catch (InterruptedException e) { 
              Thread.currentThread().interrupt();
            }
          } while (System.currentTimeMillis() < end);
        } catch (JedisException e) { 
          if (lockValue.equals(conn.get(lockKey))) { 
            conn.del(lockKey);
          }
          e.printStackTrace();
        } finally { 
          if (conn != null) { 
            conn.close();
          }
        }
        return false;
      }
    
      @Override
      public void unlock() { 
        Jedis conn = null;
        try { 
          conn = JEDIS_POOL.getResource();
          if (lockValue.equals(conn.get(lockKey))) { 
            conn.del(lockKey);
          }
        } catch (JedisException e) { 
          e.printStackTrace();
        } finally { 
          if (conn != null) { 
            conn.close();
          }
        }
      }
    
      @Override
      public Condition newCondition() { 
        return null;
      }
    }

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

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

### 1.2.2 Spring Redis Lock 实现 ###

**1. 引入库**

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

**Maven**

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-redis</artifactId>
    </dependency>
    
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-redis</artifactId>
    </dependency>

**2. 配置 redis**

在 `application-xxx.yml` 中配置

spring:
      redis:
        host: 127.0.0.1
        port: 6379
        timeout: 2500
        password: xxxxx

**3. 增加配置**

**RedisLockConfig.java**

import java.util.concurrent.TimeUnit;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.integration.redis.util.RedisLockRegistry;
    
    @Configuration
    public class RedisLockConfig { 
    
      @Bean
      public RedisLockRegistry redisLockRegistry(RedisConnectionFactory redisConnectionFactory) { 
        return new RedisLockRegistry(redisConnectionFactory, "redis-lock",
            TimeUnit.MINUTES.toMillis(10));
      }
    }

**4. 使用**

@Autowired
    private RedisLockRegistry lockRegistry;
    
    Lock lock = lockRegistry.obtain(key);
    boolean locked = false;
    try { 
      locked = lock.tryLock();
      if (!locked) { 
        // 没有获取到锁的逻辑 
      }
    
      // 获取锁的逻辑
    } finally { 
      // 一定要解锁
      if (locked) { 
        lock.unlock();
      }
    }

### 1.2.3 Redission 实现 ###

Config config = new Config();
    config.useSingleServer().setAddress("redis://127.0.0.1:6379").setPassword("xxxxxx").setDatabase(0);
    RedissonClient redissonClient = Redisson.create(config);
    RLock rLock = redissonClient.getLock("lockKey");
    boolean locked = false;
    try { 
      /* * waitTimeout 尝试获取锁的最大等待时间，超过这个值，则认为获取锁失败 * leaseTime 锁的持有时间,超过这个时间锁会自动失效 */
      locked = rLock.tryLock((long) waitTimeout, (long) leaseTime, TimeUnit.SECONDS);
      if (!locked) { 
        // 没有获取锁的逻辑
        
        
      }
      
      // 获取锁的逻辑
    } catch (Exception e) { 
      throw new RuntimeException("aquire lock fail");
    } finally { 
      if(locked)
        rLock.unlock();
    }

## 1.3 优缺点 ##

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

缺点：

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

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

# 2. 基于 Zookeeper 的实现 #

## 2.1 实现原理 ##

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

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

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

![zklock.png][]

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

![zklock2.png][]

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

![zklock3.png][]

## 2.2 使用 ##

### 2.2.1 使用 spring-integration-zookeeper 实现 ###

**Maven.**

<dependency>
        
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-integration</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.integration</groupId>
        <artifactId>spring-integration-zookeeper</artifactId>
    </dependency>

**Gradle.**

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

**增加配置**

@Configuration
    public class ZookeeperLockConfig { 
    
      @Value("${zookeeper.host}")
      private String zkUrl;
    
      @Bean
      public CuratorFrameworkFactoryBean curatorFrameworkFactoryBean() { 
        return new CuratorFrameworkFactoryBean(zkUrl);
      }
    
      @Bean
      public ZookeeperLockRegistry zookeeperLockRegistry(CuratorFramework curatorFramework) { 
        return new ZookeeperLockRegistry(curatorFramework, "/lock");
      }
    }

**使用**

@Autowired
    private ZookeeperLockRegistry lockRegistry;
    
    Lock lock = lockRegistry.obtain(key);
    boolean locked = false;
    try { 
      locked = lock.tryLock();
      if (!locked) { 
        // 没有获取到锁的逻辑 
      }
    
      // 获取锁的逻辑
    } finally { 
      // 一定要解锁
      if (locked) { 
        lock.unlock();
      }
    }

### 2.2.2 使用 Apache Curator ###

**Maven**

<dependency>
        <groupId>org.apache.curator</groupId>
        <artifactId>curator-framework</artifactId>
        <version>5.1.0</version>
    </dependency>

**使用**

CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(
                    connectString,
                    sessionTimeoutMs,
                    connectionTimeoutMs,
                    new RetryNTimes(retryCount, elapsedTimeMs));
    
    InterProcessMutex mutex = new InterProcessMutex(curatorFramework, "lock name");
    
    mutex.acquire(); // 获取锁
    mutex.acquire(long time, TimeUnit unit) // 获取锁并设置最大等待时间
    mutex.release(); // 释放锁

## 2.3 优缺点 ##

**优点：**

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

**缺点：**

1.  性能不如 Redis 实现；

# 3. 基于数据库的实现 #

## 3.1 实现原理 ##

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

**方式一：通过 insert 和 delete 实现**

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

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

**方式二：通过for update 实现**

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

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

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

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

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

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

## 3.2 优缺点 ##

**优点：**

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

**缺点：**

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

# 4. 对比 #

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

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

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

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5bmRldg_size_16_color_FFFFFF_t_70_pic_center]

[zklock.png]: /images/20221023/107bb76c396e4715a7496a752fa81f9a.png
[zklock2.png]: /images/20221023/448d56ab404e4f7b831553b5daecc701.png
[zklock3.png]: /images/20221023/acd159a2872741a4bf936a74becc1e4c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3p5bmRldg_size_16_color_FFFFFF_t_70_pic_center]: /images/20221023/fb0f7135a8854dfaa246ec5bc99f9bd6.png