- 1 抢红包排行查询
- 2 击穿现象分析
- 3 击穿解决方案
- 3.1 定时器
- 3.2 多级缓存
- 3.3 分布式锁
- 3.4 队列术
1 抢红包排行查询
上面案例我们实现了某个用户抢红包的信息查询，接下来我们实现公示抢到红包并且按照红包金额大小排序查询出前
100名用户信息，这块数据并发量将更大，我们需要做缓存处理。
Controller
Service 这里做了缓存 *** 作，缓存1分钟，1分钟过后，会再次查询数据库
测试结果如下：
我们先来了解下缓存击穿，缓存在某个时间点过期的时候，恰好在这个时间点对这个Key有大量的并发请求过来，这些请求发现缓存过期一般都会从后端DB加载数据并回设到缓存，这个时候大并发的请求可能会瞬间把后端DB压垮。
上面查询红包排名就存在击穿现象，比如10万用户请求，此时缓存刚好过期，10万用户同时到达了第②个步骤，而且此时Redis中都判断没有数据，那么此时就都查询数据库，给数据库带来巨大的压力，甚至是宕机。
针对缓存击穿现象，可以有多重解决方案。我们这里给大家讲解一下实用的5种方案。
3.1 定时器后台定义一个job(定时任务)专门主动更新缓存数据.比如,一个缓存中的数据过期时间是1分钟,那么job每隔25秒刷新数据(将从数据库中查到的数据更新到缓存中)，或者缓存不过期，直接写定时任务定时更新即可。定时器需要择优选择，比如可以用 elastic-job , xxl-job 。
这种方案比较容易理解，但会增加系统复杂度。比较适合那些 key 相对固定,cache 粒度较大的业务，key 比较分散的则不太适合，实现起来也比较复杂。
采用多级缓存也可以有效防止击穿现象，首先通过程序将缓存存入到Redis缓存，且永不过期，用户查询的时候，先查询Nginx缓存，如果Nginx缓存没有，则查询Redis缓存，并将Redis缓存存入到Nginx一级缓存中，并设置更新时间。这种方案不仅可以提升查询速度，同时又能防止击穿问题，并且提升了程序的抗压能力。
解决上面超卖问题，我们可以采用分布式锁来控制，分布式锁的原理很简单。
分布式锁主要是实现在分布式场景下保证数据的最终一致性。在单进程的系统中，存在多个线程可以同时改变某个变量（可变共享变量）时，就需要对变量或代码块做同步(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的工厂
- 引入依赖
<dependency>
<groupId>org.redissongroupId>
<artifactId>redisson-spring-boot-starterartifactId>
<version>3.11.0version>
dependency>
- 锁 *** 作方法实现
要想用到分布式锁，我们就必须要实现获取锁和释放锁，获取锁和释放锁可以编写一个 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();
}
}
- 配置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);
}
- 分布式锁实现
上面我们已经使用过队列术了，队列术在面对零点洪峰流量时，是相当有效，可以直接将所有流量存入到队列中，让后台不用同时处理很多请求，而是从队列中逐个消费逐个处理，上图是实现流程，由于前面抢单已经实现过该流程，所以这里不再重复讲解。
基于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属性。
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）