这节学习Java用Redis做分布式锁，来做秒杀场景卖货减库存的案例。
这里库存也存Redis里面，调减库存接口的时候判断一下大于0（还有库存）就拿出来减1。
这里StringRedisTemplate是Spring Boot对Redis的封装，27行和30行的写法就等同于注释里面的用Jedis的写法，就是去调Redis的GET和SET命令。
这样的代码中存在并发问题，在高并发的场景下，只要多个线程都执行读库存的操作，那么读出来的库存数目就是一样多的。比如三个线程都读出来200，然后都减少1，变成199，最后写回也是199，但是卖出了3件商品，这个就是典型的超卖问题。
如果用synchronized给这段代码加锁，实际上是保证了在一个实例上多个线程访问这段代码，每次只有获取到锁的那个能执行，这个是JVM进程级别的锁，在不同的实例上（分布式场景下）就没法用这个锁来保证安全了。
比如下面这种，访问Nginx代理服务器，然后LB到两个tomcat实例上的场景。JVM级别的锁只能保证实例内部的线程之间按这个锁来一个一个访问这段代码，不同实例之间的还是没法保证。
相应的Nginx配置，可以看到流量被转发到了8080端口和8090端口的两个实例上：
JMeter是Java开发常用的一个压力测试软件，可以给RD开发完成后自测，模拟高并发场景，暴露出可能存在的问题。
首先在测试计划（Test Plan）下添加一个线程组（Thread Group）：
然后在线程组里添加一个HTTP Request：
在Path里配置要压测的接口：
回到Thread Group，在这里配置来告诉JMeter以什么样的压力来跑下面配的请求。
下图的配置含义是，发200个请求，在0秒内（表示这200个请求一次性发过去，为了模拟高并发所以这样配置瞬间的压力），循环压4次（4个200）：
在压测之前，在HTTP Request下面添加一个聚合报告（Aggregate Report），这样就能得到测试的结果：
点击绿色运行按钮，选择聚合报告的保存位置，然后就开始压测了，完成后可以看到报告：
但是这次关注点不在这个报告，而是代码里打出来的日志，只要能看到两个实例中有打印出相同的库存数，就证明存在超卖的问题：
最后结果发现是存在超卖问题的，也证实了JVM锁不能解决分布式场景下需要加锁的问题。
使用Redis中的SETNX命令，仅当key不存在时设置，可以实现分布式锁：
和SET的区别就是，SET是在key存在时会覆盖，但是SETNX在key存在的时候会写入失败（也就可以用来表示获取锁失败）。
设置成功时（返回true）表示加锁成功，可以执行业务逻辑；设置失败（返回false）时加锁失败，这里的处理是返回一个错误码（那么前端拿到后可以显示“服务器繁忙，请稍后再试”）这样的字样。
还需要注意在业务代码执行完毕之后，要把锁释放掉，所以要在Redis里把这个key清除掉。
使用SETNX来实现分布式锁的思想，也借助了Redis是单线程的这件事，在执行的时候都会排队执行，所以只有一个能设置成功，也就实现了“分布式环境下只有一个线程设置锁成功”这件事。
注：这里在加锁失败的时候会直接返回错误码，另一种处理方式是套一个循环，在加锁失败的时候把这个线程sleep一段时间，然后再去加锁。具体采用哪种处理方式还是看具体业务场景和端上需求了。
上面的写法中，如果业务代码里抛出异常了，那么设置锁之后线程在释放锁之前就终止了，这样导致其他线程一直没法加锁，所以这里可以用try来把业务代码套起来，然后在finally的部分去执行释放锁的操作：
如果在业务代码执行的过程中机器宕机了， 那么后面的finally块还是无法执行，还是没法保证锁被正确释放，所以可以为加入的key设置一个超时时间，这样即使宕机了，只要超过这个时间还是会释放锁（下面的代码里是设置了10秒）：
在上面的代码里，如果27行设置key和28行设置超时之间发生了宕机，那么还是会出现key一直存在，锁没法释放的问题，这里可以改用一个原子操作来解决：
在Redis里会作为一个原子操作来指定，同时成功或者同时失败，显然即使同时失败了也没有影响。
目前的代码里还是有小概率发生超卖问题，因为引入了超时时间，如果线程A中业务代码执行的时间超过了超时时间（比如有慢查询之类的），那么key会自动失效（这个时候线程A还没执行解锁，但是锁实际上已经被释放了）。这个时候大量的请求过来，假设线程B又获取到了锁开始了执行，这个时候线程A执行到了释放锁的位置，就把线程B的锁给释放掉了。
在这样的场景下，可能就会导致后续释放的都是别人的锁，这样就会引起代码执行的顺序完全不可控，只要并发量一直存在，极端情况下可能导致分布式锁的永久失效问题，导致这个问题被无限放大。
到目前位置，只是在用key来标识锁，现在要解决释放别人的锁的问题，只要用value来记录加锁的是谁就可以了（不要用原生的线程id，因为不同实例上线程id可能相同）。
下面的代码里用生成的UUID（发生重复的概率极低可以忽略不计，其实就可以认为一定不会发生重复）来模拟每个线程执行时候的客户端id，用来唯一标识加锁人的身份。
上面的代码里45到47行还是存在原子性的问题，比如线程A在45行判断是自己加的锁成功，紧接着（或者是发生了GC也可能导致两行代码执行之间停顿了一下）这个key的超时时间到了，这个锁被释放掉了，然后大量的并发请求过来，线程B加锁成功，这个时候A再去执行解锁，解的是线程B加的锁。也就是说这样写的话，“释放别人的锁”问题是依然存在的。
最好的方式是把判断和解锁做成一个原子操作，引起这个问题还是因为引入了超时时间，所以可以看一下这个超时时间应该怎么处理。
把超时时间延长，是一个治标不治本的办法，例如业务代码执行的是一个定时任务，那么就是会执行比较长的时间，是没法避免的（总是可能发生超时）。
一种处理方法是，在业务线程开始执行的同时开一个分线程，例如主线程超时时间是30秒，那么分线程可以设置定时任务，每过10秒钟就检查一下主线程有没有执行完退出（锁有没有被删除掉）：如果执行完了就让自己也结束；如果没有执行完，就把超时时间重置。
从头来实现上面的这些，把所有的问题都考虑清楚是很困难的，市面上已经存在一些框架把这些事情做好了，比如Redisson。
