高并发下之redis锁优化实战

背景：模拟商城秒杀活动减库存，给定库存300
代码：

@RestController
public class IndexControllerCOPY {
    


    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RequestMapping("/deduct_stock")

    public String deductStock() {
    
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            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";
    }

}

前提条件：模拟集群部署模式下，对库存操作，所以启动两个Tomcat实例，8080和9080端口。并用nginx做负载均衡，配置见下图；


启动redis，然后再启动两个实例，用jmeter压测，通过访问80端口，让nginx监听去交给对应的服务执行请求，1秒200个请求试下，压测配置如下图：

查看结果：
可以发现在库存重复扣减了，一件商品多卖，出现超卖，显然不正确。这个时候怎么解决呢？有些道友可能会说这是因为没有加锁，所以在8080的服务器里面看到了超卖，好的满足你的需求，我加上锁在测试：

@RestController
public class IndexControllerCOPY {
    


    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @RequestMapping("/deduct_stock")

    public String deductStock() {
    
        synchronized (this){
    
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            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";
    }

}

可以发现：第198在8080和9080服务都出现了。所以要明白，你的synchronize只是针对线程上加锁，我都是集群模式，不同的进程之间你的锁没有可能给另一个加上。至于8080服务为什么会出现2条197的记录，聪明的你自己揣测下，很简单的（8080内卷，9080天天摸鱼，8080人修改的值，被摸鱼先生覆盖了）。
好了今天比较忙，深入优化放到下次继续，如果有人需要看的话，我就找时间完善。

打个分割线今天继续说集群模式下如何保证数据一致性
《-----------------------------------------------------------------------------------------》
上面做了那么多铺垫就是为了告诉大家：集群或者分布式模式下不能通过synchronize或者ReentrantLock来保持数据的一致性
解决方式有很多种，可以通过zookeeper的顺序节点或者redis的setnx命令解决
这里我以redis举例继续：
首先我们还是将商品数量置为200，修改代码使用redis的setnx命令，jmeter做压测。


观察结果：8080服务器

9090服务器：

redis上该商品剩余库存：

可以看出：两个进程没有卖出相同的货物，并且最终卖出的商量数量103+97=200，完全符合。在5秒内2000个并发下保证了数据的一致性。
当然对于此处代码还有优化空间：
1.如果程序出错，没有及时释放锁，所以我们将释放锁的逻辑放在finally代码块中

package com.redisson;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class IndexControllerCOPY {
    


    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private volatile int count;

    @RequestMapping("/deduct_stock")

    public String deductStock() {
    
// synchronized (this){
    
        // redis 的分布式锁setnx
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock","value"); //分布式锁 jedis.setnx(k,v)
        try {
    
            // 如果获取到锁会返回true
            if (!lock){
    
                return "error";
            }
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            if (stock > 0) {
    
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
                System.out.println("扣减成功，剩余库存:" + realStock);
                count++;
            } else {
    
                System.out.println("扣减失败，库存不足,总计卖出"+count);

            }
        }finally {
    
            // 没有获得锁不能释放他人的锁
            if (lock){
    
                stringRedisTemplate.delete("lock");
            }

        }
        return "end";
    }

}

2.比如说如果8080服务在获取到锁，业务逻辑还没处理到释放锁，服务宕机，锁也不会被释放，所以可以添加一个过期时间
3.但是添加了一个过期时间后会出现另两个个问题，一、（线程A获取锁后处理时间比较长大于锁的过期时间，这个时候另一个线程B会获取到所，这个时候前一个线程业务处理完，去释放的锁是B的，依次往复很大GUB），解决方式让每个线程释放自己加的锁，怎么区别呢？很简单，每个加锁的线程给其设置一个局部变量的value值，当你要释放锁时取出key-value,判断value是否与当前线程栈的值相等，相等就释放.

package com.redisson;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;
import java.util.concurrent.TimeUnit;

@RestController
public class IndexControllerCOPY {
    


    @Autowired
    private StringRedisTemplate stringRedisTemplate;
    private volatile int count;

    @RequestMapping("/deduct_stock")

    public String deductStock() {
    
        String uuid= UUID.randomUUID().toString();
// synchronized (this){
    
        // redis 的分布式锁setnx
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock","value",10, TimeUnit.MILLISECONDS); //分布式锁 jedis.setnx(k,v)
        try {
    
            // 如果获取到锁会返回true
            if (!lock){
    
                return "error";
            }
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            if (stock > 0) {
    
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
                System.out.println("扣减成功，剩余库存:" + realStock);
                count++;
            } else {
    
                System.out.println("扣减失败，库存不足,总计卖出"+count);

            }
        }finally {
    
            // 没有获得锁不能释放他人的锁
            if (lock && uuid.equals(stringRedisTemplate.opsForValue().get("lock"))){
    
                stringRedisTemplate.delete("lock");
            }

        }
        return "end";
    }

}

二、（还是线程A业务没执行完，锁过期时间到了，线程B拿到了锁，也不行，所以需要给锁续命） 解决问题：保证在加了过期时间的前提下，又要保证库库存操作完才将锁释放。锁续命，给一个子线程定时任务iterm，
// 每过一段时间看下主线程是否执行完，没有就重新在加一个过期时间
此处还有个比较好的框架redission可以完美解决上面所有问题。 方案：
// 有一个实现好了的客户端工具包
// 导入
//
// org.redisson
// redisson
// 3.6.5
//

好累，后面在补上redisson类容

package com.redisson;

import org.redisson.Redisson;
import org.redisson.api.RLock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@RestController
public class IndexController {
    

    @Autowired
    private Redisson redisson;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    private volatile int count;


    @RequestMapping("/deduct_stock")
        

    public String deductStock() {
    
        String lockKey = "product_101";
        RLock redissonLock = redisson.getLock(lockKey);
        try {
    

            //加锁
            redissonLock.lock();  //setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS);

            // synchronized (this){ 方案一加synchronized，但是只能解决在同一个进程单机应用的情况，如果分布式部署不可行
            int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
            if (stock > 0) {
    
                int realStock = stock - 1;
                stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
                count++;
                System.out.println("扣减成功，剩余库存:" + realStock);
            } else {
    
                System.out.println("扣减失败，库存不足,总计卖出"+count);
            }
            // }

        } finally {
    
            redissonLock.unlock();
            /*if (lock && clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) { stringRedisTemplate.delete(lockKey); }*/
        }

        return "end";
    }




}