上海做衣服版的网站,高端大气的医院网站,站长seo查询,虚拟空间能建多个网站源码仓库地址#xff1a;gitgitee.com:chuangchuang-liu/hm-dingping.git 1、redission介绍
目前基于redis的setnx特性实现的自定义分布式锁仍存在的问题#xff1a;
问题描述重入问题同一个线程无法多次获取统一把锁。当方法A成功获取锁后#xff0c;调用方法B#xff0… 源码仓库地址gitgitee.com:chuangchuang-liu/hm-dingping.git 1、redission介绍
目前基于redis的setnx特性实现的自定义分布式锁仍存在的问题
问题描述重入问题同一个线程无法多次获取统一把锁。当方法A成功获取锁后调用方法B方法B也要获取锁此时由于锁是不可重入的也就是被方法A占用着此时就产生了死锁的问题不可重试自定义分布式锁无失败重试机制超时释放锁的超时释放虽然可以避免死锁问题但确实也可能存在业务执行时间比较长的情况那这种情况下就仍存在安全隐患问题主从一致性如果Redis提供了主从集群当我们向集群写数据时主机需要异步的将数据同步给从机而万一在同步过去之前主机宕机了就会出现死锁问题。
什么是Redission Redission是一个用于Java的Redis客户端它提供了丰富的特性包括内存数据网格的功能。它支持同步/异步/RxJava/Reactive API拥有超过50种基于Redis的Java对象和服务。Redission的使用非常简单没有学习曲线您不需要了解任何Redis命令就可以开始使用。GitHub - redisson/redisson, Redisson官网 Redission可以让Java应用更方便地访问和操作Redis数据存储适合于需要高性能和高并发的应用场景。
2、快速开始
导入依赖
!--redission--
dependencygroupIdorg.redisson/groupIdartifactIdredisson/artifactIdversion3.13.6/version
/dependencyRedission配置客户端
Configuration
public class RedisConfig {Beanpublic RedissonClient redisClient(){Config config new Config();// 可以用rediss://来启用SSL连接config.useSingleServer().setAddress(redis://192.168.224.128:6379).setPassword(123456);return Redisson.create(config);}
}使用Redission分布式锁
Resource
private RedissionClient redissonClient;Test
void testRedisson() throws Exception{//获取锁(可重入)指定锁的名称RLock lock redissonClient.getLock(anyLock);//尝试获取锁参数分别是获取锁的最大等待时间(期间会重试)锁自动释放时间时间单位boolean isLock lock.tryLock(1,10,TimeUnit.SECONDS);//判断获取锁成功if(isLock){try{System.out.println(执行业务); }finally{//释放锁lock.unlock();}}
}3、redission可重入锁原理
3.1、原理介绍
在Lock锁中他是借助于底层的一个voaltile的一个state变量来记录重入的状态的比如当前没有人持有这把锁那么state0假如有人持有这把锁那么state1如果持有这把锁的人再次持有这把锁那么state就会1 如果是对于synchronized而言他在c语言代码中会有一个count原理和state类似也是重入一次就加一释放一次就-1 直到减少成0 时表示当前这把锁没有被人持有。 、
而在redission中也支持这种可重入锁原理是通过redis的hash数据结构实现的。其中key表示这把锁是否存在field判断锁是被哪个线程持有value则记录锁被持有次数。
3.2、源码剖析
获取锁
其中各参数解释 KEYS[1]锁的名称 ARGV[1]锁过期时间 ARGV[2]id “:” threadId
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;
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;
return redis.call(pttl, KEYS[1]);判断锁是否存在 如果不存在则设置当前线程标识计数器1设置过期时间 如果存在。做二次判断判断锁的持有线程是不是自己 如果是计数器1重置锁的过期时间 如果不是获取锁失败返回锁的剩余过期时间 释放锁
其中各参数解释 KEYS[1]锁的名称 KEYS[2]订阅频道 ARGV[1]是要发布的消息内容 ARGV[2]锁过期时间 ARGV[3]id “:” threadId
if (redis.call(hexists, KEYS[1], ARGV[3]) 0) then
return nil;
end;
local counter redis.call(hincrby, KEYS[1], ARGV[3], -1);
if (counter 0) then
redis.call(pexpire, KEYS[1], ARGV[2]);
return 0;
else
redis.call(del, KEYS[1]);
redis.call(publish, KEYS[2], ARGV[1]);
return 1;
end;
return nil;判断锁是不是当前线程 不是直接返回 是计数器– 二次判断判断计数器是否大于0 大于0重置锁过期时间 否则真正释放锁 4、redission锁重试和WatchDog机制
4.1、redission是如何解决不可重试的
源码剖析 用户调用tryLock方法时指定waitTime最大等待时间
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {long time unit.toMillis(waitTime);long current System.currentTimeMillis();long threadId Thread.currentThread().getId();Long ttl tryAcquire(waitTime, leaseTime, unit, threadId);// lock acquiredif (ttl null) {return true;}time - System.currentTimeMillis() - current;if (time 0) {acquireFailed(waitTime, unit, threadId);return false;}current System.currentTimeMillis();RFutureRedissonLockEntry subscribeFuture subscribe(threadId);if (!subscribeFuture.await(time, TimeUnit.MILLISECONDS)) {if (!subscribeFuture.cancel(false)) {subscribeFuture.onComplete((res, e) - {if (e null) {unsubscribe(subscribeFuture, threadId);}});}acquireFailed(waitTime, unit, threadId);return false;}try {time - System.currentTimeMillis() - current;if (time 0) {acquireFailed(waitTime, unit, threadId);return false;}while (true) {long currentTime System.currentTimeMillis();ttl tryAcquire(waitTime, leaseTime, unit, threadId);// lock acquiredif (ttl null) {return true;}time - System.currentTimeMillis() - currentTime;if (time 0) {acquireFailed(waitTime, unit, threadId);return false;}// waiting for messagecurrentTime System.currentTimeMillis();if (ttl 0 ttl time) {subscribeFuture.getNow().getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);} else {subscribeFuture.getNow().getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);}time - System.currentTimeMillis() - currentTime;if (time 0) {acquireFailed(waitTime, unit, threadId);return false;}}} finally {unsubscribe(subscribeFuture, threadId);}// return get(tryLockAsync(waitTime, leaseTime, unit));
}计算等待时间和获取当前时间将用户指定的等待时间转换为毫秒并记录方法调用时的当前时间。尝试获取锁。如果ttl为空则获取锁成功否则返回的是其他线程占用锁的剩余有效时间检查剩余等待时间。如果time小于等于0调用acquireFailed方法返回false订阅锁。通过subscribe方法订阅相关的锁。如果在剩余时间内未能订阅成功处理取消订阅并调用acquireFailed方法返回false。循环等待锁释放消息。等待过程中会调用tryAcquire方法获取锁如果获取成功返回true处理锁的ttl。如果ttl大于0返回锁被其他线程占用的剩余过期时间ttl。更新剩余等待时间time。以time和ttl中较小的值继续等待再次尝试。再次检查等于剩余等待时间。如果小于0调用acquireFailed方法返回false循环结束后要么获取锁成功要么超过最大等待时间了最终调用unsubscribe方法取消订阅 结论1redission不是获取锁失败后立即进行重试而是等待“一定时间”后再进行重试节省了一定的CPU资源对服务器性能有一定提升 结论2一定要采取调用tryLock方法携带参数waitTime的重载方法其他重载的tryLock方法底层是不具备重试机制的。 4.2、redission是如何解决锁超时释放的-看门狗机制
自定义分布式锁仍存在的一个问题是锁的超时释放虽然可以避免死锁问题但确实也可能存在业务执行时间比较长的情况那这种情况下业务还未执行完毕锁就被释放了存在一定的安全隐患。 源码剖析
private RFutureBoolean tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {if (leaseTime ! -1) {return tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);}RFutureBoolean ttlRemainingFuture tryLockInnerAsync(waitTime,commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(),TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);ttlRemainingFuture.onComplete((ttlRemaining, e) - {if (e ! null) {return;}// lock acquiredif (ttlRemaining) {// 开启任务更新过期时间scheduleExpirationRenewal(threadId);}});return ttlRemainingFuture;
}private void renewExpiration() {ExpirationEntry ee EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ee null) {return;}Timeout task commandExecutor.getConnectionManager().newTimeout(new TimerTask() {Overridepublic void run(Timeout timeout) throws Exception {ExpirationEntry ent EXPIRATION_RENEWAL_MAP.get(getEntryName());if (ent null) {return;}Long threadId ent.getFirstThreadId();if (threadId null) {return;}RFutureBoolean future renewExpirationAsync(threadId);future.onComplete((res, e) - {if (e ! null) {log.error(Cant update lock getName() expiration, e);return;}if (res) {// reschedule itselfrenewExpiration();}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);ee.setTimeout(task);
}protected RFutureBoolean renewExpirationAsync(long threadId) {return evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,if (redis.call(hexists, KEYS[1], ARGV[2]) 1) then redis.call(pexpire, KEYS[1], ARGV[1]); return 1; end; return 0;,Collections.singletonList(getName()),internalLockLeaseTime, getLockName(threadId));}如果没有指定leaseTime那么底层会默认传入commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout()看门狗时间在leasetime的1/3处时间会创建一个任务renewExpirationAsync方法来异步地更新重置锁过期时间递归地调用自身来更新锁过期时间直到业务处理完毕。 至此redission解决了因业务阻塞而导致锁提前释放的问题 业务执行完毕释放锁源码剖析
public RFutureVoid unlockAsync(long threadId) {RPromiseVoid result new RedissonPromiseVoid();// 释放锁RFutureBoolean future unlockInnerAsync(threadId);future.onComplete((opStatus, e) - {// 取消锁失效时间更新重置任务cancelExpirationRenewal(threadId);if (e ! null) {result.tryFailure(e);return;}if (opStatus null) {IllegalMonitorStateException cause new IllegalMonitorStateException(attempt to unlock lock, not locked by current thread by node id: id thread-id: threadId);result.tryFailure(cause);return;}result.trySuccess(null);});return result;
}void cancelExpirationRenewal(Long threadId) {ExpirationEntry task EXPIRATION_RENEWAL_MAP.get(getEntryName());if (task null) {return;}if (threadId ! null) {task.removeThreadId(threadId);}if (threadId null || task.hasNoThreads()) {Timeout timeout task.getTimeout();if (timeout ! null) {timeout.cancel();}// 删除递归更新锁时间任务EXPIRATION_RENEWAL_MAP.remove(getEntryName());}
}当业务执行完毕且锁正常释放后删除递归更新锁时间任务避免redission一直递归创建任务更新锁过期时间 5、redission锁的MultiLock原理
为了提高redis的可用性我们会搭建集群或者主从现在以主从为例 此时我们去写命令写在主机上 主机会将数据同步给从机但是假设在主机还没有来得及把数据写入到从机去的时候此时主机宕机哨兵会发现主机宕机并且选举一个slave变成master而此时新的master中实际上并没有锁信息此时锁信息就已经丢掉了。 为了解决该问题redission的方案是去掉redis集群主从关系每一个节点都是平等的。加锁逻辑是需要写入到每一个节点上才算加锁成功。这样当某一台机器宕机了这台机器的slave节点变为master节点此时另一个线程趁虚而入虽然可以正常写入但其它机器仍会写入失败最终结果仍是获取锁失败从而保证了获取锁的可靠性。 MulitLock源码剖析
public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {// try {// return tryLockAsync(waitTime, leaseTime, unit).get();// } catch (ExecutionException e) {// throw new IllegalStateException(e);// }long newLeaseTime -1;if (leaseTime ! -1) {if (waitTime -1) {newLeaseTime unit.toMillis(leaseTime);} else {newLeaseTime unit.toMillis(waitTime)*2;}}long time System.currentTimeMillis();long remainTime -1;if (waitTime ! -1) {remainTime unit.toMillis(waitTime);}long lockWaitTime calcLockWaitTime(remainTime);int failedLocksLimit failedLocksLimit();ListRLock acquiredLocks new ArrayList(locks.size());for (ListIteratorRLock iterator locks.listIterator(); iterator.hasNext();) {RLock lock iterator.next();boolean lockAcquired;try {if (waitTime -1 leaseTime -1) {lockAcquired lock.tryLock();} else {long awaitTime Math.min(lockWaitTime, remainTime);lockAcquired lock.tryLock(awaitTime, newLeaseTime, TimeUnit.MILLISECONDS);}} catch (RedisResponseTimeoutException e) {unlockInner(Arrays.asList(lock));lockAcquired false;} catch (Exception e) {lockAcquired false;}if (lockAcquired) {acquiredLocks.add(lock);} else {if (locks.size() - acquiredLocks.size() failedLocksLimit()) {break;}if (failedLocksLimit 0) {unlockInner(acquiredLocks);if (waitTime -1) {return false;}failedLocksLimit failedLocksLimit();acquiredLocks.clear();// reset iteratorwhile (iterator.hasPrevious()) {iterator.previous();}} else {failedLocksLimit--;}}if (remainTime ! -1) {remainTime - System.currentTimeMillis() - time;time System.currentTimeMillis();if (remainTime 0) {unlockInner(acquiredLocks);return false;}}}if (leaseTime ! -1) {ListRFutureBoolean futures new ArrayList(acquiredLocks.size());for (RLock rLock : acquiredLocks) {RFutureBoolean future ((RedissonLock) rLock).expireAsync(unit.toMillis(leaseTime), TimeUnit.MILLISECONDS);futures.add(future);}for (RFutureBoolean rFuture : futures) {rFuture.syncUninterruptibly();}}return true;}遍历锁集合调用lock.tryLock尝试获取锁。将获取结果传给变量lockAcquired如果获取成功将当前锁存放到acquiredLocks集合中获取成功后如果此时的剩余等待时间小于等于0释放自己已获取的锁返回false如果获取失败判断是否具备重试机制 没有重试则直接返回false有重试机制将acquiredLocks集合清空将iterator指针前移重新遍历尝试。 6、结论
目前已接触的分布式锁有
可不重入锁/自定义分布式锁
原理 利用setnx特性、expire避免死锁、添加线程标识避免锁误删 缺点 仍存在不可重入、失败不可重试、锁超时失效等问题
可重入锁
原理 利用hash数据结构存储线程标识和重入次数、利用看门狗机制延续锁失效时间、利用信号量机制控制等待重试时间 缺点 仍存在集群模式下redis宕机导致的锁失效问题
MulitLock
原理 利用多个平等的redis节点所有redis都写入才算获取锁成功 缺点 维护成本高实现相对复杂
