Redis高并发架构实战

（1）先来一个小案例作为切入点

/*
这里记为代码一
*/
@RestController
public class IndexController {

    @Autowired
    private Redisson redisson;
    @Autowired
    private StringRedisTemplate stringRedisTemplate;    //组件spring-boot-starter-data-redis

    @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";
    }
}

然后在redis中搞一个库存为200

现在很明显，代码一 存在线程安全问题，会有可能读到都是200，然后都减1后设置为199，就不对了。
很多同学都会想到加一把锁

（2）synchronized

/*
代码二
*/
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";
}

这样的确是只能有一个线程执行操作，确实是线程安全了。但是它只能在单机环境下运行，只能锁住一个tomcat，分布式的时候就不行了。

（3）分布式锁

这时，应该考虑分布式锁。SETNX（SET if Not eXists）。和set的区别是：
set tuling A
set tuling B
结果会是B
setnx tuling A
setnx tuling B
结果会是A

/*
代码三
*/
public String deductStock(){

    String lockKey = "product_101";
    //如果返回false，说明redis中有这个key了，不做任何操作。如果返回true说明执行这个命令之前没有这个key，并设置成功了
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
    if(!result){
        return "error_code";    //给前端错误码，当前系统繁忙，请稍后再试
    }

    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("扣减失败，库存不足");
    }

    stringRedisTemplate.delete(lockKey);

    return "end";
}

redis那边是单线程操作的，会排队，只有排队头的可以设置成功，后面的设置不成功，这样入门级的分布式锁设计完了。大家想想还有没有问题？
这个时候还是存在问题，当获取到锁的线程有异常，导致没法删除key，就会导致其他线程获取不到锁，就算能捕获异常，但如果是系统挂了呢，运维重启呢

/*
代码四
*/
public String deductStock(){

    String lockKey = "product_101";
    //如果返回false，说明redis中有这个key了，不做任何操作。如果返回true说明执行这个命令之前没有这个key，并设置成功了
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
    if(!result){
        return "error_code";    //给前端错误码，当前系统繁忙，请稍后再试
    }

    try{
        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("扣减失败，库存不足");
        }
    }finally {
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

示例 代码四 还是存在问题，大家先想想解决方法。

（4）锁超时

这样的话，可以加一个超时时间来解决，给key一个超时时间，即使系统挂了，一段时间之后，其他机器还是能正常访问

/*
代码五
*/
public String deductStock(){

    String lockKey = "product_101";
    //如果返回false，说明redis中有这个key了，不做任何操作。如果返回true说明执行这个命令之前没有这个key，并设置成功了
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
    stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

    if(!result){
        return "error_code";    //给前端错误码，当前系统繁忙，请稍后再试
    }

    try{
        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("扣减失败，库存不足");
        }
    }finally {
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

大家想想 代码五 还有问题吗？

（5）加锁操作原子性

假设设置了key之后，正准备设置超时时间，但系统挂了，那还是回到之前的问题了，得保证原子性。应该使用setIfAbsent的其他重载方法，有一个是可以同时设置超时时间的

/*
代码六
*/
public String deductStock(){

    String lockKey = "product_101";
    //如果返回false，说明redis中有这个key了，不做任何操作。如果返回true说明执行这个命令之前没有这个key，并设置成功了
//        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
//        stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling", 10, TimeUnit.SECONDS);

    if(!result){
        return "error_code";    //给前端错误码，当前系统繁忙，请稍后再试
    }

    try{
        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("扣减失败，库存不足");
        }
    }finally {
        stringRedisTemplate.delete(lockKey);
    }

    return "end";
}

大家思考一下，代码六 还有没有问题？
遇到高并发的时候，通常执行会比较慢，慢执行啊，中间sql语句执行很慢这样，假设执行完这个方法需要15秒，当线程执行了10秒的时候，由于设置了超时时间是10秒，并且是高并发场景，这个时候key就删除了，另外的线程就获取了锁

这样就相当于锁永久失效。虽然把过期时间放大是可以避免，但还是无法彻底解决问题。
本质是自己加的锁被别人解掉了，所以解决就是锁只能自己解锁

/*
代码七
*/
public String deductStock(){

    String lockKey = "product_101";
    //如果返回false，说明redis中有这个key了，不做任何操作。如果返回true说明执行这个命令之前没有这个key，并设置成功了
//        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
//        stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

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

    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 10, TimeUnit.SECONDS);

    if(!result){
        return "error_code";    //给前端错误码，当前系统繁忙，请稍后再试
    }

    try{
        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("扣减失败，库存不足");
        }
    }finally {
        if(clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
            stringRedisTemplate.delete(lockKey);
        }
    }

    return "end";
}

