Error message here!

Hide Error message here!

忘记密码?

Error message here!

请输入正确邮箱

Hide Error message here!

密码丢失?请输入您的电子邮件地址。您将收到一个重设密码链接。

Error message here!

返回登录

Close

分布式锁

mrguozp 2019-01-31 14:33:00 阅读数:157 评论数:0 点赞数:0 收藏数:0

单机

  • 方案比较多,synchronized和juc很丰富

分布式锁

  • 互斥性:在任意时刻,只有一个客户端能持有锁
  • 不会发生死锁:即有一个客户端在持有锁的期间崩溃而没有主动解锁,也能保证后续其他客户端能加锁

文章来源:https://www.cnblogs.com/guozp/p/10341337.html

常见方案
  1. 基于数据库
  2. 基于分布式缓存(redis、tair等)
  3. 基于zk
    要基于你的业务场景选择合适方案

数据库(mysql)

基于数据库的ACID以及MVCC(多版本并发控制机),MVCC是通过保存数据在某个时间点的快照来实现的,不同存储引擎的MVCC实现是不同的,典型的有乐观并发控制和悲观并发控制

  • 基于悲观锁(for update)

    select * from table where *** for update

  • 基于乐观锁(version)

    乐观锁是基于数据的版本号实现的,表增加一个字段version,每次读取的时候,将version取出,更新的时候,比较version是否一致,一致,处理完后把version加1;不一致,本次未拿到锁

    • 表定义(根据需求增加)

      id resource status expire version
      1 1 2 2019-01-01 12:00:00 1
      2 2 2 2019-01-01 12:00:01 1
    • 含义
      • resource:代表资源
      • status:锁定状态
      • expire:过期时间,根据需求看是否需要增加使用
    • 执行流程:
      1. 执行查询操作获取当前数据的数据版本号,例如:select id, resource, state,version from table where state=1 and id=1;
      2. 执行更新:update table set state=2, version=上次+1 where resource=1 and state=1 and version=1
      3. 上述执行影响1行,加锁成功,影响0行,自己加锁失败,其它人已经加锁锁定

tair

