1 抢红包排行查询
上面案例我们实现了某个用户抢红包的信息查询，接下来我们实现公示抢到红包并且按照红包金额大小排序查询出前
100名用户信息，这块数据并发量将更大，我们需要做缓存处理。
Controller
Service 这里做了缓存操作，缓存1分钟，1分钟过后，会再次查询数据库
测试结果如下：
2 击穿现象分析
我们先来了解下缓存击穿，缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
上面查询红包排名就存在击穿现象，比如10万用户请求，此时缓存刚好过期，10万用户同时到达了第②个步骤，而且此时Redis中都判断没有数据，那么此时就都查询数据库，给数据库带来巨大的压力，甚至是宕机。
3 击穿解决方案
针对缓存击穿现象，可以有多重解决方案。我们这里给大家讲解一下实用的5种方案。
3.1 定时器
后台定义一个job(定时任务)专门主动更新缓存数据.比如,一个缓存中的数据过期时间是1分钟,那么job每隔25秒刷新数据(将从数据库中查到的数据更新到缓存中)，或者缓存不过期，直接写定时任务定时更新即可。定时器需要择优选择，比如可以用 elastic-job , xxl-job 。
这种方案比较容易理解，但会增加系统复杂度。比较适合那些 key 相对固定,cache 粒度较大的业务，key 比较分散的则不太适合，实现起来也比较复杂。
3.2 多级缓存
采用多级缓存也可以有效防止击穿现象，首先通过程序将缓存存入到Redis缓存，且永不过期，用户查询的时候，先查询Nginx缓存，如果Nginx缓存没有，则查询Redis缓存，并将Redis缓存存入到Nginx一级缓存中，并设置更新时间。这种方案不仅可以提升查询速度，同时又能防止击穿问题，并且提升了程序的抗压能力。
3.3 分布式锁
解决上面超卖问题，我们可以采用分布式锁来控制，分布式锁的原理很简单。
分布式锁主要是实现在分布式场景下保证数据的最终一致性。在单进程的系统中，存在多个线程可以同时改变某个变量（可变共享变量）时，就需要对变量或代码块做同步(lock—synchronized)，使其在修改这种变量时能够线性执行消除并发修改变量。但分布式系统是多部署、多进程的，开发语言提供的并发处理API在此场景下就无能为力了。
目前市面上分布式锁常见的实现方式有三种：
1.基于数据库实现分布式锁；（性能效率比较低，不推荐）
2.基于缓存（Redis等）实现分布式锁； （推荐）
3.基于Zookeeper实现分布式锁；（推荐）
大部分网站使用的分布式锁是基于缓存的，有更好的性能，而缓存一般是以集群方式部署，保证了高可用性。而Redis分布式锁官方推荐使用redisson。
Redission分布式锁说明：
1、redission获取锁释放锁的使用和JDK里面的lock很相似，底层的实现采用了类似lock的处理方式
2、redisson 依赖redis，因此使用redisson 锁需要服务端安装redis，而且redisson 支持单机和集群两种模式下的锁的实现
3、redisson 在多线程或者说是分布式环境下实现机制，其实是通过设置key的方式进行实现，也就是说多个线程为了抢占同一个锁，其实就是争抢设置key。
基于Redisson分布式锁实现
步骤：
1、引入依赖包
2、定义锁的操作方法
a.获取锁的方法
b.解锁
3、创建链接Redis集群的配置文件，在里面做相关配置
4、RedissonClient，实现锁的操作->获取锁和解锁
5、创建工厂对象Redisson的工厂
1.引入依赖
<dependency> <groupId>org.redisson</groupId> <artifactId>redisson-spring-boot-starter</artifactId> <version>3.11.0</version> </dependency>
2.锁操作方法实现
要想用到分布式锁，我们就必须要实现获取锁和释放锁，获取锁和释放锁可以编写一个 DistributedLocker 接口，
代码如下：
public interface DistributedLocker { /*** * lock(), 拿不到lock就不罢休，不然线程就一直block * @param lockKey * @return */ RLock lock(String lockKey); /*** * timeout为加锁时间，单位为秒 * @param lockKey * @param timeout * @return */ RLock lock(String lockKey, long timeout); /*** * timeout为加锁时间，时间单位由unit确定 * @param lockKey * @param unit * @param timeout * @return */ RLock lock(String lockKey, TimeUnit unit, long timeout); /*** * tryLock()，马上返回，拿到lock就返回true，不然返回false。 * 带时间限制的tryLock()，拿不到lock，就等一段时间，超时返回false. * @param lockKey * @param unit * @param waitTime * @param leaseTime 租赁时间，如果超过指定时间还没解锁，就强制解锁 * @return */ boolean tryLock(String lockKey, TimeUnit unit, long waitTime, long leaseTime); /*** * 解锁 * @param lockKey */ void unlock(String lockKey); /*** * 解锁 * @param lock */ void unlock(RLock lock); }
实现上面接口中对应的锁管理方法,编写一个锁管理类 RedissonDistributedLocker ，代码如下：
@Component public class RedissonDistributedLocker implements DistributedLocker { @Autowired private RedissonClient redissonClient; /*** * lock(), 拿不到lock就不罢休，不然线程就一直block * @param lockKey * @return */ @Override public RLock lock(String lockKey) { RLock lock = redissonClient.getLock(lockKey); lock.lock(); return lock; } /*** * timeout为加锁时间，单位为秒 * @param lockKey * @param timeout * @return */ @Override public RLock lock(String lockKey, long timeout) { RLock lock = redissonClient.getLock(lockKey); lock.lock(timeout, TimeUnit.SECONDS); return lock; } /*** * timeout为加锁时间，时间单位由unit确定 * @param lockKey * @param unit * @param timeout * @return */ @Override public RLock lock(String lockKey, TimeUnit unit, long timeout) { RLock lock = redissonClient.getLock(lockKey); lock.lock(timeout, unit); return lock; } /*** * tryLock()，马上返回，拿到lock就返回true，不然返回false。 * 带时间限制的tryLock()，拿不到lock，就等一段时间，超时返回false. * @param lockKey * @param unit * @param waitTime * @param leaseTime * @return */ @Override public boolean tryLock(String lockKey, TimeUnit unit, long waitTime, long leaseTime) { RLock lock = redissonClient.getLock(lockKey); try { return lock.tryLock(waitTime, leaseTime, unit); } catch (InterruptedException e) { e.printStackTrace(); } return false; } /*** * 解锁 * @param lockKey */ @Override public void unlock(String lockKey) { RLock lock = redissonClient.getLock(lockKey); lock.unlock(); } /*** * 解锁 * @param lock */ @Override public void unlock(RLock lock) { lock.unlock(); } }
3.配置Redis链接
在resources下新建文件 redisson.yml ，主要用于配置redis集群节点链接配置，代码如下：
clusterServersConfig: # 连接空闲超时，单位：毫秒 默认10000 idleConnectionTimeout: 10000 pingTimeout: 1000 # 同任何节点建立连接时的等待超时。时间单位是毫秒 默认10000 connectTimeout: 10000 # 等待节点回复命令的时间。该时间从命令发送成功时开始计时。默认3000 timeout: 3000 # 命令失败重试次数 retryAttempts: 3 # 命令重试发送时间间隔，单位：毫秒 retryInterval: 1500 # 重新连接时间间隔，单位：毫秒 reconnectionTimeout: 3000 # 执行失败最大次数 failedAttempts: 3 # 密码 #password: test1234 # 单个连接最大订阅数量 subscriptionsPerConnection: 5 clientName: null # loadBalancer 负载均衡算法类的选择 loadBalancer: !<org.redisson.connection.balancer.RoundRobinLoadBalancer> {} #从节点发布和订阅连接的最小空闲连接数 slaveSubscriptionConnectionMinimumIdleSize: 1 #从节点发布和订阅连接池大小 默认值50 slaveSubscriptionConnectionPoolSize: 50 # 从节点最小空闲连接数 默认值32 slaveConnectionMinimumIdleSize: 32 # 从节点连接池大小 默认64 slaveConnectionPoolSize: 64 # 主节点最小空闲连接数 默认32 masterConnectionMinimumIdleSize: 32 # 主节点连接池大小 默认64 masterConnectionPoolSize: 64 # 订阅操作的负载均衡模式 subscriptionMode: SLAVE # 只在从服务器读取 readMode: SLAVE # 集群地址 nodeAddresses: - "redis://192.168.211.141:7001" - "redis://192.168.211.141:7002" - "redis://192.168.211.141:7003" - "redis://192.168.211.141:7004" - "redis://192.168.211.141:7005" - "redis://192.168.211.141:7006" # 对Redis集群节点状态扫描的时间间隔。单位是毫秒。默认1000 scanInterval: 1000 #这个线程池数量被所有RTopic对象监听器，RRemoteService调用者和RExecutorService任务共同共享。默认2 threads: 0 #这个线程池数量是在一个Redisson实例内，被其创建的所有分布式数据类型和服务，以及底层客户端所一同共享的线程池里 保存的线程数量。默认2 nettyThreads: 0 # 编码方式 默认org.redisson.codec.JsonJacksonCodec codec: !<org.redisson.codec.JsonJacksonCodec> {} #传输模式 transportMode: NIO # 分布式锁自动过期时间，防止死锁，默认30000 lockWatchdogTimeout: 30000 # 通过该参数来修改是否按订阅发布消息的接收顺序出来消息，如果选否将对消息实行并行处理，该参数只适用于订阅发布消 息的情况, 默认true keepPubSubOrder: true # 用来指定高性能引擎的行为。由于该变量值的选用与使用场景息息相关（NORMAL除外）我们建议对每个参数值都进行尝试。 # #该参数仅限于Redisson PRO版本。 #performanceMode: HIGHER_THROUGHPUT
4)创建Redisson管理对象
Redisson管理对象有2个，分别为 RedissonClient 和 RedissonConnectionFactory ，我们只用在项目的
RedisConfig 中配置一下这2个对象即可，在 RedisConfig 中添加的代码如下：
/**** * Redisson客户端 * @return * @throws IOException */ @Bean public RedissonClient redisson() throws IOException { ClassPathResource resource = new ClassPathResource("redisson.yml"); Config config = Config.fromYAML(resource.getInputStream()); RedissonClient redisson = Redisson.create(config); return redisson; } /*** * Redisson工厂对象 * @param redisson * @return */ @Bean public RedissonConnectionFactory redissonConnectionFactory(RedissonClient redisson) { return new RedissonConnectionFactory(redisson); }
5.分布式锁实现
3.4 队列术
上面我们已经使用过队列术了，队列术在面对零点洪峰流量时，是相当有效，可以直接将所有流量存入到队列中，让后台不用同时处理很多请求，而是从队列中逐个消费逐个处理，上图是实现流程，由于前面抢单已经实现过该流程，所以这里不再重复讲解。
基于Nginx缓存队列术
针对一些特定操作，如果请求并发量极高，我们可以采用Nginx自身的队列术，在上一章我们已经学过了Nginx的代理缓存，其中有一个属性叫 proxy_cache_lock ，该属性的意思是：当多个客户端请求一个缓存中不存在的文件（或称之为一个MISS），只有这些请求中的第一个被允许发送至服务器。其他请求在第一个请求得到满意结果之后在缓存中得到文件。如果不启用 proxy_cache_lock ，则所有在缓存中找不到文件的请求都会直接与服务器通信。
proxy_cache_lock 的作用其实和队列术及其类似，只不过发生的地方以及处理的语言不同而已。
我们正好可以使用代理缓存来处理一些查询量大的相同数据，例如查询抢红包Top100就可以用Nginx的代理缓存中
proxy_cache_lock 来实现。
后台代码移除分布式锁：
此时只有第1次从后台获取数据，当然这里会请求3次后才会从缓存拿数据，因为有个属性 proxy_cache_min_use 3属性。
4 缓存雪崩介绍
缓存雪崩是指，由于缓存层承载着大量请求，有效的保护了存储层，但是如果缓存层由于某些原因整体不能提供服务，于是所有的请求都会达到存储层，存储层的调用量会暴增，造成存储层也会挂掉的情况。
解决方案
1)缓存高可用即使个别节点、个别机器、甚至是机房宕掉，依然可以提供服务，比如 Redis Sentinel 和 Redis Cluster 都实现了高可用。
2)限流
微服务网关或者Nginx做好限流操作，防止大量请求直接进入后端，使后端载荷过重最后宕机。
3)数据预热
预先去更新缓存，再即将发生大并发访问前手动触发加载缓存不同的key，设置不同的过期时间，让缓存失效的时间点尽量均匀，不要同时失效。
4)队列术限流
使用Nginx队列或者MQ队列，缓存用户的请求，让所有相同操作只有1次查询数据库，并将查询的数据加入到缓存中，下次查询直接从缓存中获取数据。
5)加锁
数据操作，如果是带有缓存查询的，均使用分布式锁，防止大量请求直接操作数据库。
6)多级缓存（推荐）
采用多级缓存，Nginx+Redis+MyBatis二级缓存，当Nginx缓存失效时，查找Redis缓存，Redis缓存失效查找MyBatis二级缓存。
