深入理解redis分布式锁大家好，我是北君。在这篇文档中，我们将介绍Redis是如何实现分布式锁的演进过程，以及为什么不能直接使用Setnx来实现分布式锁。一、分布式锁简介分布式锁是在分布式系统中控制不同进程共同访问共享资源的锁的一种实现。如果不同的系统或者同一系统的不同主机共享某个关键资源，往往需要互斥来防止相互干扰，保证一致性。业界流行的分布式锁实现方式一般有三种：基于数据库实现的分布式锁基于Redis实现的分布式锁基于Zookeeper实现的分布式锁这里主要介绍如何通过Redis实现分布式锁。在介绍Redis分布式锁之前，我们先介绍实现Redis分布式锁的关键命令。2.setnxsetnxkeyvalueSetnx(SETifNoteXists)命令在指定的键不存在时为该键设置指定的值。设置成功，返回1。设置失败，返回0。PS：Redis官方不推荐基于setnx命令实现分布式锁，因为会出现很多问题，①，单点问题。例如：clientA从master那里拿到锁lock01。当master正要同步lock01（redis主从同步一般是异步的）到slave时，突然宕机，导致lock01无法同步到slave主从切换，slave节点被提升为master节点，客户端B仍然可以从master获取lock01。这将不可避免地导致同一把锁被多人使用。②.对于锁的高级用法，比如读写锁、可重入锁等，setnx实现起来比较困难。这里先介绍基于sentnx的分布式锁，后面介绍官方推荐的基于redisson的分布式锁。3.redis-distributedlock-stage1收到以上后，查询三级分类数据。如果我们部署多个商品服务，然后多个线程同时获取三级分类数据，如果不加分布式锁，那么每个部署的商品服务的第一个查询都会去DB。publicMap>getCatelogJsonWithRedisLock()throwsInterruptedException{//1.获取分布式锁Booleanlock=stringRedisTemplate.opsForValue().setIfAbsent("lock","111");if(lock){//true表示加锁成功，执行相关业务Map>dataFromDb=getDataFromDb();stringRedisTemplate.delete("锁定");返回dataFromDb；}else{System.out.println("获取分布式锁失败...等待重试...");//锁失败...重试机制//休眠100毫秒try{TimeUnit.MILLISECONDS.sleep(100);}catch(InterruptedExceptione){e.printStackTrace();}//自旋方法返回getCatelogJsonWithRedisLock();}}4.redis-分布式锁-phase2设置锁自动过期publicMap>getCatelogJsonWithRedisLock()throwsInterruptedException{//1.获取分布式锁booleanlock=stringRedisTemplate.opsForValue().setIfAbsent(“锁”，“111”）；if(lock){//true表示加锁成功，执行相关业务//设置过期时间stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);Map>dataFromDb=getDataFromDb();stringRedisTemplate.delete("锁定");返回dataFromDb；}else{System.out.println("获取分布式锁失败...等待重试...");//加锁失败...重试机制//休眠100毫秒try{TimeUnit.MILLISECONDS.sleep(100);}catch(InterruptedExceptione){e.printStackTrace();}//自旋模式returngetCatelogJsonWithRedisLock();}}5、redis-distributedlock-stage3setnx命令和过期时间保证原子性publicMap>getCatelogJsonWithRedisLock()throwsInterruptedException{//1.获取分布式锁Booleanlock=stringRedisTemplate.opsForValue().setIfAbsent("lock","111",30,TimeUnit.SECONDS);if(lock){//true表示加锁成功，执行相关业务//设置过期时间//stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);Map>dataFromDb=getDataFromDb();stringRedisTemplate.delete("锁定");返回dataFromDb；}else{System.out.println("获取分布式锁失败...等待重试...");//锁定失败...重试机制//休眠一百毫秒try{TimeUnit.MILLISECONDS.sleep(100);}catch(InterruptedExceptione){e.printStackTrace();}//自旋方法返回getCatelogJsonWithRedisLock();}}6、Redis——分布式锁——第4阶段保证被删除的是自己的锁。publicMap>getCatelogJsonWithRedisLock()throwsInterruptedException{//1.获取分布式锁Stringuuid=UUID.randomUUID().toString();Booleanlock=stringRedisTemplate.opsForValue().setIfAbsent("lock",uuid,30,TimeUnit.SECONDS);if(lock){//true表示加锁成功，执行相关业务//设置过期时间//stringRedisTemplate.expire("lock",30,TimeUnit.SECONDS);Map>dataFromDb=getDataFromDb();StringlockValue=stringRedisTemplate.opsForValue().get("lock");if(uuid.equals(lockValue)){stringRedisTemplate.delete("锁");}返回dataFromDb；}else{System.out.println("获取分布式锁失败...等待重试...");//加锁失败...重试机制//休眠100毫秒try{TimeUnit.MILLISECONDS.sleep(100);}catch(InterruptedExceptione){e.printStackTrace();}//自旋方法返回getCatelogJsonWithRedisLock();}}7.Redis-distributedlock-stage5使用lua脚本保证删除锁和判断锁这两个操作的原子性publicMap>getCatelogJsonWithRedisLock(){//1.获取分布式锁Stringuuid=UUID.randomUUID().toString();布尔锁=stringRedisTemplate.opsForValue().setIfAbsent("lock",uuid,30,TimeUnit.SECONDS);if(lock){System.out.println("获取分布式锁成功...");Map>dataFromDb=null;try{//锁定成功...执行业务dataFromDb=getDataFromDb();}finally{Stringscript="ifredis.call('get',KEYS[1])==ARGV[1]thenreturnredis.call('del',KEYS[1])elsereturn0end";//删除锁stringRedisTemplate.execute(newDefaultRedisScript(script,Long.class),Arrays.asList("lock"),uuid);}//先去redis查询，确保当前锁是自己的//获取值比较，比较成功deletion=atomicluascriptunlock//StringlockValue=stringRedisTemplate.opsForValue().get("lock");//如果(uuid.equals(lockValue)){////删除我自己的锁//stringRedisTemplate.delete("lock");//}返回dataFromDb；}else{System.out.println("获取分布式锁失败...等待重分配尝试...");//锁定失败...重试机制//休眠100毫秒try{TimeUnit.MILLISECONDS.sleep(100);}catch(InterruptedExceptione){e.printStackTrace();}//Spin方法返回getCatelogJsonWithRedisLock();}}这也是分布式锁的final模式，需要保证两点：锁【设置锁+过期时间】和删除锁【判断+删除】原子性