代码七 按上面的例子，锁是自己过期的，这代码只是能保证线程1无法删除线程2的锁，但线程1和线程2还是同时在跑啊。这个时间还有问题，但是先不管，先放放，因为不是想要引申的内容，要继续思考这个代码还有除时间外的什么其他问题？
就是finally中的两行代码非原子，写并发代码和写高并发代码时的区别，应该要习惯性的在代码之间空几行，表明这里执行有时间差，非原子。
假设执行判断完clientId确实是等于当前线程的value，假设这时刚好是9.9秒，突然发生卡顿，但这个if判断已经是true了，正准备delete的时候，卡顿了，这时已经过了10秒，线程2已经获取了锁，然后线程1执行delete，又出问题了，仍然是线程1删除了线程2的锁。
怎么处理？

（6）锁续命

锁续命：通常是这样处理的，有一个分线程定时任务，用来监测线程还是否持有锁，还持有的就延长锁的过期时间，例如锁超时是30秒，那么分线程每10秒判断一下，线程还是否持有锁，还持有就更新过期时间，不能说是延长，是按当前时间又重新设置30秒过期，当不持有了，定时任务就结束，分线程也结束。
redisson：操作redis的客户端，有很多分布式功能，其中就有分布式锁。想起了吧？代码一 中就已经引入了redisson

/*
代码八
*/
public String deductStock(){

    String lockKey = "product_101";
    //如果返回false，说明redis中有这个key了，不做任何操作。如果返回true说明执行这个命令之前没有这个key，并设置成功了
//        Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
//        stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

    /*String clientId = UUID.randomUUID().toString();
    Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 10, TimeUnit.SECONDS);
    if(!result){
        return "error_code";    //给前端错误码，当前系统繁忙，请稍后再试
    }*/

    RLock redissonLock = redisson.getLock(lockKey);

    try{
        //加锁
        redissonLock.lock();    //理解为执行了setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS)
        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("扣减失败，库存不足");
        }
    }finally {
        redissonLock.unlock();
        /*if(clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))){
            stringRedisTemplate.delete(lockKey);
        }*/
    }

    return "end";
}

redisson加锁核心lua脚本

KEYS[1]：product_101
ARGV[2]：getLockName(threadId)
ARGV[1]：internalLockLeaseTime（初始化是30秒）
可以看到第250行和251行，就相当于 代码五 中的

Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "tuling");     //就理解为jedis.setnx(key,value)
stringRedisTemplate.expire(lockKey, 10, TimeUnit.SECONDS);

而这两行代码是不具有原子性的，线程不安全。Lua脚本可以保证原子性

锁续命：

/*
https://github.com/redisson/redisson/blob/redisson-3.6.5/redisson/src/main/java/org/redisson/RedissonLock.java
redisson-3.6.5 RedissonLock.java，其他版本会不太不一样，但原理应该不变吧
*/
    private void scheduleExpirationRenewal(final long threadId) {
        if (expirationRenewalMap.containsKey(getEntryName())) {
            return;
        }

        Timeout task = commandExecutor.getConnectionManager().newTimeout(new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {

                RFuture<Boolean> future = commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                        "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                            "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                            "return 1; " +
                        "end; " +
                        "return 0;",
                          Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId));

                future.addListener(new FutureListener<Boolean>() {
                    @Override
                    public void operationComplete(Future<Boolean> future) throws Exception {
                        expirationRenewalMap.remove(getEntryName());
                        if (!future.isSuccess()) {
                            log.error("Can't update lock " + getName() + " expiration", future.cause());
                            return;
                        }

                        if (future.getNow()) {
                            // reschedule itself
                            scheduleExpirationRenewal(threadId);
                        }
                    }
                });
            }
        }, internalLockLeaseTime / 3, TimeUnit.MILLISECONDS);

        if (expirationRenewalMap.putIfAbsent(getEntryName(), task) != null) {
            task.cancel();
        }
    }