Tair没有直接提供分布式锁的api,但是可以借助提供的其他api实现分布式锁。

  • incr/decr(不可重入锁)
    • 原理:通过计数api的上下限值约束来实现(增加/减少计数。可设置最大值和最小值)
    • api:
      1. 增加计数(加锁):
        Result<Integer> incr(int namespace, Serializable key, int value, int defaultValue, int expireTime, int lowBound, int upperBound)
      2. 减少计数(释放锁):
      Result<Integer> decr(int namespace, Serializable key, int value, int defaultValue, int expireTime, int lowBound, int upperBound) 
      1. 关键参数解释:
      defaultValue: 第一次调用incr时的key的count初始值,第一次返回的值为defaultValue + value, decr第一次返回defaultValue - value lowBound: 最小值 upperBound: 最大值
  • 使用
    1. 线程一调用incr加锁,加锁后,key的值变成1,而key的上限值为1,其他线程再调用该接口时会报错COUNTER_OUT_OF_RANGE
    2. 待线程一使用完成后,调用decr解锁,此时key已经有值1,返回 1-1=0,解锁成功。多次调用会失败,因为范围是0~1。
    3. 通过0、1的来回变化,达到分布式锁的目的,当key为1时获取到锁,为0时释放锁
  • Get/Put
    • 原理:使用put的version校验实现
    • api
      1. put
      ResultCode put(int namespace, Serializable key, Serializable value, int version, int expireTime)`
      

      一定要设置过期参数expireTime,否则锁执行过程中进程crash,锁不会释放,会长期占有,影响业务,加上后,业务至少可以自行恢复

      1. 关键参数解释:

         version - 为了解决并发更新同一个数据而设置的参数。当version为0时,表示强制更新
        这里注意:
        此处version,除了0、1外的任何数字都可以,传入0,tair会强制覆盖;而传入1,第一个client写入会成功,但是新写入时服务端的version以0开始计数啊,所以此时version也是1,所以下一个到来的client写入也会成功,这样造成了冲突。
  • 实现

这里针对网络等问题做了重试,同时改造支持可重入锁,不可重入锁,目前这里可重入没有做计数以及重新设置过期时间,使用的各位可以根据实际情况进行改造

 @Override
public boolean tryLock(String lockKey, int expireTime, boolean reentrant) {
if (expireTime <= 0) {
expireTime = DEFAULT_EXPIRE_TIME;
}
int retryGet = 0;
try {
Result<DataEntry> result = tairManager.get(NAMESPACE, lockKey);
while (retryGet++ < LOCK_GET_MAX_RETRY && result != null && isError(result.getRc())) {
result = tairManager.get(NAMESPACE, lockKey);
}
if (result == null) {
log.error("tryLock error, maybe Tair service is unavailable");
return false;
}
if (ResultCode.DATANOTEXSITS.equals(result.getRc())) {
// version 2表示为空,若不是为空,则返回version error
ResultCode code = tairManager.put(NAMESPACE, lockKey, getLockValue(), DEFAULT_VERSION, expireTime);
if (ResultCode.SUCCESS.equals(code)) {
return true;
} else if (retryPut.get() < LOCK_PUT_MAX_RETRY && isError(code)) {
retryPut.set(retryPut.get() + 1);
return tryLock(lockKey, expireTime);
}
} else if (reentrant && result.getValue() != null && getLockValue().equals(result.getValue().getValue())) {
return true;
}
} catch (Exception e) {
log.error("try lock is error, msg is {}", e);
} finally {
retryPut.remove();
}
return false;
}
@Override
public void unlock(String lockKey) {
unlock(lockKey, false);
}
@Override
public boolean unlock(String lockKey, boolean reentrant) {
if (!reentrant) {
ResultCode invalid = tairManager.invalid(NAMESPACE, lockKey);
return invalid != null && invalid.isSuccess();
}
Result<DataEntry> result = tairManager.get(NAMESPACE, lockKey);
if (result != null && result.isSuccess() && result.getValue() != null) {
String value = result.getValue().getValue().toString();
if (getLockValue().equals(value)) {
ResultCode rc = tairManager.invalid(NAMESPACE, lockKey);
if (rc != null && rc.isSuccess()) {
return true;
} else {
log.error("unlock failed, tairLockManager.invalidValue fail, key is {}, ResultCode is {}",
lockKey, rc);
return false;
}
} else {
log.warn("unlock failed,value is not equal lockValue, key is {}, lockValue is {}, value is {}",
lockKey, getLockValue(), value);
return false;
}
}
return false;
}
@Override
public boolean lockStatus(String lockKey) {
Result<DataEntry> result = tairManager.get(NAMESPACE, lockKey);
if (result != null && result.isSuccess() && result.getValue() != null) {
return true;
}
return false;
}
private boolean isError(ResultCode code) {
return code == null || ResultCode.CONNERROR.equals(code) || ResultCode.TIMEOUT.equals(code) || ResultCode.UNKNOW
.equals(code);
}
private String getLockValue() {
return NetUtils.getLocalIp() + "_" + Thread.currentThread().getName();
}

redis

  • 正确的加锁逻辑
    • API:
      1. 加锁
        SET key value [EX seconds] [PX milliseconds] [NX|XX]
      2. 释放锁
        EVAL script numkeys key [key ...] arg [arg ...]
    • 关键参数解释

      加锁

      ```
      EX second :设置键的过期时间为 second 秒。 SET key value EX second 效果等同于 SETEX key second value
      PX millisecond :设置键的过期时间为 millisecond 毫秒。SET key value PX millisecond 效果等同于 PSETEX key millisecond value
      NX :只在键不存在时,才对键进行设置操作。 SET key value NX 效果等同于 SETNX key value
      XX :只在键已经存在时,才对键进行设置操作。
      
      >释放
      

      script 参数是一段 Lua 5.1 脚本程序,它会被运行在 Redis 服务器上下文中,这段脚本不必(也不应该)定义为一个 Lua 函数。
      numkeys 参数用于指定键名参数的个数。
      键名参数 key [key ...] 从 EVAL 的第三个参数开始算起,表示在脚本中所用到的那些 Redis 键(key),这些键名参数可以在 Lua 中通过全局变量 KEYS 数组,用 1 为基址的形式访问( KEYS[1] , KEYS[2] ,以此类推)。
      在命令的最后,那些不是键名参数的附加参数 arg [arg ...] ,可以在 Lua 中通过全局变量 ARGV 数组访问,访问的形式和 KEYS 变量类似( ARGV[1] 、 ARGV[2] ,诸如此类)。
      ```

    • 实现

      /**
      *1. 当前没有锁(key不存在),那么就进行加锁操作,并对锁设置个有效期,同时value表示加锁的客户端。2. 已有锁存在,不做任何操作
      **/
      public boolean tryLock(String lockKey, String requestId, int expireTime) {
      String result = jedis.set(lockKey, requestId, SET_IF_NOT_EXIST, SET_WITH_EXPIRE_TIME, expireTime);
      if (LOCK_SUCCESS.equals(result)) {
      return true;
      }
      return false;
      }
      public boolean unlock(String lockKey, String requestId) {
      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));
      if (RELEASE_SUCCESS.equals(result)) {
      return true;
      }
      return false;
      }
      
      • 首先,set()加入了NX参数,可以保证如果key已存在,则函数不会调用成功,即只有一个客户端能持有锁。其次,由于我们对锁设置了过期时间,即使锁的持有者后续发生crash而没有解锁,锁也会因为到了过期时间而自动解锁(即key被删除),不会发生死锁。最后,因为我们将value赋值为requestId,代表加锁的客户端请求标识,那么在客户端在解锁的时候就可以进行校验是否是同一个客户端。
      • 释放锁,这段Lua代码的功能:首先获取锁对应的value值,检查是否与requestId相等,如果相等则删除锁(解锁)。那么为什么要使用Lua语言来实现呢?因为lua可以确保上述操作是原子性的
  • tair的rdb引擎目前不支持上述命令,所以需要写成两行命令(或许新版本支持了,因为我使用的的还是旧版本,所以rdb的实现方式:

    支持可重入锁,不可重入锁,目前这里可重入没有做计数以及重新设置过期时间,使用的各位可以根据实际情况进行改造

    /**
    * rdb 不支持多参数,所以使用两个命令
    *
    * @param lockKey
    * @param expireTime 超时时间
    * @param reentrant 是否可重入,重入后会延长时间
    * @return
    */
    @Override
    public boolean tryLock(String lockKey, int expireTime, boolean reentrant) {
    if (expireTime <= 0) {
    expireTime = DEFAULT_EXPIRE_TIME;
    }
    boolean result = redisRepo.setNx(lockKey, getLockValue(), expireTime);
    if (!reentrant) {
    return result;
    }
    String value = redisRepo.get(lockKey);
    if (getLockValue().equals(value)) {
    result = redisRepo.setNx(lockKey, getLockValue(), expireTime);
    }
    return result;
    }
    /**
    * 版本不支持lua,所以使用两个命令
    *
    * @param lockKey
    * @param reentrant 是否可以释放其它人创建的锁
    * @return
    */
    @Override
    public boolean unlock(String lockKey, boolean reentrant) {
    if (!reentrant) {
    return redisRepo.delKeys(lockKey) > 0;
    }
    long result = 0;
    String value = redisRepo.get(lockKey);
    if (getLockValue().equals(value)) {
    result = redisRepo.delKeys(lockKey);
    }
    return result > 0;
    }
    @Override
    public boolean lockStatus(String lockKey) {
    String value = redisRepo.get(lockKey);
    return StringUtils.isNotBlank(value);
    }
    private String getLockValue() {
    return NetUtils.getLocalIp() + "_" + Thread.currentThread().getName();
    }
    
  • 错误的加锁示例
    1. setnx()方法作用就是SET IF NOT EXIST,expire()方法就是给锁加一个过期时间。乍一看好像和前面的set()方法结果一样,但是由于这是两条Redis命令,不具有原子性,如果程序在执行完setnx()之后crash,由于锁没有设置过期时间,将会发生死锁

       public static void wrongGetLock1(Jedis jedis, String lockKey, String requestId, int expireTime) {
      Long result = jedis.setnx(lockKey, requestId);
      if (result == 1) {
      jedis.expire(lockKey, expireTime);
      }
      }
      
      1. 通过setnx()方法尝试加锁,如果当前锁不存在,返回加锁成功。
      2. 如果锁存在则获取锁过期时间,和当前时间比较,如果锁已经过期,则设置新的过期时间,返回加锁成功

      ```
      public static boolean wrongGetLock2(Jedis jedis, String lockKey, int expireTime) {

      long expires = System.currentTimeMillis() + expireTime;
      String expiresStr = String.valueOf(expires);

      if (jedis.setnx(lockKey, expiresStr) == 1) {
      return true;
      }

      String currentValueStr = jedis.get(lockKey);
      if (currentValueStr != null && Long.parseLong(currentValueStr) < System.currentTimeMillis()) {
      String oldValueStr = jedis.getSet(lockKey, expiresStr);
      if (oldValueStr != null && oldValueStr.equals(currentValueStr)) {
      return true;
      }
      }
      return false;
      }

    ```

    上述代码问题出在哪里?
    * 由于是客户端自己生成过期时间,所以强制要求每个客户端的时间必须同步
    * 当锁过期的时候,如果多个客户端同时执行jedis.getSet()方法,那么虽然最终只有一个客户端可以加锁,但是这个客户端的锁的过期时间可能被其他客户端覆盖。
    * 锁不具备拥有者标识,即任何客户端都可以解锁(看个人业务)
  • 错误的锁释放示例
    1. 使用jedis.del()方法删除锁,这种不先判断锁的拥有者而直接解锁的方式,会导致任何客户端都可以随时进行解锁
    ```
    public static void wrongReleaseLock1(Jedis jedis, String lockKey) {
    jedis.del(lockKey);
    }
    ```
    2. 以下代码分成两条命令去执行,如果调用jedis.del()的时候,锁已经不属于当前客户端的时,会解除他人加的锁
    ```
    public static void wrongReleaseLock2(Jedis jedis, String lockKey, String requestId) {
    // 判断加锁与解锁是不是同一个客户端
    if (requestId.equals(jedis.get(lockKey))) {
    // 若在此时,这把锁过期不属于这个客户端的,则会误解锁
    jedis.del(lockKey);
    }
    }
    ``` 
redis官方锁

Redis的官方曾提出了一个容错的分布式锁算法:RedLock,只要有超过一半的缓存服务器能够正常工作,系统就可以保证分布式锁的可用性。详情参考

zk

有机会或者留言需要的在写吧, 略略略
文章来源:https://www.cnblogs.com/guozp/p/10341337.html

方案比较(从低到高)

  • 从理解的难易程度角度:数据库 > 缓存 > Zookeeper

  • 从实现的复杂性角度:Zookeeper >= 缓存 > 数据库

  • 从性能角度:缓存 > Zookeeper >= 数据库

  • 从可靠性角度:Zookeeper > 缓存 > 数据库

版权声明
本文为[mrguozp]所创,转载请带上原文链接,感谢
https://www.cnblogs.com/guozp/p/10341337.html

编程之旅,人生之路,不止于编程,还有诗和远方。
阅代码原理,看框架知识,学企业实践;
赏诗词,读日记,踏人生之路,观世界之行;

支付宝红包,每日可领