目录
八、分布式锁
8.1、分布式锁
8.2、redis
1、加锁
2、解锁
3、实现
8.3、redisson
1、可重入锁
2、获取锁实例
3、加锁
4、解锁
九、常见缓存问题与解决方法
9.1、缓存处理流程
9.2、缓存穿透
9.3、缓存击穿
9.4、缓存雪崩
9.5缓存双写一致性
9.5.1、先更新数据库，再更新缓存
9.5.2、先删除缓存，再更新数据库
9.5.3、先更新数据库，再删除缓存
9.5.4、缓存的不适用场景
在Java中，关于锁我想大家都很熟悉。在并发编程中，我们通过锁，来避免由于竞争而造成的数据不一致问题。通常，我们以synchronized 、Lock来使用它。
但是Java中的锁，只能保证在同一个JVM进程内中执行。如果在分布式集群环境下呢？
分布式锁，是一种思想，它的实现方式有很多。比如，我们将沙滩当做分布式锁的组件，那么它看起来应该是这样的：
加锁在沙滩上踩一脚，留下自己的脚印，就对应了加锁操作。其他进程或者线程，看到沙滩上已经有脚印，证明锁已被别人持有，则等待。
解锁把脚印从沙滩上抹去，就是解锁的过程。
锁超时为了避免死锁，我们可以设置一阵风，在单位时间后刮起，将脚印自动抹去。
分布式锁的实现有很多，比如基于数据库、memcached、Redis、系统文件、zookeeper等。它们的核心的理念跟上面的过程大致相同。
我们先来看如何通过单节点Redis实现一个简单的分布式锁。
加锁实际上就是在redis中，给Key键设置一个值，为避免死锁，并给定一个过期时间。
SET lock_key random_value NX PX 5000
值得注意的是：random_value 是客户端生成的唯一的字符串。NX 代表只在键不存在时，才对键进行设置操作。PX 5000 设置键的过期时间为5000毫秒。
这样，如果上面的命令执行成功，则证明客户端获取到了锁。
解锁的过程就是将Key键删除。但也不能乱删，不能说客户端1的请求将客户端2的锁给删除掉。这时候random_value的作用就体现出来。
为了保证解锁操作的原子性，我们用LUA脚本完成这一操作。先判断当前锁的字符串是否与传入的值相等，是的话就删除Key，解锁成功。
if redis.call('get',KEYS[1]) == ARGV[1] then return redis.call('del',KEYS[1]) else return 0 end首先，我们在pom文件中，引入Jedis。在这里，笔者用的是最新版本，注意由于版本的不同，API可能有所差异。
<dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>3.0.1</version> </dependency>加锁的过程很简单，就是通过SET指令来设置值，成功则返回；否则就循环等待，在timeout时间内仍未获取到锁，则获取失败。
@Service public class RedisLock { Logger logger = LoggerFactory.getLogger(this.getClass()); private String lock_key = "redis_lock"; //锁键 protected long internalLockLeaseTime = 30000;//锁过期时间 private long timeout = 999999; //获取锁的超时时间 //SET命令的参数 SetParams params = SetParams.setParams().nx().px(internalLockLeaseTime); @Autowired JedisPool jedisPool; /** * 加锁 * @param id * @return */ public boolean lock(String id){ Jedis jedis = jedisPool.getResource(); Long start = System.currentTimeMillis(); try{ for(;;){ //SET命令返回OK ，则证明获取锁成功 String lock = jedis.set(lock_key, id, params); if("OK".equals(lock)){ return true; } //否则循环等待，在timeout时间内仍未获取到锁，则获取失败 long l = System.currentTimeMillis() - start; if (l>=timeout) { return false; } try { Thread.sleep(100); } catch (InterruptedException e) { e.printStackTrace(); } } }finally { jedis.close(); } } }解锁我们通过jedis.eval来执行一段LUA就可以。将锁的Key键和生成的字符串当做参数传进来。
/** * 解锁 * @param id * @return */ public boolean unlock(String id){ Jedis jedis = jedisPool.getResource(); String script = "if redis.call('get',KEYS[1]) == ARGV[1] then" + " return redis.call('del',KEYS[1]) " + "else" + " return 0 " + "end"; try { Object result = jedis.eval(script, Collections.singletonList(lock_key), Collections.singletonList(id)); if("1".equals(result.toString())){ return true; } return false; }finally { jedis.close(); } }最后，我们可以在多线程环境下测试一下。我们开启1000个线程，对count进行累加。调用的时候，关键是唯一字符串的生成。这里，笔者使用的是Snowflake算法。
@Controller public class IndexController { @Autowired RedisLock redisLock; int count = 0; @RequestMapping("/index") @ResponseBody public String index() throws InterruptedException { int clientcount =1000; CountDownLatch countDownLatch = new CountDownLatch(clientcount); ExecutorService executorService = Executors.newFixedThreadPool(clientcount); long start = System.currentTimeMillis(); for (int i = 0;i<clientcount;i++){ executorService.execute(() -> { //通过Snowflake算法获取唯一的ID字符串 String id = IdUtil.getId(); try { redisLock.lock(id); count++; }finally { redisLock.unlock(id); } countDownLatch.countDown(); }); } countDownLatch.await(); long end = System.currentTimeMillis(); logger.info("执行线程数:{},总耗时:{},count数为:{}",clientcount,end-start,count); return "Hello"; } }至此，单节点Redis的分布式锁的实现就已经完成了。比较简单，但是问题也比较大，最重要的一点是，锁不具有可重入性。
可参考如何优雅地用Redis实现分布式锁一步步讲解
http://www.redis.cn/articles/20181020004.html
Redisson是架设在Redis基础上的一个Java驻内存数据网格（In-Memory Data Grid）。充分的利用了Redis键值数据库提供的一系列优势，基于Java实用工具包中常用接口，为使用者提供了一系列具有分布式特性的常用工具类。使得原本作为协调单机多线程并发程序的工具包获得了协调分布式多机多线程并发系统的能力，大大降低了设计和研发大规模分布式系统的难度。同时结合各富特色的分布式服务，更进一步简化了分布式环境中程序相互之间的协作。
相对于Jedis而言，Redisson强大的一批。当然了，随之而来的就是它的复杂性。它里面也实现了分布式锁，而且包含多种类型的锁，更多请参阅分布式锁和同步器
上面我们自己实现的Redis分布式锁，其实不具有可重入性。那么下面我们先来看看Redisson中如何调用可重入锁。
在这里，笔者使用的是它的最新版本，3.10.1。
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson</artifactId> <version>3.10.1</version> </dependency>首先，通过配置获取RedissonClient客户端的实例，然后getLock获取锁的实例，进行操作即可。
public static void main(String[] args) { Config config = new Config(); config.useSingleServer().setAddress("redis://127.0.0.1:6379"); config.useSingleServer().setPassword("redis1234"); final RedissonClient client = Redisson.create(config); RLock lock = client.getLock("lock1"); try{ lock.lock(); }finally{ lock.unlock(); } }我们先来看RLock lock = client.getLock("lock1"); 这句代码就是为了获取锁的实例，然后我们可以看到它返回的是一个RedissonLock对象。
public RLock getLock(String name) { return new RedissonLock(connectionManager.getCommandExecutor(), name); }在RedissonLock构造方法中，主要初始化一些属性。
public RedissonLock(CommandAsyncExecutor commandExecutor, String name) { super(commandExecutor, name); //命令执行器 this.commandExecutor = commandExecutor; //UUID字符串 this.id = commandExecutor.getConnectionManager().getId(); //内部锁过期时间 this.internalLockLeaseTime = commandExecutor. getConnectionManager().getCfg().getLockWatchdogTimeout(); this.entryName = id + ":" + name; }当我们调用lock方法，定位到lockInterruptibly。在这里，完成了加锁的逻辑。
public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException { //当前线程ID long threadId = Thread.currentThread().getId(); //尝试获取锁 Long ttl = tryAcquire(leaseTime, unit, threadId); // 如果ttl为空，则证明获取锁成功 if (ttl == null) { return; } //如果获取锁失败，则订阅到对应这个锁的channel RFuture<RedissonLockEntry> future = subscribe(threadId); commandExecutor.syncSubscription(future); try { while (true) { //再次尝试获取锁 ttl = tryAcquire(leaseTime, unit, threadId); //ttl为空，说明成功获取锁，返回 if (ttl == null) { break; } //ttl大于0 则等待ttl时间后继续尝试获取 if (ttl >= 0) { getEntry(threadId).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS); } else { getEntry(threadId).getLatch().acquire(); } } } finally { //取消对channel的订阅 unsubscribe(future, threadId); } //get(lockAsync(leaseTime, unit)); }如上代码，就是加锁的全过程。先调用tryAcquire来获取锁，如果返回值ttl为空，则证明加锁成功，返回；如果不为空，则证明加锁失败。这时候，它会订阅这个锁的Channel，等待锁释放的消息，然后重新尝试获取锁。流程如下：
获取锁
获取锁的过程是怎样的呢？接下来就要看tryAcquire方法。在这里，它有两种处理方式，一种是带有过期时间的锁，一种是不带过期时间的锁。
private <T> RFuture<Long> tryAcquireAsync(long leaseTime, TimeUnit unit, final long threadId) { //如果带有过期时间，则按照普通方式获取锁 if (leaseTime != -1) { return tryLockInnerAsync(leaseTime, unit, threadId, RedisCommands.EVAL_LONG); } //先按照30秒的过期时间来执行获取锁的方法 RFuture<Long> ttlRemainingFuture = tryLockInnerAsync( commandExecutor.getConnectionManager().getCfg().getLockWatchdogTimeout(), TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG); //如果还持有这个锁，则开启定时任务不断刷新该锁的过期时间 ttlRemainingFuture.addListener(new FutureListener<Long>() { @Override public void operationComplete(Future<Long> future) throws Exception { if (!future.isSuccess()) { return; } Long ttlRemaining = future.getNow(); // lock acquired if (ttlRemaining == null) { scheduleExpirationRenewal(threadId); } } }); return ttlRemainingFuture; }接着往下看，tryLockInnerAsync方法是真正执行获取锁的逻辑，它是一段LUA脚本代码。在这里，它使用的是hash数据结构。
<T> RFuture<T> tryLockInnerAsync(long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) { //过期时间 internalLockLeaseTime = unit.toMillis(leaseTime); return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command, //如果锁不存在，则通过hset设置它的值，并设置过期时间 "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('hset', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + //如果锁已存在，并且锁的是当前线程，则通过hincrby给数值递增1 "if (redis.call('hexists', KEYS[1], ARGV[2]) == 1) then " + "redis.call('hincrby', KEYS[1], ARGV[2], 1); " + "redis.call('pexpire', KEYS[1], ARGV[1]); " + "return nil; " + "end; " + //如果锁已存在，但并非本线程，则返回过期时间ttl "return redis.call('pttl', KEYS[1]);", Collections.<Object>singletonList(getName()), internalLockLeaseTime, getLockName(threadId)); }这段LUA代码看起来并不复杂，有三个判断：
通过exists判断，如果锁不存在，则设置值和过期时间，加锁成功通过hexists判断，如果锁已存在，并且锁的是当前线程，则证明是重入锁，加锁成功如果锁已存在，但锁的不是当前线程，则证明有其他线程持有锁。返回当前锁的过期时间，加锁失败加锁成功后，在redis的内存数据中，就有一条hash结构的数据。Key为锁的名称；field为随机字符串+线程ID；值为1。如果同一线程多次调用lock方法，值递增1。
127.0.0.1:6379> hgetall lock1 1) "b5ae0be4-5623-45a5-8faa-ab7eb167ce87:1" 2) "1"我们通过调用unlock方法来解锁。
public RFuture<Void> unlockAsync(final long threadId) { final RPromise<Void> result = new RedissonPromise<Void>(); //解锁方法 RFuture<Boolean> future = unlockInnerAsync(threadId); future.addListener(new FutureListener<Boolean>() { @Override public void operationComplete(Future<Boolean> future) throws Exception { if (!future.isSuccess()) { cancelExpirationRenewal(threadId); result.tryFailure(future.cause()); return; } //获取返回值 Boolean opStatus = future.getNow(); //如果返回空，则证明解锁的线程和当前锁不是同一个线程，抛出异常 if (opStatus == null) { IllegalMonitorStateException cause = new IllegalMonitorStateException(" attempt to unlock lock, not locked by current thread by node id: " + id + " thread-id: " + threadId); result.tryFailure(cause); return; } //解锁成功，取消刷新过期时间的那个定时任务 if (opStatus) { cancelExpirationRenewal(null); } result.trySuccess(null); } }); return result; }然后我们再看unlockInnerAsync方法。这里也是一段LUA脚本代码。
protected RFuture<Boolean> unlockInnerAsync(long threadId) { return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, EVAL, //如果锁已经不存在， 发布锁释放的消息 "if (redis.call('exists', KEYS[1]) == 0) then " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; " + "end;" + //如果释放锁的线程和已存在锁的线程不是同一个线程，返回null "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " + "return nil;" + "end; " + //通过hincrby递减1的方式，释放一次锁 //若剩余次数大于0 ，则刷新过期时间 "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " + "if (counter > 0) then " + "redis.call('pexpire', KEYS[1], ARGV[2]); " + "return 0; " + //否则证明锁已经释放，删除key并发布锁释放的消息 "else " + "redis.call('del', KEYS[1]); " + "redis.call('publish', KEYS[2], ARGV[1]); " + "return 1; "+ "end; " + "return nil;", Arrays.<Object>asList(getName(), getChannelName()), LockPubSub.unlockMessage, internalLockLeaseTime, getLockName(threadId)); }如上代码，就是释放锁的逻辑。同样的，它也是有三个判断：
如果锁已经不存在，通过publish发布锁释放的消息，解锁成功
如果解锁的线程和当前锁的线程不是同一个，解锁失败，抛出异常
通过hincrby递减1，先释放一次锁。若剩余次数还大于0，则证明当前锁是重入锁，刷新过期时间；若剩余次数小于0，删除key并发布锁释放的消息，解锁成功
至此，Redisson中的可重入锁的逻辑，就分析完了。但值得注意的是，上面的两种实现方式都是针对单机Redis实例而进行的。如果我们有多个Redis实例，请参阅Redlock算法。该算法的具体内容，请参考http://redis.cn/topics/distlock.html
前台请求，后台先从缓存中取数据，取到直接返回结果，取不到时从数据库中取，数据库取到更新缓存，并返回结果，数据库也没取到，那直接返回空结果。
描述：
缓存穿透是指缓存和数据库中都没有的数据，而用户不断发起请求，如发起为id为“-1”的数据或id为特别大不存在的数据。这时的用户很可能是攻击者，攻击会导致数据库压力过大。
解决方案：
接口层增加校验，如用户鉴权校验，id做基础校验，id<=0的直接拦截； 从缓存取不到的数据，在数据库中也没有取到，这时也可以将key-value对写为key-null，缓存有效时间可以设置短点，如30秒（设置太长会导致正常情况也没法使用）。这样可以防止攻击用户反复用同一个id暴力攻击
描述：
缓存击穿是指缓存中没有但数据库中有的数据（一般是缓存时间到期），这时由于并发用户特别多，同时读缓存没读到数据，又同时去数据库去取数据，引起数据库压力瞬间增大，造成过大压力
解决方案：
设置热点数据永远不过期。 加互斥锁，互斥锁参考代码如下：
说明：
1）缓存中有数据，直接走上述代码13行后就返回结果了
2）缓存中没有数据，第1个进入的线程，获取锁并从数据库去取数据，没释放锁之前，其他并行进入的线程会等待100ms，再重新去缓存取数据。这样就防止都去数据库重复取数据，重复往缓存中更新数据情况出现。
3）当然这是简化处理，理论上如果能根据key值加锁就更好了，就是线程A从数据库取key1的数据并不妨碍线程B取key2的数据，上面代码明显做不到这点。
描述：
缓存雪崩是指缓存中数据大批量到过期时间，而查询数据量巨大，引起数据库压力过大甚至down机。和缓存击穿不同的是， 缓存击穿指并发查同一条数据，缓存雪崩是不同数据都过期了，很多数据都查不到从而查数据库。
解决方案：缓存数据的过期时间设置随机，防止同一时间大量数据过期现象发生。 如果缓存数据库是分布式部署，将热点数据均匀分布在不同搞得缓存数据库中。 设置热点数据永远不过期。
缓存由于其高并发和高性能的特性，已经在项目中被广泛使用。一般读缓存流程如下：
缓存更新的争议
主要是解决并发情况下缓存读与写的一致性和时效性 一 缓存不能读到脏数据 二 缓存可能会读到过期数据，但要在可容忍时间内实现最总一致 三 这个可容忍时间尽可能的小
基本核心套路（当然有很多变种，但主要思想如此）：
先更新数据库，再更新缓存先删除缓存，再更新数据库先更新数据库，再删除缓存为什么没有先更新缓存，再更新数据库这种策略（ram、rom）问题：同时有请求A和请求B进行更新操作，可能出现 （1）线程A更新了数据库 key = 1 value = 2 （2）线程B更新了数据库 key = 1 value = 3 （3）线程B更新了缓存 key = 1 value = 3 （4）线程A更新了缓存 key = 1 value = 2
这就出现请求A更新缓存应该比请求B更新缓存早才对，但是因为网络等原因，B却比A更早更新了缓存。这就导致了脏数据
方法：
更新缓存数据时判是否最新（如增加版本号）只有比缓存中的新才可更新，否则不管（版本号大于当前版本号才可更新）问题：同时有一个请求A进行更新操作，一个请求B进行查询操作。可能出现: （1）请求A进行写操作（key = 1 value = 2），先删除缓存 key = 1 value = 1 （2）请求B查询发现缓存不存在 （3）请求B去数据库查询得到旧值 key = 1 value = 1 （4）请求B将旧值写入缓存 key = 1 value = 1 （5）请求A将新值写入数据库key = 1 value = 2
如果不采用给缓存设置过期时间策略，该数据永远都是脏数据
方法：
先删除缓存，再更新数据库，再回写缓存（出现套路一的情况）先删除缓存，再更新数据库，再删缓存（双删，第二次删可异步）数据库读写分离，本地缓存都不再适用，数据一致性需要用到远程数据同步
有两篇参考：《Cache-Aside pattern》、《Scaling Memcache at Facebook》 主要的思路就是先更新数据库，再删缓存的策略。
问题：一个请求A做查询操作，一个请求B做更新操作，可能情况
（1）缓存刚好失效 （2）请求A查询数据库，得一个旧值 （3）请求B将新值写入数据库 （4）请求B删除缓存 （5）请求A将查到的旧值写入缓存
如果发生上述情况，确实是会发生脏数据。然而，发生这种情况的概率又有多少呢？ 我们知道一般来说读比写快很多，这一情形很难出现。
二、三套路的共性问题：删缓存失败了怎么办？
方案一
方案缺点：
引入中间件，增加复杂度对业务线代码造成大量的侵入方案二
启动一个订阅程序去订阅数据库的binlog，获得需要操作的数据。在应用程序中，另起一段程序，获得这个订阅程序传来的信息，进行删除缓存操作。
mysql中有现成的中间件叫canal，可以完成订阅binlog日志的功能。 另我团的DataBus更强大。
作者：jiangmo 链接：https://www.jianshu.com/p/0275ecca2438 来源：简书 著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
参考文献：
https://www.jianshu.com/p/47fd7f86c848
https://blog.csdn.net/kongtiao5/article/details/82771694
https://www.jianshu.com/p/0275ecca2438
