经验总结

  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();}}

经验总结扩展阅读


上一篇：抖音团购套餐佣金抽成是多少

下一篇：沈阳师范大学播音主持专业怎么样