redis专题：数据库和redis缓存一致性解决方案

太阳不下山

　　
　　文章目录

• 
  
  
  
  • 1.双写模式
    
  • 2.失效模式
    
  • 3.缓存一致性解决方案
    
  
  

　　redis缓存和数据库都保存了数据信息，当我们更新了数据库的数据时，应该如何保证redis和数据库的数据同步呢？当前比较常用的是双写模式和失效模式。

1.双写模式
　　双写模式：每次修改数据库的数据后，然后在更新redis中的数据，使用了两次写操作，称为双写模式
　　双写模式存在的问题：高并发下有可能会有脏数据
　　场景：

• 线程A在修改数据库之后、更新缓存之前，由于其他原因，cpu时间片被线程B抢到。
  
• 线程B改库、刷缓存一气呵成。
  
• 此时线程A再拿到cpu，执行更新缓存操作，那么此时缓存中的数据更新的就是线程A的脏数据，
  但其实我们想要的是线程B最新修改的数据，这样就出现了双写模式下不一致的情况，产生了脏数据！
  

　　原理如下图：


2.失效模式
　　失效模式：每次修改数据库的数据后，删除redis中缓存的数据，当有redis查询请求时，会先去数据库查询，然后更新到redis中，后续继续请求redis。更新时删除缓存数据，称为失效模式
　　失效模式存在的问题：同样会存在脏数据问题
　　场景一：先改数据库，再删redis！

• 线程A修改数据库数据，并删除了缓存中对应的数据。
  
• 线程B要获取数据，发现缓存中没有，就去数据库查到线程A修改后的数据，然后准备更新缓存，但在更新之前，由于其他原因，cpu时间片被线程C抢到了，
  
• 此时线程C也修改了数据库数据，并删除已经为空的缓存。
  
• 然后cpu又被线程B抢到，线程B继续执行更新缓存操作，此时缓存中更新的是线程A修改后的数据，但其实我们想要的是线程C修改后的数据，这就产生了数据不一致的情况！！
  

　　场景二：先删redis，再改数据库！

• 以减库存为例，假设库存量为100。线程A要减库存，会先删除redis数据，再对数据库中的库存量执行减 1 操作。
  
• 当线程A删除完redis数据， 准备执行减 1 操作时，cpu时间片被线程B抢占
  
• 线程B要查询库存，此时查到的还是原来的库存量（100），因为此时线程A还没来得及执行减 1 操作。然后线程B把原来的库存量100更新到redis
  
• 线程B执行完毕后，cpu时间片又回到线程A手中，线程A继续执行减 1 操作，执行完毕，数据库库存量为99，与redis中的100不相等，数据不一致！
  

　　2.1 延迟双删
　　延迟双删是在失效模式的基础上，在删除reids缓存时，让程序睡眠几十毫秒，再次执行删除缓存操作，可有效预防失效模式中缓存不一致问题，但是并不推荐。因为数据不一致问题本来就是极少情况发生的，如果使用延时双删，那么大部分正常的请求都会被阻塞几十毫秒，系统性能下降，显然得不偿失！

3.缓存一致性解决方案
　　如上所示，无论是双写模式还是失效模式，都无法完美解决缓存一致性问题！但在不同的业务场景对数据一致性的要求也不同，并非所有场景都需要数据强一致性，我们要根据实际业务场景来分析，具体解决方案如下所示：

• 对于并发很小的数据，比如个人信息、用户数据等。这些数据在使用双写或者失效模式后，由于并发量小，根本不需要考虑缓存一致性问题。可以给缓存数据加上过期时间，每隔一段时间触发读操作的主动更新即可！
  
• 如果并发量很高，但业务上能容忍短时间的缓存数据不一致，比如商品名称，商品分类三级菜单等。为缓存数据加上过期时间依然可以解决大部分业务对于缓存的要求。
  
• 如果并发量很高，且无法容忍数据不一致，比如库存。可以使用分布式锁来保证一致性！但也不用读写操作都加一把重量级的分布式锁，使用轻量级读写锁即可，通过添加读写锁保证写数据时读写都阻塞，仅读数据时相当于无锁！
  
