浙江龙网

  1. 首页 > 生活百科 > >

Redis高并发分布式锁详解( 四 )

生活百科

(newPromise));return;}// 3:创建一个监听器，别的线程进行redis-pub命令之后进行调用RedisPubSubListener listener = createListener(channelName, value);// 4:底层交给netty调用redis-sub命令subscribeService.subscribe(LongCodec.INSTANCE, channelName, semaphore, listener);}};semaphore.acquire(listener);listenerHolder.set(listener);return newPromise;}3）RedissonLock类#tryAcquireAsync方法（核心点主体）
//RedissonLock类#tryAcquireAsync方法private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {if (leaseTime != -1) {return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);}//尝试加锁逻辑 RFuture<Long> ttlRemainingFuture=tryLockInnerAsync(commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);//添加监听器ttlRemainingFuture.addListener(new FutureListener<Long>() {@Override//Future任务执行完会回调该方法public void operationComplete(Future<Long> future) throws Exception {if (!future.isSuccess()) {return;}Long ttlRemaining = future.getNow();// 加锁成功if (ttlRemaining == null) {//看门狗续命scheduleExpirationRenewal(threadId);}}});return ttlRemainingFuture;}4）RedissonLock类#tryLockInnerAsync方法（核心点，加锁逻辑）
//RedissonLock类#tryLockInnerAsync方法//利用redis的单线程执行任务，redis会将整个脚本作为一个整体执行 ， 且中间不会被其他命令插入//采用的是hash的类型来存储锁 ， 为了实现重入锁的概念//Redis pttl命令以毫秒为单位返回 key 的剩余过期时间<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {internalLockLeaseTime = unit.toMillis(leaseTime);return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('hset', 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]);",//对应为KEYS[1]（对应传入的锁的命名），ARGV[1]（设置的超时时间，默认30s），ARGV[2] -》（uuid + ":" + threadId）Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));}5）RedissonLock类#scheduleExpirationRenewal方法（核心点，看门狗的逻辑【续命】）
//RedissonLock类#scheduleExpirationRenewal方法//采用Future+事件监听的方式，方法嵌套调用来实现定时任务private void scheduleExpirationRenewal(final long threadId) {if (expirationRenewalMap.containsKey(getEntryName())) {return;}Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {@Overridepublic void run(Timeout timeout) throws Exception {RFuture<Boolean> future = commandExecutor.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.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));//再次添加监听器，重复检查future.addListener(new FutureListener<Boolean>() {@Overridepublic void operationComplete(Future<Boolean> future) throws Exception {expirationRenewalMap.remove(getEntryName());if (!future.isSuccess()) {log.error("Can't update lock " + getName() + " expiration", future.cause());return;}if (future.getNow()) {// reschedule itself//递归调用scheduleExpirationRenewal(threadId);}}});}}, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);//如果该任务已经存在一个了，就把新建的任务关闭，Map中的key为（uuid + ":" + threadId）if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {task.cancel();}}6）Redisson类#unlock方法
//RedissonLock类#unlock方法public void unlock() {Boolean opStatus = get(unlockInnerAsync(Thread.currentThread().getId()));if (opStatus == null) {throw new IllegalMonitorStateException(...);}if (opStatus) {//移除看门狗的定时任务cancelExpirationRenewal();}}//RedissonLock类#unlockInnerAsync方法protected RFuture<Boolean> unlockInnerAsync(long threadId) {return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,//如果不存在锁"if (redis.call('exists', KEYS[1]) == 0) then " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; " +"end;" +//当前线程并没有持有锁，则返回nil"if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +"return nil;" +"end; " +//前线程持有锁，则对value-1，拿到-1之后的vlaue"local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +//value>0，以毫秒为单位返回剩下的过期时间 。（保证可重入）"if (counter > 0) then " +"redis.call('pexpire', KEYS[1], ARGV[2]); " +"return 0; " +//value<=0，则对key进行删除操作，return 1 (方法返回 true) 。然后进行redis-pub指令，用于唤醒其他正在休眠的线程 。"else " +"redis.call('del', KEYS[1]); " +"redis.call('publish', KEYS[2], ARGV[1]); " +"return 1; "+"end; " +"return nil;",//参数顺序KEYS[1]（锁的名称），KEYS[2]（发布订阅的Channel名：redisson_lock__channel+锁名），ARGV[1]（发布的消息），ARGV[2]（锁超时时间），ARGV[3]（uuid + ":" + threadId）Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));}

推荐阅读


上一篇：Jupyter notebook导入Pycharm项目的.py文件里的模块及方法

下一篇：Jupyter Notebook单元格加宽的方法3种