文章目录
一、redis高并发分布式锁实战
日常开发中,秒杀下单等业务,都需要用到分布式锁。而Redis非常适合作为分布式锁使用。
1、什么是分布式锁
分布式锁其实就是,控制分布式系统不同进程共同访问共享资源的一种锁的实现。如果不同的系统或同一个系统的不同主机之间共享了某个临界资源,往往需要互斥来防止彼此干扰,以保证一致性。
我们先来看下,一把靠谱的分布式锁应该有哪些特征:
2、实战
接下来,我们以秒杀减库存场景模拟redis分布式锁的使用,逐步完善锁的使用。
使用分布式的前提:localKey的颗粒度尽量小,降低锁竞争造成的影响。
例如,下图中的秒杀场景如何防止超卖咧?
SpringBoot搭建简单的demo;
依赖:
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-pool2</artifactId>
</dependency>
配置application.yml
spring:
redis:
database: 0
timeout: 3000
host: 192.168.1.188
port: 6379
lettuce:
pool:
MaxTotal: 50 #最大连接数
minIdle: 1
maxWaitMillis: 5000
maxIdle: 5
testOnBorrow: true
testOnReturn: true
testWhileIdle: true
2.1 方案一:SETNX + EXPIRE
问题:非原子性,可能造成死锁;
setnx加锁和EXPIRE给锁设置时间两个命令分开了,不是原子操作;如果执行完setnx加锁,正要执行expire设置过期时间时,进程crash或者要重启维护了,那么就造成了死锁;
demo:
@Autowired
private StringRedisTemplate stringRedisTemplate;
@RequestMapping("/deduct/stock")
public String deductStock(String productId) {
String lockKey = "lock:product_"+productId;
//非原子性操作,容易造成死锁
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"test");
stringRedisTemplate.expire(lockKey,10,TimeUnit.SECONDS);
if(!result){
return "error code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功,剩余库存:" + realStock);
} else {
System.out.println("扣减失败,库存不足");
}
} finally {
stringRedisTemplate.delete(lockKey);
}
return "end";
}
2.2 方案二:SET的扩展命令(SET EX PX NX)
该方案操作为原子性操作,解决了死锁问题
问题:锁过期释放了,业务还没执行完,误删其他线程的锁;发生超卖
图解:
假设线程1执行完后,去释放锁。但是它不知道当前的锁可能是线程2持有的(线程1去释放锁时,有可能过期时间已经到了,此时线程2进来占有了锁)。那线程1就把线程2的锁释放掉了,但是线程2临界区业务代码可能都还没执行完呢。
demo:
@RequestMapping("/deduct/stock")
public String deductStock(String productId) {
String lockKey = "lock:product_"+productId;
//非原子性操作,容易造成死锁
// Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"test");
// stringRedisTemplate.expire(lockKey,10,TimeUnit.SECONDS);
//SET的扩展命令(SET EX PX NX) 原子性操作
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "test", 30, TimeUnit.SECONDS); //jedis.setnx(k,v)
if(!result){
return "error code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功,剩余库存:" + realStock);
} else {
System.out.println("扣减失败,库存不足");
}
} finally {
stringRedisTemplate.delete(lockKey);
}
return "end";
}
2.3 方案三:SET EX PX NX + 校验唯一随机值,再删除
既然锁可能被别的线程误删,那我们给value值设置一个标记当前线程唯一的随机数,在删除的时候,校验一下判断是不是当前线程加的锁,不就OK了嘛。
问题:存在锁过期释放,业务没执行完「
实际就是当线程未执行完前,锁的时间到了,不能给锁续命的问题
判断是不是当前线程加的锁」和「释放锁」不是一个原子操作;跟方案二一样,在极致情况下,线程一刚进入判断时,出行卡顿,然后此时锁超时,线程二获取锁对象,然后执行任务,此时线程一又恢复执行,那么就把线程二的锁给删除了。
demo:
@RequestMapping("/deduct/stock")
public String deductStock(String productId) {
String lockKey = "lock:product_"+productId;
//非原子性操作,容易造成死锁
// Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey,"test");
// stringRedisTemplate.expire(lockKey,10,TimeUnit.SECONDS);
String clientId = UUID.randomUUID().toString();
//SET的扩展命令(SET EX PX NX) 原子性操作
Boolean result = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, clientId, 30, TimeUnit.SECONDS); //jedis.setnx(k,v)
if(!result){
return "error code";
}
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功,剩余库存:" + realStock);
} else {
System.out.println("扣减失败,库存不足");
}
} finally {
//判断是不是当前线程加的锁,是才释放
if (clientId.equals(stringRedisTemplate.opsForValue().get(lockKey))) {
stringRedisTemplate.delete(lockKey);
}
}
return "end";
}
2.4 方案四:Redisson框架
问题:单机问题
已解决锁的原子性和锁过期的问题
有些小伙伴认为,稍微把锁过期时间设置长一些就可以啦。其实我们设想一下,是否可以给获得锁的线程,开启一个定时守护线程,每隔一段时间检查锁是否还存在,存在则对锁的过期时间延长,防止锁过期提前释放。
当前开源框架Redisson解决了这个问题。我们一起来看下Redisson底层原理图吧:
只要线程一加锁成功,就会启动一个watch dog看门狗,它是一个后台线程,会每隔10秒检查一下,如果线程1还持有锁,那么就会不断的延长锁key的生存时间。因此,Redisson就是使用Redisson解决了「锁过期释放,业务没执行完」问题。
demo:
依赖:
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>
配置项:
@Bean
public Redisson redisson() {
// 此为单机模式
Config config = new Config();
config.useSingleServer().setAddress("redis://192.168.1.188:6379").setDatabase(0);
return (Redisson) Redisson.create(config);
}
@Autowired
private Redisson redisson;
@RequestMapping("/deduct/stock")
public String deductStock(String productId) {
String lockKey = "lock:product_"+productId;
//获取锁对象
RLock redissonLock = redisson.getLock(lockKey);
//加分布式锁
redissonLock.lock();
try {
int stock = Integer.parseInt(stringRedisTemplate.opsForValue().get("stock")); // jedis.get("stock")
if (stock > 0) {
int realStock = stock - 1;
stringRedisTemplate.opsForValue().set("stock", realStock + ""); // jedis.set(key,value)
System.out.println("扣减成功,剩余库存:" + realStock);
} else {
System.out.println("扣减失败,库存不足");
}
} finally {
redissonLock.unlock();
}
return "end";
}
2.5 方案五:Redisson 红锁和联锁
前面四种方案都只是基于单机版的讨论,还不是很完美。其实Redis一般都是集群部署的:
如果线程一在Redis的master节点上拿到了锁,但是加锁的key还没同步到slave节点。恰好这时,master节点发生故障,一个slave节点就会升级为master节点。线程二就可以获取同个key的锁啦,但线程一也已经拿到锁了,锁的安全性就没了。
为了解决这个问题,Redis作者 antirez提出一种高级的分布式锁算法:Redlock。Redlock核心思想是这样的:
搞多个Redis master部署,以保证它们不会同时宕掉。并且这些master节点是完全相互独立的,相互之间不存在数据同步。同时,需要确保在这多个master实例上,是与在Redis单实例,使用相同方法来获取和释放锁。
具体查看网页
redisson分布式锁
我们假设当前有5个Redis master节点,在5台服务器上面运行这些Redis实例。
简化下步骤就是:
- 按顺序向5个master节点请求加锁
- 根据设置的超时时间来判断,是不是要跳过该master节点。
- 如果大于等于3个节点加锁成功,并且使用的时间小于锁的有效期,即可认定加锁成功啦。
- 如果获取锁失败,解锁!
另一种强一致性的是联锁,一般不会用这个:
3、锁优化方案
锁的粒度问题:锁的粒度越少,影响范围就越少,性能就越高;
例如秒杀商品有1000个,为了提高秒杀效率,可以使用分段锁的策略,把1000个商品分成10份100个商品的key竞争,这样效率理论上就提升了10倍。当然要考虑每次秒杀减少的是哪份库存,以及库存是否购买和库存清零的问题。
二、缓存几大问题
1、缓存粒度控制
通俗来讲,缓存粒度问题就是我们在使用缓存时,是将所有数据缓存还是缓存部分数据?
缓存粒度问题是一个容易被忽视的问题,如果使用不当,可能会造成很多无用空间的浪费,可能会造成网络带宽的浪费,可能会造成代码通用性较差等情况,必须学会综合数据通用性、空间占用比、代码维护性 三点评估取舍因素权衡使用。
2、缓存穿透
定义:缓存穿透是指查询一个根本不存在的数据, 缓存层和存储层都不会命中, 通常出于容错的考虑, 如果从存储层查不到数据则不写入缓存层。
结果:缓存穿透将导致不存在的数据每次请求都要到存储层去查询, 失去了缓存保护后端存储的意义。
造成缓存穿透的基本原因有两个:
第一, 自身业务代码或者数据出现问题。
第二, 一些恶意攻击、 爬虫等造成大量空命中。
缓存穿透问题解决方案:
1)缓存空对象
if productid不存在,set key null缓存;
思考:若后期上架了productid一样的产品或者太多的productid不存在,会有什么结果?
Answer:可能会导致后上架的产品一直查不到数据或空对象占用内存过多
方案:缓存空对象时,设置超时时间
String get(String key) {
// 从缓存中获取数据
String cacheValue = cache.get(key);
// 缓存为空
if (StringUtils.isBlank(cacheValue)) {
// 从存储中获取
String storageValue = storage.get(key);
cache.set(key, storageValue);
// 如果存储数据为空, 需要设置一个过期时间(300秒)
if (storageValue == null) {
cache.expire(key, 60 * 5);
}
return storageValue;
} else {
// 缓存非空
return cacheValue;
}
}
2)布隆过滤器
对于恶意攻击,向服务器请求大量不存在的数据造成的缓存穿透,还可以用布隆过滤器先做一次过滤,对于不存在的数据布隆过滤器一般都能够过滤掉,不让请求再往后端发送。
当布隆过滤器说某个值存在时,这个值可能不存在;当它说不存在时,那就肯定不存在
。
布隆过滤器就是一个大型的位数组和几个不一样的无偏 hash 函数。所谓无偏就是能够把元素的 hash 值算得比较均匀。
向布隆过滤器中添加 key 时,会使用多个 hash 函数对 key 进行 hash 算得一个整数索引值然后对位数组长度进行取模运算得到一个位置,每个 hash 函数都会算得一个不同的位置。再把位数组的这几个位置都置为 1 就完成了 add 操作。
向布隆过滤器询问 key 是否存在时,跟 add 一样,也会把 hash 的几个位置都算出来,看看位数组中这几个位置是否都为 1,只要有一个位为 0,那么说明布隆过滤器中这个key 不存在。如果都是 1,这并不能说明这个 key 就一定存在,只是极有可能存在,因为这些位被置为 1 可能是因为其它的 key 存在所致。如果这个位数组长度比较大,存在概率就会很大,如果这个位数组长度比较小,存在概率就会降低。
这种方法适用于数据命中不高、 数据相对固定、 实时性低(通常是数据集较大) 的应用场景, 代码维护较为复杂, 但是缓存空间占用很少。
可以用redisson实现布隆过滤器,引入依赖:
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson</artifactId>
<version>3.6.5</version>
</dependency>
import org.redisson.Redisson;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
public class RedissonBloomFilter {
public static void main(String[] args) {
Config config = new Config();
config.useSingleServer().setAddress("redis://localhost:6379");
//构造Redisson
RedissonClient redisson = Redisson.create(config);
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("nameList");
//初始化布隆过滤器:预计元素为100000000L,误差率为3%,根据这两个参数会计算出底层的bit数组大小
bloomFilter.tryInit(100000000L,0.03);
//将zhuge插入到布隆过滤器中
bloomFilter.add("zhuge");
//判断下面号码是否在布隆过滤器中
System.out.println(bloomFilter.contains("guojia"));//false
System.out.println(bloomFilter.contains("baiqi"));//false
System.out.println(bloomFilter.contains("zhuge"));//true
}
}
使用布隆过滤器需要把所有数据提前放入布隆过滤器,并且在增加数据时也要往布隆过滤器里放,布隆过滤器缓存过滤伪代码:
//初始化布隆过滤器
RBloomFilter<String> bloomFilter = redisson.getBloomFilter("nameList");
//初始化布隆过滤器:预计元素为100000000L,误差率为3%
bloomFilter.tryInit(100000000L,0.03);
//把所有数据存入布隆过滤器
void init(){
for (String key: keys) {
bloomFilter.put(key);
}
}
String get(String key) {
// 从布隆过滤器这一级缓存判断下key是否存在
Boolean exist = bloomFilter.contains(key);
if(!exist){
return "";
}
// 从缓存中获取数据
String cacheValue = cache.get(key);
// 缓存为空
if (StringUtils.isBlank(cacheValue)) {
// 从存储中获取
String storageValue = storage.get(key);
cache.set(key, storageValue);
// 如果存储数据为空, 需要设置一个过期时间(300秒)
if (storageValue == null) {
cache.expire(key, 60 * 5);
}
return storageValue;
} else {
// 缓存非空
return cacheValue;
}
}
注意:布隆过滤器不能删除数据,如果要删除得重新初始化数据。
3、缓存击穿(缓存失效)
定义:由于大批量缓存在同一时间失效可能导致大量请求同时穿透缓存直达数据库,可能会造成数据库瞬间压力过大甚至挂掉。
结果:会造成数据库瞬间压力过大甚至挂掉。
解决方案:
1、在批量增加缓存时最好将这一批数据的缓存过期时间设置为一个时间段内的不同时间(random)
2、使用分布式锁
示例伪代码:
String get(String key) {
// 从缓存中获取数据
String cacheValue = cache.get(key);
// 缓存为空
if (StringUtils.isBlank(cacheValue)) {
// 从存储中获取
String storageValue = storage.get(key);
cache.set(key, storageValue);
//设置一个过期时间(300到600之间的一个随机数)
int expireTime = new Random().nextInt(300) + 300;
if (storageValue == null) {
cache.expire(key, expireTime);
}
return storageValue;
} else {
// 缓存非空
return cacheValue;
}
}
4、缓存雪崩
定义:是缓存层支撑不住或宕掉后,流量会打向后端存储层(原因:大并发访问,redis设计不合理),请求全部转发到DB,DB瞬时压力过重雪崩。
结果:由于缓存层承载着大量请求, 有效地保护了存储层, 但是如果缓存层由于某些原因不能提供服务(比如超大并发过来,缓存层支撑不住,或者由于缓存设计不好,类似大量请求访问bigkey,导致缓存能支撑的并发急剧下降), 于是大量请求都会打到存储层, 存储层的调用量会暴增, 造成存储层也会级联宕机的情况。
预防和解决缓存雪崩问题, 可以从以下三个方面进行着手。
解决方案:
1) 保证缓存层服务高可用性,比如使用Redis Sentinel或Redis Cluster。
2) 依赖隔离组件为后端限流熔断并降级。比如使用Sentinel或Hystrix限流降级组件。
比如服务降级,我们可以针对不同的数据采取不同的处理方式。当业务应用访问的是非核心数据(例如电商商品属性,用户信息等)时,暂时停止从缓存中查询这些数据,而是直接返回预定义的默认降级信息、空值或是错误提示信息;当业务应用访问的是核心数据(例如电商商品库存)时,仍然允许查询缓存,如果缓存缺失,也可以继续通过数据库读取。
3) 提前演练。 在项目上线前, 演练缓存层宕掉后, 应用以及后端的负载情况以及可能出现的问题, 在此基础上做一些预案设定。
4)做多级缓存,A1为原始缓存,A2为拷贝缓存,A1失效时,可以访问A2,A1缓存失效时间设置为短期,A2设置为长期
5、热点缓存key重建优化
开发人员使用“缓存+过期时间”的策略既可以加速数据读写, 又保证数据的定期更新, 这种模式基本能够满足绝大部分需求。 但是有两个问题如果同时出现, 可能就会对应用造成致命的危害:
- 当前key是一个热点key(例如一个热门的娱乐新闻),并发量非常大。
- 重建缓存不能在短时间完成, 可能是一个复杂计算, 例如复杂的SQL、 多次IO、 多个依赖等。
在缓存失效的瞬间, 有大量线程来重建缓存, 造成后端负载加大, 甚至可能会让应用崩溃。
要解决这个问题主要就是要避免大量线程同时重建缓存。
我们可以利用互斥锁来解决,此方法只允许一个线程重建缓存, 其他线程等待重建缓存的线程执行完, 重新从缓存获取数据即可。
String get(String key) {
// 从Redis中获取数据
String value = redis.get(key);
// 如果value为空, 则开始重构缓存
if (value == null) {
// 只允许一个线程重建缓存, 使用nx, 并设置过期时间ex
String mutexKey = "mutext:key:" + key;
if (redis.set(mutexKey, "1", "ex 180", "nx")) {
// 从数据源获取数据
value = db.get(key);
// 回写Redis, 并设置过期时间
redis.setex(key, timeout, value);
// 删除key_mutex
redis.delete(mutexKey);
}// 其他线程休息50毫秒后重试
else {
Thread.sleep(50);
get(key);
}
}
return value;
}
这里只是简单示例,可以用上面的分布式锁redisson实现更好
6、缓存与数据库双写不一致问题
定义:在大并发下,同时操作数据库与缓存会存在数据不一致性问题
1、双写不一致情况
2、读写并发不一致
解决方案:
1、对于并发几率很小的数据(如个人维度的订单数据、用户数据等),这种几乎不用考虑这个问题,很少会发生缓存不一致,可以给缓存数据加上过期时间,每隔一段时间触发读的主动更新即可。
2、就算并发很高,如果业务上能容忍短时间的缓存数据不一致(如商品名称,商品分类菜单等),缓存加上过期时间依然可以解决大部分业务对于缓存的要求。
3、如果不能容忍缓存数据不一致,可以通过加分布式读写锁保证并发读写或写写的时候按顺序排好队,读读的时候相当于无锁。
4、也可以用阿里开源的canal通过监听数据库的binlog日志及时的去修改缓存,但是引入了新的中间件,增加了系统的复杂度。
@Transactional
public Product update(Product product) {
Product productResult = null;
//RLock productUpdateLock = redisson.getLock(LOCK_PRODUCT_UPDATE_PREFIX + product.getId());
RReadWriteLock productUpdateLock = redisson.getReadWriteLock(LOCK_PRODUCT_UPDATE_PREFIX + product.getId());
RLock writeLock = productUpdateLock.writeLock();
//加分布式写锁解决缓存双写不一致问题
writeLock.lock();
try {
productResult = productDao.update(product);
redisUtil.set(RedisKeyPrefixConst.PRODUCT_CACHE + productResult.getId(), JSON.toJSONString(productResult),
genProductCacheTimeout(), TimeUnit.SECONDS);
} finally {
writeLock.unlock();
}
return productResult;
}
public Product get(Long productId) {
Product product = null;
String productCacheKey = RedisKeyPrefixConst.PRODUCT_CACHE + productId;
//从缓存里查数据
product = getProductFromCache(productCacheKey);
if (product != null) {
return product;
}
//加分布式锁解决热点缓存并发重建问题
RLock hotCreateCacheLock = redisson.getLock(LOCK_PRODUCT_HOT_CACHE_CREATE_PREFIX + productId);
hotCreateCacheLock.lock();
// 这个优化谨慎使用,防止超时导致的大规模并发重建问题
// hotCreateCacheLock.tryLock(1, TimeUnit.SECONDS);
try {
//查询数据的逻辑
product = getProductFromCache(productCacheKey);
if (product != null) {
return product;
}
//RLock productUpdateLock = redisson.getLock(LOCK_PRODUCT_UPDATE_PREFIX + productId);
RReadWriteLock productUpdateLock = redisson.getReadWriteLock(LOCK_PRODUCT_UPDATE_PREFIX + productId);
RLock rLock = productUpdateLock.readLock();
//加分布式读锁解决缓存双写不一致问题
rLock.lock();
try {
product = productDao.get(productId);
if (product != null) {
redisUtil.set(productCacheKey, JSON.toJSONString(product),
genProductCacheTimeout(), TimeUnit.SECONDS);
} else {
//设置空缓存解决缓存穿透问题
redisUtil.set(productCacheKey, EMPTY_CACHE, genEmptyCacheTimeout(), TimeUnit.SECONDS);
}
} finally {
rLock.unlock();
}
} finally {
hotCreateCacheLock.unlock();
}
return product;
}
总结:
以上我们针对的都是读多写少的情况加入缓存提高性能,如果写多读多的情况又不能容忍缓存数据不一致,那就没必要加缓存了,可以直接操作数据库。当然,如果数据库抗不住压力,还可以把缓存作为数据读写的主存储,异步将数据同步到数据库,数据库只是作为数据的备份。
放入缓存的数据应该是对实时性、一致性要求不是很高的数据。切记不要为了用缓存,同时又要保证绝对的一致性做大量的过度设计和控制,增加系统复杂性!
以上信息来我在
图灵学院
学习的总结,图灵学院很不错,值得报班