在如今这个微服务分布式的大环境下,集群分布式部署几乎已经是我们每个人都熟知的了。
缓存也一样，对我们来说，如果只是一个单体应用，那只要有本地缓存就足以了，但是倘若分布式部署了很多台机器上，那我们该如何缓存呢，如果依然用本地缓存，那我们不同机器之间的缓存数据该如何同步呢。这样就需要我们的分布式缓存出场了。下面我将分别介绍本地缓存和分布式缓存。

一、本地缓存

我们通常可以在类中声明一个静态的Map类型变量，以键值对的方式存放我们的数据，需要数据时，直接从map中拿即可。这就是本地缓存。在单体应用下本地缓存毫无疑问是最简单最方便的。但是随着业务的不断扩大，我们的服务仅在一台机器上部署远远不够支撑流量了，于是我们就需要分布式部署，利用负载均衡的手段让请求分散到不同的机器上。这样就会出现下面的两个问题

1、倘若我们第一次请求到第一台机器，本地缓存的map中没有数据，回进行一次查库操作。将数据存入map中。但是第二次请求到了第二台机器，第二台机器的map中依然没值。这样就需要再次查库。

2、倘若是一个修改请求，第一次请求到第一台机器。修改了数据库，同时将本地缓存数据修改了，但是此时一个查询请求到了第二台机器，由于修改的请求是在第一台机器触发的，因此也只有第一台机器的缓存是和数据库一致的，此时读第二台请求，相当于读的还是旧数据。这样就造成了数据一致性的问题。

基于以上的分析，我们发现本地缓存已经无法满足我们分布式下的系统缓存需要了。那么我们就需要引出分布式缓存了。常见的分布式缓存就是我们最熟悉的redis了。分布式缓存的原理无非就是将缓存的数据统一存到一个地方，让所有机器去这一个地方读取或修改数据即可。同时redis的集群也可以从某种意义上帮助我们实现一台机器内存不够的情况。

二、redis分布式缓存

redis缓存虽然解决了分布式情况下本地缓存带来的一些问题，但是也同时会出现一些新的问题,下面我们就来分析一下这些问题吧。

谈到redis缓存，我们大家都很熟悉的高并发下可能会导致redis缓存失效的三类问题，下面就让我们一起来分析一下吧~

缓存穿透

导致缓存穿透的情况就是，当我们访问一个缓存中不存在的数据，就会去查库，这样就会导致缓存未命中疯狂扫射数据库的情况。这样的话我们的缓存也就失去了意义。说的再通俗一点就是，当我们访问一个永远不存在的数据，比方我们查询数据库中商品id为-1的数据，这样的数据在数据库始终就没有存储，缓存中也没有此数据，在这样的情况下，查询缓存未命中，我们就会去查询数据库，但是当查数据库时，查到的数据为null，但是并没有将null值写入redis缓存，因此我们下次查询此数据，依然缓存未命中，这样就会导致查询始终都是访问数据库，系统就会出现被利用不存在的数据进行攻击的风险。导致服务挂掉。

解决方案就是查数据库查询到null，我们也写入redis缓存中，这样即便查不存在的数据，缓存也依然命中。

缓存雪崩

缓存雪崩的意思就像时雪山崩塌，一瞬间大量的redis缓存key过期，并且还是在高并发流量的情况下过期，这样就会大量并发查询请求落到数据库上，导致数据库压力过大直接奔溃。

** 解决方案通常我们可以在设置缓存过期时间时加一个随机值，这样就可以大大降低缓存时间都相同的概率，通常就很难引发缓存集体失效的情况了。**

缓存击穿

缓存击穿的情况就像是射击一样，盯住一个一直设计，突然这个被放开了，设计就穿过去了。
描述的是，当我们的一个热点key在某一时刻失效，刚好此时大并发同时在访问此key，这样的情况下就会让大量并发在key失效的一瞬间同时去访问数据库。导致数据库奔溃。

对于此问题我们通常的解决方式是加锁，我们在访问数据库的时候上锁。只让一个请求可以到达数据库。其他请求都等待，当到达数据库的请求获取到数据后就可以写入缓存了，写入缓存后，释放锁，其他请求此时排队获取到锁后进入方法，在查数据库之前再去先判断一下缓存是否有数据，发现缓存有数据，直接返回，也就不会落到数据库上了。

再解决缓存击穿时我们提到了加锁，我们可能第一时间想到的就是 synchronized 关键字。或者 Lock。但是Juc（lock）或者synchronized 他们都是本地锁，也就是说只能锁本地服务，当有多台服务器部署服务时，负载均很后的情况通常来说就是，每台服务器都会允许一个请求去查库。但是我们是否可以锁所有呢？让多台服务器也只有一个请求可以去查库。那就提到了我们的分布式锁。