它延迟 internalLockLeaseTime / 3 秒执行run方法，为它重新设置expire为 internalLockLeaseTime。
commandExecutor.evalWriteAsync返回了一个future，然后future又添加监听器，最后执行当前方法scheduleExpirationRenewal(threadId);，就是一直重复续命，又再延迟调用，相当于定时任务。

到目前为止，基本就没有什么坑了，redisson已经是填了很多坑，可以放心使用 代码八 进行实现。
但是，还有点问题，假设有多个请求在执行redissonLock.lock()加锁，只能有一个线程在处理，其他都得等着，系统就会很慢，存在性能问题，该怎么优化能做到双十一能用的级别？

（7）zookeeper

redis一般都是有主从架构的，基本不会是单机使用

redis主节点马上告诉客户端加锁成功，线程1就执行业务代码逻辑，然后redis准备把key同步给从节点时候，结果主节点挂了，某个从节点选举成为新的Master主节点，来了个线程3访问新的主节点加锁，线程3就发现没有product_101这个key，又可以加锁成功了，线程1业务逻辑还没执行完毕，线程3就开始执行，就又出现了问题

主从架构锁失效的问题，可以用zookeeper来实现分布式锁，和redis类似，是树形结构。redis更多的实现是AP架构，zookeeper更多的实现是CAP架构。
zookeeper的话，当要写一个key，不是就立即返回成功的，会先把key同步给集群的其他节点，子节点会返回同步成功的信息，主节点会判断是否已经有超过半数的子节点都同步成功，这时才告诉客户端成功了，是为了保证一致性，牺牲了及时响应，但它能保证那些已经同步了子节点才能成功leader，redis就没有这个机制，也就是线程3再来请求leader的时候，必然会有key，加锁就不成功，解决了上述问题。
但如果不使用zookeeper，就是要使用redis来解决呢？（因为redis的并发比zookeeper高不少）如果要高并发，就用redis，就有上述主从锁问题，如果要保证健壮性就用zookeeper，但牺牲了并发数。

（8）Redlock

硬是要使用redis的话，看看Redlock

redis没有主从关系，是对等的，往每个节点发送加锁命令，只有超过半数的节点返回成功才认为客户端加锁成功，和zookeeper原理类似。但这种方式不推荐，原来是一个redis节点，现在搞多个，要半数加锁成功，对我们加锁性能受一定影响，这样的话，还不如用zookeeper，因为redlock还有不少问题。

@RequestMapping("/redlock")
public String redlock(){

    String lockKey = "product_101";
    RLock lock1 = redisson.getLock(lockKey);
    RLock lock2 = redisson.getLock(lockKey);
    RLock lock3 = redisson.getLock(lockKey);

    //根据多个RLock对象构件RedissonRedLock
    RedissonRedLock redLock = new RedissonRedLock(lock1, lock2, lock3);

    try{
        /*
        * waitTimeout 尝试获取锁的最大等待时间，超过这个数，则认为获取锁失败
        * leaseTime   锁的持有时间，超过这个时间锁会自动失败（值应设置为大于业务处理的时间，确保在锁有效期内业务能处理完）
        * */
        boolean res = redLock.tryLock(10, 30, TimeUnit.SECONDS);
        if(res){
            //成功获取锁，处理业务
        }
    }catch (Exception e) {
        throw  new RuntimeException("lock fail");
    }finally {
        //无论如何，最后都要解锁
        redLock.unlock();
    }

    return "end";
}

回到 代码八，redissonLock.lock();会导致其他线程等待，也就是分布式锁把并行请求变串行化执行了。那么如何提升分布式锁性能？

（9）分段锁

模仿ConcurrentHashMap，分段锁。
假设product_101的数量是200，那么可以分十段，
product_101_1=20
product_101_2=20
product_101_3=20
......
product_101_10=20
200个库存分10个key存到redis中去，让每个线程去减不同的段位的库存，如果不够减的话就减一下个段位，实现的话有点难，但可以理解这个思想，就不再去扩展了。

（10）缓存数据库双写不一致

