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

/*这里记为代码一*/@RestControllerpublic 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.javaredisson-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操作较慢（小卡顿）呢？

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

但这样还是有问题

还有什么办法解决？

（10）解决双写不统一的办法

提早双删：删缓存删两次，删除之后sleep(一段时间)后再删一次
但这种办法只能说是缩小，并不能解决问题，并且还让所有的写申请都得sleep一段时间

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

还有没有其余解决办法？

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

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

（11）读多写少的状况

间接上分布式锁会有问题，应用读写锁
读写锁：读操作加读锁，写操作加写锁，读操作不互斥，写锁跟读锁、写锁跟写锁互斥。
因为很多零碎都是读多写少的状况，所以能够进步性能

/*代码九*/@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的数据即可。

（12）读多写多的状况

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

原本想附上视频地址的，但被驳回了说是广告。。。这里我说一下以上是从诸葛老师那里学习的，仅作集体记录