下面就让我们来谈谈分布式锁的实现方式吧。

分布式锁

首先我们先来自己实现一下分布式锁。我们想一下本地锁是怎么实现的，本地锁最核心的就是那个同步监视器，我们要求同步监视器（也就是锁）必须是同一个，这样的话，我们就可以将资源锁住。现在分布式的情况下，问题就是多台服务器之间无法通信，相当于每台服务器都有一个自己的锁，我们想实现分布式锁，那我们就可以将这个锁同分布式缓存一样，统一放在一个地方，让所有的服务都去这一个地方去获取锁。这样就相当于所有的服务器拿到的都是同一把锁。那么如何实现呢？

在redis中有这样一个命令

set key value NX

这里的这条命令NX的意思就是如果缓存中有这个key ，那就set失败，如果缓存中没有key就set该key到缓存。而这里，能够set成功的那个就相当于是抢到锁的。

下面我们来看看我们的一步步实现吧。

上图这样基本上我们可以说是实现了分布式锁，但是我们需要分析一下上面的实现是否有一些潜在问题。

在上面的这种实现中我们时候存在死锁的可能，假设我们执行完了业务，但是还没来得及删除锁服务宕机了，或者是在执行业务的过程中出了异常。就会导致我们的锁永远都无法被删除了。为了防止这样产生的死锁，我们就需要给锁加一个时效，就是加一个过期时间，到了指定的时间，不管怎样都将锁删除，这样就可以解决死锁的情况了。

改进成了这样。

我们虽然加了过期时间，但是这样的情况下还是有可能导致死锁，为什么呢，我们来分析一下。
假设我们执行到获取锁，服务宕机了，那我们的过期时间相当于还是没设置上，依旧会造成死锁，因此我们必须保证设置锁和设置过期时间是一个原子操作。在redis中我们有这样一个命令 set key value NX EX。我们再来看看实现。

改进为这样的代码，就保证了设置锁和设置过期时间的原子性。

但是这样的代码是否还存在问题呢？我们的锁是有了过期时间了。但是我们删除锁的时候，我们可以直接删除吗？我们试想一种场景：倘若我们业务执行了50s ，但是我们设置的锁过期时间是30s，在我们业务执行到30s的时候，锁就已经被删除了，那么此时其他的线程就可以获取这个锁了，当其他线程拿到这个锁之后，业务执行到50s，结束了，此时这个线程就要执行删除锁的操作了，但是这个线程删除锁直接删除了就相当于把别人的锁给删除掉了。这显然是不对的。那我们该怎么解决这件事呢？我们就需要加一个限制，让每个线程只能删除他自己的锁。怎么限制呢？我们的redis存储都是存的key-vaule键值对呀，我们现在相当于都只是用到了key去保证是一把锁。但是value我们也可以用起来。怎么用呢？
我们在获取锁的时候，指定一个value。有那个随机UUID指定即可。每个人删除锁的时候先判断一下，是自己的锁，再删除。
那么代码就如下了。

以上操作可以保证每个人只删除自己的锁吗？
答案是无法保证的，为什么还是无法保证呢？让我们一起来分析一下。

针对以上的情况我们该怎么办呢？
我们就需要保证获取锁的值和删除锁这两个操作要是原子操作，那怎么办呢？参照redis官方文档，我们可以看到我们可以使用Lua脚本保证原子性。

这样我们就大工告成了，但是这里还存在一个问题，就是假设我们业务执行时间特别长，在我们业务执行还没有将数据写入缓存，锁已经过期了，那么此时我们的其他线程就可以获取到锁，但是获取到锁后，发现缓存中依然没有，因此就还会查数据库，这样就相当于我们没锁住。

针对上面最后的这种情况，我们只需要把锁的过期时间设置的长一些，只要保证大于所有业务执行时间即可。但是有没有更优雅的方案呢？当然有，那就是锁的自动续期。这里我们就提到了一个分布式锁的中间件，帮助我们做了这些事情，只需要简单的一些配置即可做到我们以上的实现。那就是redisson。让我们一起来看看吧。

redisson分布式锁

redisson中的锁，我们查看源码发现，他实现了Juc包中的Lock，因此我们在并发编程JUC中所学的在redisson中同样可以使用。没有太大的学习成本，下面就让我们一起来学习一下吧。
在使用redisson锁时， lock中传的参数即锁的名字。一定要注意，锁的名字一样就是一把锁。锁的粒度，越细越快。

在redisson中我们只需要调用 lock.lock();即可加锁，这个方法试阻塞式的锁。默认锁时间式30s。