• 还有一种方案，但代价也挺大。方案：所有的对库存的增、删、改、查操作都通过nginx路由到集群中某一台固定的机器上，在这台机器上定义一个内存队列。然后对库存的操作类型进行判断，每一个增、删操作放入队列，读操作不放入(防止队列元素过多)。然后从队列中逐个取出增、删操作并执行，读操作需要等到队列中没有增、删操作时才可执行！ 这种方案也可以阻止并发带来的数据不一致，但却降低了系统可用性！
  

　　3.1 redisson读写锁的底层原理

//获取读写锁
    RReadWriteLock readWriteLock = redisson.getReadWriteLock("123");
    
    //写锁
    RLock writeLock = readWriteLock.writeLock();
    writeLock.lock();
    System.out.println(111);
    writeLock.unlock();
    
//读锁
    RLock readLock = readWriteLock.readLock();
    readLock.lock();
    System.out.println(111);
    readLock.unlock();

　　redisson的readWriteLock其实和redisson的lock差不多，只不过加了个mode标识。底层的lua脚本根据mode的值，区分读写逻辑
　　①：如果加的是读锁，mode = read，并发读数据不阻塞

return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
    //增加mode属性
                "local mode = redis.call('hget', KEYS[1], 'mode'); " +
                "if (mode == false) then " +
                  "redis.call('hset', KEYS[1], 'mode', 'read'); " +
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                  "redis.call('set', KEYS[2] .. ':1', 1); " +
                  "redis.call('pexpire', KEYS[2] .. ':1', ARGV[1]); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
                "end; " +
                //如果mode为读，其他的读不阻塞
                "if (mode == 'read') or (mode == 'write' and redis.call('hexists', KEYS[1], ARGV[3]) == 1) then " +
                  "local ind = redis.call('hincrby', KEYS[1], ARGV[2], 1); " + 
                  "local key = KEYS[2] .. ':' .. ind;" +
                  "redis.call('set', key, 1); " +
                  "redis.call('pexpire', key, ARGV[1]); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
                "end;" +
                "return redis.call('pttl', KEYS[1]);",
            Arrays.<Object>asList(getName(), getReadWriteTimeoutNamePrefix(threadId)), 
            internalLockLeaseTime, getLockName(threadId), getWriteLockName(threadId));

　　②：如果加的是写锁，mode = write，并发写数据，其他读写都阻塞

return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
    //增加mode属性
              "local mode = redis.call('hget', KEYS[1], 'mode'); " +
              "if (mode == false) then " +
                  "redis.call('hset', KEYS[1], 'mode', 'write'); " +
                  "redis.call('hset', KEYS[1], ARGV[2], 1); " +
                  "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                  "return nil; " +
                "end; " +
                //如果mode为写，判断haxists，使其他的读写都阻塞
                "if (mode == 'write') then " +
                  "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " +
                    "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + 
                    "local currentExpire = redis.call('pttl', KEYS[1]); " +
                    "redis.call('pexpire', KEYS[1], currentExpire + ARGV[1]); " +
                    "return nil; " +
                  "end; " +
                "end;" +
                "return redis.call('pttl', KEYS[1]);",
            Arrays.<Object>asList(getName()), 
            internalLockLeaseTime, getLockName(threadId));

　　3.2 使用Canal解决缓存一致性问题
　　Canal使我们的业务代码只关注于数据库的交互，不用管redis缓存的问题，因为Canal可以订阅mysql数据库的每一次更新，只要mysql数据库有更新，Canal就会把数据同步到redis
　　使用Canal注意：

• ①：mysql要开启binlog日志，才能被Canal所监控
  
• ②：使用Canal在业务代码中执行修改缓存就可以
  
• ③：使用Canal需要额外增加Canal中间件，加重系统复杂度。
  

　　Canal操作流程图

　　总结：
　　以上我们针对的都是读多写少的情况加入缓存提高性能，如果写多读多的情况又不能容忍缓存数据不一致，那就没必要加缓存了，可以直接操作数据库！ 放入缓存的数据应该是对实时性、一致性要求不是很高的数据。切记不要为了用缓存，同时又要保证绝对的一致性做大量的过度设计和控制，增加系统复杂性！