专栏首页Java架构师进阶分布式锁实现大型连续剧之(一):Redis

分布式锁实现大型连续剧之(一):Redis

前言:

单机环境下我们可以通过JAVA的Synchronized和Lock来实现进程内部的锁,但是随着分布式应用和集群环境的出现,系统资源的竞争从单进程多线程的竞争变成了多进程的竞争,这时候就需要分布式锁来保证。

实现分布式锁现在主流的方式大致有以下三种

基于数据库的索引和行锁

基于Redis的单线程原子操作:setNX

基于Zookeeper的临时有序节点

这篇文章我们用Redis来实现,会基于现有的各种锁实现来分析,最后分享Redission的锁源码分析来看下分布式锁的开源实现

设计实现

加锁

一、 通过setNx和getSet来实现

这是现在网上大部分版本的实现方式,笔者之前项目里面用到分布式锁也是通过这样的方式实现

public boolean lock(Jedis jedis, String lockName, Integer expire) {

//返回是否设置成功//setNx加锁long now = System.currentTimeMillis(); boolean result = jedis.setnx(lockName,String.valueOf(now + expire *1000)) ==1;if(!result) {//防止死锁的容错Stringtimestamp = jedis.get(lockName);if(timestamp !=null&& Long.parseLong(timestamp) < now) {//不通过del方法来删除锁。而是通过同步的getSetStringoldValue = jedis.getSet(lockName,String.valueOf(now + expire));if(oldValue !=null&& oldValue.equals(timestamp)) {            result =true;            jedis.expire(lockName, expire);        }    } }if(result) {    jedis.expire(lockName, expire); }returnresult;

}

代码分析:

通过setNx命令老保证操作的原子性,获取到锁,并且把过期时间设置到value里面

通过expire方法设置过期时间,如果设置过期时间失败的话,再通过value的时间戳来和当前时间戳比较,防止出现死锁

通过getSet命令在发现锁过期未被释放的情况下,避免删除了在这个过程中有可能被其余的线程获取到了锁

存在问题

防止死锁的解决方案是通过系统当前时间决定的,不过线上服务器系统时间一般来说都是一致的,这个不算是严重的问题

锁过期的时候可能会有多个线程执行getSet命令,在竞争的情况下,会修改value的时间戳,理论上来说会有误差

锁无法具备客户端标识,在解锁的时候可能被其余的客户端删除同一个key

虽然有小问题,不过大体上来说这种分布式锁的实现方案基本上是符合要求的,能够做到锁的互斥和避免死锁

二、 通过Redis高版本的原子命令

jedis的set命令可以自带复杂参数,通过这些参数可以实现原子的分布式锁命令

jedis.set(lockName, "", "NX", "PX", expireTime);

复制代码代码分析

redis的set命令可以携带复杂参数,第一个是锁的key,第二个是value,可以存放获取锁的客户端ID,通过这个校验是否当前客户端获取到了锁,第三个参数取值NX/XX,第四个参数 EX|PX,第五个就是时间

NX:如果不存在就设置这个key XX:如果存在就设置这个key

EX:单位为秒,PX:单位为毫秒

这个命令实质上就是把我们之前的setNx和expire命令合并成一个原子操作命令,不需要我们考虑set失败或者expire失败的情况

解锁

一、 通过Redis的del命令

public boolean unlock(Jedis jedis, String lockName) {

jedis.del(lockName);

return true;

}

代码分析

通过redis的del命令可以直接删除锁,可能会出现误删其他线程已经存在的锁的情况

二、 Redis的del检查

public static void unlock2(Jedis jedis, String lockKey, String requestId) {

// 判断加锁与解锁是不是同一个客户端

if (requestId.equals(jedis.get(lockKey))) {

// 若在此时,这把锁突然不是这个客户端的,则会误解锁

jedis.del(lockKey);

}

}

代码分析

新增了requestId客户端ID的判断,但由于不是原子操作,在多个进程下面的并发竞争情况下,无法保证安全

三、 Redis的LUA脚本

public static boolean unlock3(Jedis jedis, String lockKey, String requestId) {

Stringscript ="if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";Objectresult = jedis.eval(script, Collections.singletonList(lockKey), Collections.singletonList(""));if(1L == (long) result) {returntrue;  }returnfalse;

}

代码分析

通过Lua脚本来保证操作的原子性,其实就是把之前的先判断再删除合并成一个原子性的脚本命令,逻辑就是,先通过get判断value是不是相等,若相等就删除,否则就直接return

Redission的分布式锁

Redission是redis官网推荐的一个redis客户端,除了基于redis的基础的CURD命令以外,重要的是就是Redission提供了方便好用的分布式锁API

