1. java锁的回顾：

• 我们在学习java线程的时候都知道，当有多个线程访问共享资源的时候有可能会产生线程安全的问题，那解决的办法大多数就是使用锁了。

• java中加锁的方式有很多，例如使用synchronized关键字，或者ReentrantLock可重入锁等。当然在使用锁的时候我们一定要考虑的问题就是: (1) 死锁问题 (2) 锁的性能问题。

• 对于死锁问题，我们都是在使用锁的时候尽量避免锁的嵌套或者能保证锁可以得到释放等;

• 而对于提升锁的性能问题，我们往往会选择锁粒度更小的加锁方式 (比如Lock使用起来可以控制锁的范围粒度，而synchronized就没那么灵活虽然idk1.6之后对synchronized也做了升级)缩短持有锁的时间、读写操作分离等手段。

2. 分布式锁:

• 我们之前介绍的使用synchronized或者ReentrantLock加锁可以保证多个线程访问共享资源的安全性，但是那有一个前提:就是我们多个线程访问的都在同一个服务器节点，这时候我们使用synchronized或者ReentrantLock可以保证这个服务器上JVM中数据的安全性; 但是如果我们我们有多个服务器节点 (分布式)功能处理业务呢? 这时候使用的锁只能保证一台服务器上线程的安全，并不能保证分布式的环境下的线程的安全。

• 以常见的电商减库存为典例，在分布式的情况下如果不做处理，很容易出现超买超卖的情况。原因就是，当一个交易线程下单成功之后我们需要减库存，但是如果修改库存的时候另外一个服务器处理的线程也要下订单减库存此时它读取到的还是没有减的库存认为还有商品卖所以就导致了超卖了。

• 那怎么解决分布式情况下多个线程访问不安全的问题呢? 那就是使用分布式锁了!当然分布式锁的实现方式还是有几种，例如:

使用ZooKeeper，基于临时有序节点。

使用Redis，基于set命令

• Redis 分布式锁本质上就是“占位置”，当一个线程操作redis缓存的时候Redis就为其占位 (使用setnx命令: 表示当且仅当key不存在的时候创建。若给定key已经存在，则setnx不做任何动作)，并且只允许被一个客户端占位置;当别的进程也要来占时，发现已经有了，就只好放弃或者稍后再试。当占位的线程访问完之后redis会通过 del删除占位。

• 当然这中间会出现一些问题，比如当线程占位成功了也就是获取锁成功，但是在释放锁前出现了问题那么就会导致死锁，所以使用redis加锁最好是给一个过期时间也就是使用expire。又因为expire和setnx两条指令的执行不是原子性的，所以也可能出现在setnx之后expire设置成功之前出现故障导致添加有效期失败，所以redis在2.8后对set命令进行了扩展，可以把setnx 和 expire一起执行，这样就解决了redis分布式锁的一些乱象。命令格式如下:

• 当然分布式锁不能解决锁超时问题，我们还需要保证锁的可重入性。如果一个锁支持同一个线程的多次加锁，那么这个锁就是可重入的。比如 Java 里有个 ReentrantLock 就是可重入锁。Redis 分布式锁如果要支持可重入，需要对客户端的 set 方法进行包装，使用线程的Threadlocal 变量存储当前持有锁的计数。

3. 分布式锁实现一（减库存案例）