接下来说redis作为缓存使用的时候，常见问题有：缓存无底洞、缓存穿透、缓存雪崩、缓存失效、热点key倾斜、热点key重建、缓存数据库双写不一致。
这里针对缓存数据库双写不一致的问题说一下。
什么是缓存数据库双写不一致？

看上去线程1写数据库，然后更新缓存，线程2写数据库，然后更新缓存，没有什么问题，但如果线程1操作较慢（小卡顿）呢？

有些人就会说，通常不会直接更新缓存，而是把缓存删掉，即更新就删缓存，读数据的时候再设置缓存，的确这样是比较好，因为每次写完就更新缓存的话，如果不读缓存，相当于白更新。

但这样还是有问题

还有什么方法解决？

（11）解决双写不一致的方法

延迟双删：删缓存删两次，删除之后sleep(一段时间)后再删一次
但这种方法只能说是减少，并不能解决问题，并且还让所有的写请求都得sleep一段时间

内存队列：用hash运算把操作路由到某个队列中顺序执行。是可以解决，但复杂，写不好很可能有性能问题或是bug

还有没有其他解决方法？

问题的本质就是操作过程中不是原子性，如果（写数据库-删除缓存）是不可分割的操作，（查缓存-查数据库-更新缓存）是不可分割的操作，即在操作前加分布式锁，操作完后解锁，所有线程的操作为队列，把多个并发执行的线程串行化

直接这样上锁，性能肯定是有问题的，怎么优化？

（12）读多写少的情况

直接上分布式锁会有问题，使用读写锁
读写锁：读操作加读锁，写操作加写锁，读操作不互斥，写锁跟读锁、写锁跟写锁互斥。
由于很多系统都是读多写少的情况，所以可以提高性能

/*
代码九
*/
@RequestMapping("/get_stock")
public String getStock(@RequestParam("clientId") Long clientId) throws InterruptedException{

    String lockKey = "product_101";

    RReadWriteLock readWriteLock = redisson.getReadWriteLock(lockKey);
    RLock rLock = readWriteLock.readLock();

    rLock.lock();
    System.out.println("获取读锁成功：client="+clientId);
    String stock = stringRedisTemplate.opsForValue().get("stock");
    if(StringUtils.isEmpty(stock)){
        System.out.println("查询数据库库存为10。。。");
        Thread.sleep(5000);
        stringRedisTemplate.opsForValue().set("stock", 10);
    }
    rLock.unlock();
    System.out.println("释放读锁成功：client="+clientId);

    return "end";
}

@RequestMapping("/update_stock")
public String updateStock(@RequestParam("clientId") Long clientId) throws InterruptedException{

    String lockKey = "product_101";

    RReadWriteLock readWriteLock = redisson.getReadWriteLock(lockKey);
    RLock writeLock = readWriteLock.writeLock();

    writeLock.lock();
    System.out.println("获取写锁成功：client="+clientId);
    System.out.println("修改商品101的数据库库存为6。。。");
    stringRedisTemplate.delete("stock");
    Thread.sleep(5000);
    writeLock.unlock();
    System.out.println("释放写锁成功：client="+clientId);

    return "end";
}

原理就是lua脚本为每个key设置一个mode的值来记录是read还是write。
RedissonWriteLock.java

但如果读多写也多的情况呢，怎么处理？
不采用上面的方法，仍然是给缓存过期时间，然后操作的时候直接操作数据库。例如在页面上看到的库存，其实很多时候都是和数据库的值不一致的，就是为了实现高并发，又要用数据库又要用缓存，只能牺牲一致性，牺牲一致性其实关系并不大，想一想，假设一致的话，加入购物车、下订单，中间是有时间差的，这个时候可能就没有了库存了，对用户来说是不一致，但对程序来说，程序以及保证了一致，只是意义不大，所以牺牲一致性来提高性能。假设过期时间是一分钟，那在这一分钟内可能是不一致，但如果一分钟后库存不变，又读取更新了缓存，这个时候就变一致了，只需要确保在下单的时候是用db的数据即可。

（13）读多写多的情况

如果是读多写多，又要保证缓存数据库一致性，怎么办？
对读多写多的场景，就不应该用缓存，直接操作数据库就好了，对吧。
也有方法既使用缓存，又应对读多写多的场景，中间件canal。后面就学不着了，需要报课。。。。以后再看看