一、 基本用法

RedissonClient redissonClient = RedissonTool.getInstance();

RLock distribute_lock = redissonClient.getLock("distribute_lock");try{booleanresult = distribute_lock.tryLock(3,10, TimeUnit.SECONDS);    }catch(InterruptedException e) {        e.printStackTrace();    }finally{if(distribute_lock.isLocked()) {            distribute_lock.unlock();        }    }

代码流程

通过redissonClient获取RLock实例

tryLock获取尝试获取锁,第一个是等待时间,第二个是锁的超时时间,第三个是时间单位

执行完业务逻辑后,最终释放锁

二、 具体实现

我们通过tryLock来分析redission分布式的实现,lock方法跟tryLock差不多,只不过没有最长等待时间的设置,会自旋循环等待锁的释放,直到获取锁为止

long time = unit.toMillis(waitTime);

long current = System.currentTimeMillis();

//获取当前线程ID,用于实现可重入锁

final long threadId = Thread.currentThread().getId();

//尝试获取锁

Long ttl = tryAcquire(leaseTime, unit, threadId);

// lock acquired

if (ttl == null) {

return true;

}

time -= (System.currentTimeMillis() - current);if(time <=0) {//等待时间结束,返回获取失败acquireFailed(threadId);returnfalse;    }    current = System.currentTimeMillis();//订阅锁的队列,等待锁被其余线程释放后通知finalRFuture subscribeFuture = subscribe(threadId);if(!await(subscribeFuture, time, TimeUnit.MILLISECONDS)) {if(!subscribeFuture.cancel(false)) {            subscribeFuture.addListener(newFutureListener() {@OverridepublicvoidoperationComplete(Future future)throwsException{if(subscribeFuture.isSuccess()) {                        unsubscribe(subscribeFuture, threadId);                    }                }            });        }        acquireFailed(threadId);returnfalse;    }try{        time -= (System.currentTimeMillis() - current);if(time <=0) {            acquireFailed(threadId);returnfalse;        }while(true) {longcurrentTime = System.currentTimeMillis();            ttl = tryAcquire(leaseTime, unit, threadId);// lock acquiredif(ttl ==null) {returntrue;            }            time -= (System.currentTimeMillis() - currentTime);if(time <=0) {                acquireFailed(threadId);returnfalse;            }// waiting for message,等待订阅的队列消息currentTime = System.currentTimeMillis();if(ttl >=0&& ttl < time) {                getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);            }else{                getEntry(threadId).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);            }            time -= (System.currentTimeMillis() - currentTime);if(time <=0) {                acquireFailed(threadId);returnfalse;            }        }    }finally{        unsubscribe(subscribeFuture, threadId);    }

代码分析

首先tryAcquire尝试获取锁,若返回ttl为null,说明获取到锁了

判断等待时间是否过期,如果过期,直接返回获取锁失败

通过Redis的Channel订阅监听队列,subscribe内部通过信号量semaphore,再通过await方法阻塞,内部其实是用CountDownLatch来实现阻塞,获取subscribe异步执行的结果,来保证订阅成功,再判断是否到了等待时间

再次尝试申请锁和等待时间的判断,循环阻塞在这里等待锁释放的消息RedissonLockEntry也维护了一个semaphore的信号量

无论是否释放锁,最终都要取消订阅这个队列消息

redission内部的getEntryName是客户端实例ID+锁名称来保证多个实例下的锁可重入

tryAcquire获取锁

redisssion获取锁的核心代码,内部其实是异步调用,但是用get方法阻塞了

private Long tryAcquire(long leaseTime, TimeUnit unit, long threadId) {

return get(tryAcquireAsync(leaseTime, unit, threadId));

}

private RFuture tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) {

if (leaseTime != -1) {

return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG);

}

RFuture ttlRemainingFuture = tryLockInnerAsync(LOCK_EXPIRATION_INTERVAL_SECONDS, TimeUnit.SECONDS, threadId, RedisCommands.EVAL_LONG);