并且redisson中为我们实现了锁的自动续期，也就是说当业务时间超长，锁会自动续期，不会中途失效，导致锁不住的情况。
同时redisson中的锁在业务执行完之后会自动删除，也就是在业务执行完之后就不会再续期了，因此，即使未调用 unlock() 锁最终也会被删除，不会产生死锁的情况。
如果我们传递了锁的超时时间，即我们调用的是 lock(10 ， TimeUnit.SECONDS)这个方法。就执行脚本，发送给redis进行占锁，默认的超时时间就是我们指定的时间。
如果我们未指定锁的超时时间，就使用 30 * 1000；即为看门狗的默认时间lockWatchdogTimeout
未指定超时时间，只要占锁成功，就会启动一个定时任务，重新给锁设置过期时间，新的过期时间就是看门狗的默认时间。1/3 的看门狗时间回自动续期一次。续期为我们的看门狗时间。
.
我们使用redisson改造一下我们上面的代码

只需要调用这一个方法就可以做到我们上面写的那一堆代码的功能，并且还可以实现锁的自动续期，我们一起来探究一下吧。

在源码中我们能看到在这里传的参数是 -1 ，也就是说我们调用lock什么参数都不传的话默认是传一个-1。这个参数是锁的过期时间。我们继续往下跟踪代码。
我们来看这个方法尝试获取锁

进入我们会发现在这里会判断这个过期时间的参数
如果是不等于-1 ，也就是说是我们调用lock时候指定了过期时间就执行4这个方法，值是一个lua脚本。4. 我们进去看发现是一个lua脚本。
5. 若是我们没指定过期时间的话，我们重点来看这个方法 scheduleExpirationRenewal(threadId);

进入之后我们重点来看这个方法renewExpiration();
进入方法后我们发现这里有一个定时任务。去做锁的自动续期。
具体规则我们可以看到是时间的三分之一每次续期。

这就是我们大概的redisson中简单的加锁的源码。我们可以从中发现我们调用lock时指定了过期时间的话，redisson就不会为我们进行锁的自动续期了。

下面我们可以再看看redisson中的其他锁。redisson中也为我们提供了诸如读写锁、闭锁、信号量、等这些锁。
我们一起来探究一下~

读写锁：

写锁：

RReadWriteLock lock = redisson.getReadWriteLock("read-write-lock");
RLock rLock = lock.writeLock();
rLock.lock();

读锁：

RReadWriteLock lock = redisson.getReadWriteLock("read-write-lock");
RLock rLock = lock.readLock();
rLock.lock();

读写锁的需求就是保证业务一定能读到最新的数据。修改期间，写锁是一个排它锁，也就是互斥锁。读锁是一个共享锁。
写锁没释放，读就必须等待。并发读只会在redis中记录好，所有当前的读锁，并且他们都会同时加锁成功。

读 + 读：相当于无锁。
写 + 读：等待写锁释放才能读。
写 + 写。阻塞方式。
读 + 写。有读锁，写需要等待。
只要有写的存在，都必须等待。

闭锁

就是必须都完成了才能解锁。

获取闭锁： RCountDownLatch door = redisson.getCountDownLatch(“door”);
设置锁的计数次数：door.trySetCount(5);
阻塞等待锁完成：door.await();
锁计数减一： door.countDown();

/**
     * 放假锁门
     * 必须每个班都没人了才锁门。
     * @return
     */
    @ResponseBody
    @GetMapping("/lockDoor")
    public String lockDoor() throws InterruptedException {

        RCountDownLatch door = redisson.getCountDownLatch("door");
        door.trySetCount(5);
        door.await(); // 等待闭锁都完成。

        return "放假了。。。";
    }

    @ResponseBody
    @GetMapping("/gogogo/{id}")
    public String gogogo(@PathVariable("id") Long id) {

        RCountDownLatch door = redisson.getCountDownLatch("door");
        door.countDown(); // 计数减1。

        return id + "班的人都走了";
    }

信号量

车库停车；3车位。

获取信号量：park.acquire();
释放一个信号量： park.release();
以上两个都是阻塞的，也就是说，获取不到的话，会一直阻塞等待。
我们也可以使用 park.tryAcquire() 该方法返回true 或者 false 。表示获取成功或失败。也就不会阻塞了。


    /**
     * 车库停车 三车位。
     * @return
     */
    @ResponseBody
    @GetMapping("/park")
    public String park() throws InterruptedException {

        RSemaphore park = redisson.getSemaphore("park");
        park.acquire(); // 获取一个值。获取一个信号。  阻塞 ， 获取失败了  会一直等，直到获取成功。  占一个车位。

        return "ok";
    }
    @ResponseBody
    @GetMapping("/go")
    public String go() {

        RSemaphore park = redisson.getSemaphore("park");
        park.release(); // 释放一个车位。

        return "ok";
    }

到这里我们就大概了解了redisson实现的分布式锁相关内容了。