Redis实现高并发分布式锁

　　分布式锁场景

　　在分布式环境下多个操作需要以原子的方式执行

　　首先启一个springboot项目，再引入redis依赖包：

　　org.springframework.boot

　　spring-boot-starter-data-redis

　　2.2.2.RELEASE

　　以下是一个扣减库存的接口作为例子：

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　return "end";

　　}

　　}

　　1.单实例应用场景

　　以上代码使用JMeter压测工具进行调用，设置参数为：

　　Number Of Threads[users]:100

　　Ramp Up Period[in seconds]:0

　　Loop Count：2

　　用单个web调用，结果出现并发问题：

　　解决方案：加入同步锁(synchronized)

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　synchronized(this) {

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　return "end";

　　}

　　}

　　}

　　2.多实例分布式场景

　　以上代码，比如有多个应用程序，用nginx做负载均衡，进行同时调用压测

　　两个程序存在同样的扣减，出现并发现象。

　　第一个应用扣减结果显示：

　　第二个应用扣减结果显示：

　　解决方案：redis的setnx方法(可参考SETNX的api)

　　多个线程setnx调用时，有且仅有一个线程会拿到这把锁，所以拿到锁的执行业务代码，最后释放掉锁，代码如下：

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　String lockkey = "lockkey";

　　Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,"lockvalue");//jedis.setnx

　　if(!result) {

　　return "";

　　}

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　springRedisTemplate.delete(lockkey);

　　return "end";

　　}

　　}

　　调用200次，压测结果显示还是有问题，只减掉了一部分：

　　这时，加大压测次数，结果正常了：

　　第一个应用扣减结果显示：

　　第二个应用扣减结果显示：

　　这个只是因为加大了调用次数，执行业务代码需要一点时间，这段时间拒绝了很多等待获取锁的请求。但是，还是有问题，假如redis服务挂掉了，抛出异常了，这时锁不会被释放掉，出现死锁问题，可以添加try catch处理，代码如下：

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　String lockkey = "lockkey";

　　try{郑州专业妇科医院 http://fk.zyfuke.com/

　　Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,"lockvalue");//jedis.setnx

　　if(!result) {

　　return "";

　　}

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　}finally{

　　springRedisTemplate.delete(lockkey);

　　}

　　return "end";

　　}

　　}

　　这时，Redis服务挂掉导致死锁的问题解决了，但是，如果服务器果宕机了，又会导致锁不能被释放的现象，所以可以设置超时时间为10s，代码如下：

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　String lockkey = "lockkey";

　　try{

　　Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,"lockvalue",10,TimeUnit.SECONDS);//jedis.setnx

　　//Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,"lockvalue");//jedis.setnx

　　//stringRedisTemplate.expire(lockkey,10,TimeUnit.SECONDS);

　　if(!result) {

　　return "";

　　}

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　}finally{

　　springRedisTemplate.delete(lockkey);

　　}

　　return "end";

　　}

　　}

　　这时，如果有一个线程执行需要15s，当执行到10s时第二个线程进来拿到这把锁，会出现多个线程拿到同一把锁执行，在第一个线程执行完时会释放掉第二个线程的锁，以此类推…就会导致锁的永久失效。所以，只能自己释放自己的锁，可以给当前线程取一个名字，代码如下：

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　String lockkey = "lockkey";

　　String clientId = UUID.randomUUID().toString();

　　try{

　　Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,clientId ,10,TimeUnit.SECONDS);//jedis.setnx

　　//Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,"lockvalue");//jedis.setnx

　　//stringRedisTemplate.expire(lockkey,10,TimeUnit.SECONDS);

　　if(!result) {

　　return "";

　　}

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　}finally{

　　springRedisTemplate.delete(lockkey);

　　}

　　return "end";

　　}

　　}

　　永久失效的问题解决了，但是，如果第一个线程执行15s，还是会存在多个线程拥有同一把锁的现象。所以，需要续期超时时间，当一个线程执行5s后对超时时间进行续期都10s，就可以解决了，续期设置可以借助redission工具。

　　Redission使用

　　Redission分布式锁实现原理

　　pom.xml

　　org.redisson

　　redisson

　　3.6.5

　　Application.java启动类

　　@bean

　　public Redission redission {

　　//此为单机模式

　　Config config = new Config();

　　config.useSingleServer().setAddress("redis://120.0.0.1:6379").setDatabase(0);

　　return (Redission)Redission.creat(config);

　　}

　　最终解决以上所有问题的代码如下：

　　@RestController

　　public class IndexController {

　　@Autowired

　　private StringRedisTemplate stringRedisTemplate;

　　@Autowired

　　private Redissionredission;

　　@RequestMapping("/deduct_stock")

　　public Stirng deductStock() {

　　String lockkey = "lockkey";

　　//String clientId = UUID.randomUUID().toString();

　　RLock lock = redission.getLock();

　　try{

　　//Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,clientId ,10,TimeUnit.SECONDS);//jedis.setnx

　　//Boolean result = stringRedisTemplate.opsForValue.setIfAbsent(lockkey,"lockvalue");//jedis.setnx

　　//stringRedisTemplate.expire(lockkey,10,TimeUnit.SECONDS);

　　//加锁:redission默认超时时间为30s，每10s续期一次，也可以自己设置时间

　　lock.lock(60,TimeUnit.SECONDS);

　　int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock"));//jedis.get(key)

　　if (stock > 0) {

　　int realStock = stock - 1;

　　stringRedisTemplate.opsForValue.set("stock",realStock+"");//jedis.set(key,value)

　　System.out.println(扣减成功，剩余库存：" + realStock + "");

　　} else {

　　System.out.println(扣减失败，库存不足!" );

　　}

　　}finally{

　　lock.unlock();

　　//springRedisTemplate.delete(lockkey);

　　}

　　return "end";

　　}

　　}

　　高并发分布式锁的问题得到解决。