ttlRemainingFuture.addListener(new FutureListener() {br/>@Override

public void operationComplete(Future future) throws Exception {

if (!future.isSuccess()) {

return;

}

Long ttlRemaining = future.getNow();// lock acquiredif(ttlRemaining ==null) {                scheduleExpirationRenewal(threadId);            }        }    });returnttlRemainingFuture;}

tryLockInnerAsync方法内部是基于Lua脚本来获取锁的

先判断KEYS[1](锁名称)对应的key是否存在,不存在获取到锁,hset设置key的value,pexpire设置过期时间,返回null表示获取到锁

存在的话,锁被占,hexists判断是否是当前线程的锁,若是的话,hincrby增加重入次数,重新设置过期时间,不是当前线程的锁,返回当前锁的过期时间

RFuture tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand command) {

internalLockLeaseTime = unit.toMillis(leaseTime);

returncommandExecutor.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]);",                Collections.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));}

Redission避免死锁的解决方案:

Redission为了避免锁未被释放,采用了一个特殊的解决方案,若未设置过期时间的话,redission默认的过期时间是30s,同时未避免锁在业务未处理完成之前被提前释放,Redisson在获取到锁且默认过期时间的时候,会在当前客户端内部启动一个定时任务,每隔internalLockLeaseTime/3的时间去刷新key的过期时间,这样既避免了锁提前释放,同时如果客户端宕机的话,这个锁最多存活30s的时间就会自动释放(刷新过期时间的定时任务进程也宕机)

// lock acquired,获取到锁的时候设置定期更新时间的任务if(ttlRemaining) {                scheduleExpirationRenewal(threadId);            }//expirationRenewalMap的并发安全MAP记录设置过的缓存,避免并发情况下重复设置任务,internalLockLeaseTime / 3的时间后重新设置过期时间privatevoidscheduleExpirationRenewal(finallongthreadId){if(expirationRenewalMap.containsKey(getEntryName())) {return;    }    Timeout task = commandExecutor.getConnectionManager().newTimeout(newTimerTask() {@Overridepublicvoidrun(Timeout timeout)throwsException{            RFuture 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.singletonList(getName()), internalLockLeaseTime, getLockName(threadId));            future.addListener(newFutureListener() {@OverridepublicvoidoperationComplete(Future future)throwsException{                    expirationRenewalMap.remove(getEntryName());if(!future.isSuccess()) {                        log.error("Can't update lock "+ getName() +" expiration", future.cause());return;                    }if(future.getNow()) {// reschedule itselfscheduleExpirationRenewal(threadId);                    }                }            });        }    }, internalLockLeaseTime /3, TimeUnit.MILLISECONDS);if(expirationRenewalMap.putIfAbsent(getEntryName(), task) !=null) {        task.cancel();    }}

unlock解锁

protected RFuture 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;" +

"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;",

Arrays.asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId));

}

Redission的unlock解锁也是基于Lua脚本实现的,内部逻辑是先判断锁是否存在,不存在说明已经被释放了,发布锁释放消息后返回,锁存在再判断当前线程是否锁拥有者,不是的话,无权释放返回,解锁的话,会减去重入的次数,重新更新过期时间,若重入数捡完,删除当前key,发布锁释放消息

总结:

主要基于Redis来设计和实现分布式锁,通过常用的设计思路引申到Redission的实现,无论是设计思路还是代码健壮性Redission的设计都是优秀的,值得学习,下一步会讲解关于Zookeeper的分布式锁实现和相关开源源码分析。

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • MyBatis框架及原理分析

    <!DOCTYPE configuration PUBLIC "-//mybatis.org//DTD Config 3.0//EN" "http://myba...

    java架构师
  • 10多年程序员总结的20多条经验教训

    1.从小事做起,然后再扩展 无论是创建一个新的系统,还是添加功能到现有的系统中,我总是从一个简单到几乎没有任何所需功能的版本启动,然后再一步一步地解决问题,直到...

    java架构师
  • 如何解决代码中if…else 过多的问题

    if...else 是所有高级编程语言都有的必备功能。但现实中的代码往往存在着过多的 if...else。虽然 if...else 是必须的,但滥用 if......

    java架构师
  • 复杂sql

    IT云清
  • 设置第一个字母字体变大并且所有字母大小写 及下划线

    大当家
  • C++ OpenCV播放视频及调用摄像头显示

    前一篇我们介绍了 《C++ OpenCV摄像头及视频操作类VideoCapture介绍》,我们现在就针对这个类里的API进行DEMO的演示。

    Vaccae
  • 《剑指offer》之扑克牌顺子

    LL今天心情特别好,因为他去买了一副扑克牌,发现里面居然有2个大王,2个小王(一副牌原本是54张^_^)...他随机从中抽出了5张牌,想测测自己的手气,看看能不...

    程序员爱酸奶
  • Python--切片学习记录

                 2.第一个索引的元素包含在切片内,第二个索引的元素不包含在切片内

    明天依旧可好
  • Day45:扑克牌顺子

    思路一:   这个题可以先把整数数组排序,然后由于左端可能是大小王,所以右端开始往左判断。首尾各一个指针,当尾部指针的值比它左面的值恰好大1的情况下。不需要大...

    stefan666
  • 使用Python,怎么生成一个整数列表?

    TalkPython

扫码关注云+社区

领取腾讯云代金券