@Autowiredprivate RedisTemplate redisTemplate;private static final String LOCK="goods_luck";@PostMapping("/buy01")public String buy01(Long gid){//获取redis分布式锁String uuid = UUID.randomUUID().toString().replace("-","");try{Boolean flag = redisTemplate.opsForValue().setIfAbsent(LOCK, uuid, 3L, TimeUnit.SECONDS);if (!flag){return "获取锁失败，商品减库存失败！";}log.info(uuid+" 获取锁成功！");//获取库存String nums = (String) redisTemplate.opsForValue().get("buy-" + gid);long totalNums = nums == null? 0: Integer.valueOf(nums);// 减库存if (totalNums >0){long nowTotal = totalNums -1;redisTemplate.opsForValue().set("buy-" + gid,String.valueOf(nowTotal));return "商品购买成功，当前库存量为："+nowTotal;}}catch (Exception e){}finally {//释放锁redisTemplate.delete(LOCK);}return "减库存失败！";}

缺陷: 如果因意外情况如服务器卡顿或者业务处理太慢，导致处理时间超过了key的有效期，也就是说key在处理业务卡顿的时候就已经释放了锁被其他进程占用了锁，那么当之前key的业务处理完之后有释放锁的行为就会释放其他key的锁，这样

就很严重。

分布式锁实现二(value值比较放置误删，也就是说释放锁失败或者释放了其他的锁)：

@PostMapping("/buy02")public String buy02(Long gid){//获取redis分布式锁String uuid = UUID.randomUUID().toString().replace("-","");try{Boolean flag = redisTemplate.opsForValue().setIfAbsent(LOCK, uuid, 3L, TimeUnit.SECONDS);if (!flag){return "获取锁失败，商品减库存失败！";}log.info(uuid+" 获取锁成功！");//获取库存String nums = (String) redisTemplate.opsForValue().get("buy-" + gid);long totalNums = nums == null? 0: Integer.valueOf(nums);// 减库存if (totalNums >0){long nowTotal = totalNums -1;redisTemplate.opsForValue().set("buy-" + gid,String.valueOf(nowTotal));return "商品购买成功，当前库存量为："+nowTotal;}}catch (Exception e){}finally {/*释放锁优化1：释放锁一定确认释放的是自己的锁缺点：因为我们还是分了两步首先先从redis中获取值，然后再删除锁。这两步redis的操作并不是原子性的。可能会出现在刚获取值比较成功之后，还没有删除lock此时本身的key有效期到了，它就会自动释放锁了，跟之前说的情况一样。*/if (redisTemplate.opsForValue().get(LOCK).equals(uuid)){redisTemplate.delete(LOCK);}}return "减库存失败！";}

• 该方式的实现跟方式一几乎差不多，只不过是在finally中释放锁的时候进行了一次锁value的校验，这样释放锁能够确保释放的是自己的锁而不是其他key的锁。

• 缺陷: 在finally中我们先从redis中获取lock的值进行比较，然后再进行delete删除锁，但是这两个操作在redis中并不是原子性操作，在高并发下也可能会出现比较值成功了要马上删除锁，但是锁的时效时间却到了其他的key可以获取锁了，这时候就出现跟方式二类似的问题。

分布式锁实现三（lua脚本解决原子性删除问题）：

先准备两个lua文件放到resource目录下，lua脚本的内容分别如下:

@PostMapping("/buy03")public String buy02(Long gid){//获取redis分布式锁String uuid = UUID.randomUUID().toString().replace("-","");try{//泛型为 锁资源的类型  当前商品数量为Long型DefaultRedisScript redisScript = new DefaultRedisScript();redisScript.setResultType(Long.class);//指定lua脚本目录redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("redis_lock.lua")));Object result = redisTemplate.execute(redisScript, Arrays.asList(LOCK), Collections.singletonList(uuid), 3);if ("0".equals(result)){return "抢夺分布式锁失败！，商品减库存失败！";}//            Boolean flag = redisTemplate.opsForValue().setIfAbsent(LOCK, uuid, 3L, TimeUnit.SECONDS);
//            if (!flag){
//                return "获取锁失败，商品减库存失败！";
//            }log.info(uuid+" 获取锁成功！");//获取库存String nums = (String) redisTemplate.opsForValue().get("buy-" + gid);long totalNums = nums == null? 0: Integer.valueOf(nums);// 减库存if (totalNums >0){long nowTotal = totalNums -1;redisTemplate.opsForValue().set("buy-" + gid,String.valueOf(nowTotal));return "商品购买成功，当前库存量为："+nowTotal;}}catch (Exception e){}finally {/*释放锁优化2：Redis作者推荐使用Lua脚本进行加锁和释放锁。（1）在Lua写入一些redis指令进行原子性操作，中间可以使用判断等逻辑处理（2）使用RedisTemplate可以 运行加载并脚本Lua不足之处：Lua脚本的分布式锁的加锁和释放锁能够满足单机redis的需要，也满足很多中小型公司需要。Lua脚本方式不能满足Redis集群模式的使用。RedLock解决方式可以解决集群模式的分布式锁的使用，在集成框架Redssion中有实现。*/DefaultRedisScript redisScript = new DefaultRedisScript();redisScript.setResultType(Long.class);//指定lua脚本目录redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("redis_unlock.lua")));Object result = redisTemplate.execute(redisScript, Arrays.asList(LOCK), Collections.singletonList(uuid), 3);if ("0".equals(result)){log.info("释放锁失败！");}else{log.info("释放锁成功！");}}return "减库存失败！";}

缺陷: 无法适用到redis集群的复杂环境中。redis作者就推荐使用一种RedLock的锁解决redis集群分布式锁的问题，这里不再详细介绍RedLock了有兴趣可以去了解。目前RedLock在Redisson中有集成实现。