关于高并发下缓存失效的问题(本地锁 && 分布式锁 && Redission 详解)

本文涉及的产品
云数据库 Tair(兼容Redis),内存型 2GB
Redis 开源版,标准版 2GB
推荐场景:
搭建游戏排行榜
简介: 关于高并发下缓存失效的问题(本地锁 && 分布式锁 && Redission 详解)


问题引入

缓存穿透

缓存穿透是指大并发下突然访问一个不存在的数据,导致一直去查询数据库,数据库瞬时压力增大,最终导致崩溃。

解决方法:设置具有过期时间的null数据

缓存雪崩

缓存雪崩是指我们会给缓存中的key value设置过期时间,假如100w条数据的过期时间是一样的,当数据过期的一瞬间突然100w的并发来,这时候数据库会崩溃。

解决方法:为不同的数据设置随机的过期时间,让他们不至于同时失效

缓存击穿

缓存击穿是指某一个热点数据在失效的一瞬间进来了高并发,所有对这个key的数据查询都落到了db上,我们称之为缓存击穿

解决方法:大量并发让一个人查,其他人等待,查到以后释放锁,其他人获取锁查看缓存有没有数据,没有再去db查

解决方案

本地锁

本地锁只能锁住this也就是当前实例对象,如果是单体应用不采用分布式的情况下是可以的。因为如果一个采用集群的方式部署,每个节点都有一把锁,并发进来的时候几把锁就可以就放进来了几个进程。

使用本地锁要注意在锁中就要进行数据的缓存,不然的话当锁释放掉,数据还没来得及缓存的时候,其他线程进来发现缓存中还是没有数据,就又会去查询数据库,造成缓存击穿的问题。

public Map<String, List<Catelog2VO>> getCatelogJsonFromDB() {
        synchronized (this){
            //得到锁以后我们应该再去缓存中查看一次,如果没有才继续确定查询
            String catelogJson = stringRedisTemplate.opsForValue().get("catelogJson");
            if (!StringUtils.isEmpty(catelogJson)){
                //缓存不为空直接返回
                Map<String, List<Catelog2VO>> result = JSON.parseObject(catelogJson,
                        new TypeReference<Map<String, List<Catelog2VO>>>() {
                        });
                return result;
            }
            System.out.println("开始查数据库喽....");
            //先将数据库中所有的分类都遍历出来
            List<CategoryEntity> selectList = baseMapper.selectList(null);
            // 一级分类
            List<CategoryEntity> level1Categorys = this.getLevel1Categorys();
            //封装数据
            Map<String, List<Catelog2VO>> map = level1Categorys.stream()
                    .collect(Collectors.toMap(k -> k.getCatId().toString(), v -> {
                        //遍历每一个一级分类,查到一级分类对应的二级分类
                        List<CategoryEntity> catelog2 = getParentCid(selectList, v.getCatId());
                        //封装上面的结果
                        List<Catelog2VO> catelog2VOS = null;
                        if (catelog2 != null) {
                            catelog2VOS = catelog2.stream().map(item2 -> {
                                Catelog2VO catelog2VO =
                                        new Catelog2VO(item2.getParentCid().toString(),
                                                null,
                                                item2.getCatId().toString(),
                                                item2.getName());
                                //找当前二级分类的三级分类封装成vo
                                List<CategoryEntity> catelog3 = getParentCid(selectList, item2.getCatId());
                                if (catelog3 != null) {
                                    List<Catelog2VO.Catelog3VO> catelog3VOS = catelog3.stream().map(item3 -> {
                                        Catelog2VO.Catelog3VO catelog3VO =
                                                new Catelog2VO.Catelog3VO(item3.getParentCid().toString(),
                                                        item3.getCatId().toString(),
                                                        item3.getName());
                                        return catelog3VO;
                                    }).collect(Collectors.toList());
                                    catelog2VO.setCatalog3List(catelog3VOS);
                                }
                                return catelog2VO;
                            }).collect(Collectors.toList());
                        }
                        return catelog2VOS;
                    }));
            //将查到的数据放入缓存,将对象转为json
            String valueJson = JSON.toJSONString(map);
            stringRedisTemplate.opsForValue().set("catelogJson",valueJson,1,TimeUnit.DAYS);
            return map;
        }
    }

不足

本地锁在分布式下的问题:本地锁在分布式的情况下锁不住进程。

分布式锁

原理

我们可以同时去一个地方占坑,如果占到,就去执行逻辑,否则就等待,直到锁的释放。等待锁释放可以使用自旋的方式。

分布式下死锁问题

死锁问题简单来说某一个进程拿到了锁,并开始执行业务的时候突然发生宕机的情况,这就导致锁还没删,其他进程也拿不到锁,线程就会不断等待。

解决方法:给锁设置过期时间,注意此处加入过期时间一定要和加锁是原子性的,不然的话可能在添加过期时间的时候宕机,还是死锁

SET resource_name my_random_value NX PX 30000

分布式下删锁的问题

假如拿到锁的进行业务时间大于锁的自动过期时间,这就会导致当业务还没执行结束锁已经删除,其他进程进来拿到了锁,当前面的进程业务执行完进行删锁的时候删的却是其他进程的锁,这时候又会有其他进程进来了,继续重复这个流程。

解决方法:设置uuid,每个进程只能删除自己的锁,如果自己的锁已经过期了,就不用删了。但是这种方式存在问题是因为网络是有开销的,假如在拿到缓存中的uuid网络进行往返的时候,锁突然过期,其他进程拿到锁设置自己的uuid,那么之前的进程拿到的uuid和自己设置的uuid进行比对,发现相同进行删除缓存的锁,这时候删除的还是别人的锁。

所以最终解决的方案是:判断uuid是否相同和删除锁也要是原子性的。所以可以使用redis的lua脚本进行操作

if redis.call("get",KEYS[1]) == ARGV[1] then
    return redis.call("del",KEYS[1])
else
    return 0
end
if (lock){
            System.out.println("获取分布式锁成功");
            Map<String, List<Catelog2VO>> dataFromDB = null;
            //加锁成功。。。执行业务
            //设置过期时间,必须和加锁是同步的原子的
            try {
                dataFromDB = getCatelogJsonFromDB();
            }finally {
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                //删除锁
                stringRedisTemplate.execute(new DefaultRedisScript<Long>(script,Long.class),Arrays.asList("lock"),uuid);
            }
            return dataFromDB;
        }
        else {
            System.out.println("获取分布式锁失败...等待重试...");
            //加锁失败,开启重试机制
            //休眠一百毫秒
            try {
                TimeUnit.MICROSECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return  getCatelogJsonFromDBWithRedisLock();  //自选方式
        }
    }

简单实现

@Override
    public Map<String, List<Catelog2VO>> getCatelogJson() {
        // 空结果缓存:解决缓存穿透问题
        // 设置过期时间: 解决缓存雪崩
        // 加锁:解决缓存击穿问题
        //加入缓存逻辑先查看redis中有没有数据,有的话直接返回并且将redis中string数组取出来进行反序列化,没有的调用查数据库的方法进行查询
        //查询redis中数据
        ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
        String catelogJson = ops.get("catelogJson");
        //如果redis中没有数据那么就调方法返回数据然后序列化存到redis中
        if (StringUtils.isEmpty(catelogJson)) {
            //System.out.println("缓存没命中....开始查询数据库....");
            Map<String, List<Catelog2VO>> catelogJsonFromDB = getCatelogJsonFromDBWithRedisLock();
//            String catelogJsonFromDBString = JSON.toJSONString(catelogJsonFromDB);
//            ops.set("catelogJson", catelogJsonFromDBString);
            //取数据
            catelogJson = ops.get("catelogJson");
        }
        //将redis中的json数据进行反序列 并返回
        return JSON.parseObject(catelogJson, new TypeReference<Map<String, List<Catelog2VO>>>() {});
    }
    public Map<String, List<Catelog2VO>> getCatelogJsonFromDBWithRedisLock() {
        ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
        String catelogJson = ops.get("catelogJson");
        if (!StringUtils.isEmpty(catelogJson)) {
            //如果此时缓存中有数据则直接返回即可
            return JSON.parseObject(catelogJson, new TypeReference<Map<String, List<Catelog2VO>>>() {});
        }
        //将redis中的json数据进行反序列 并返回
        //占分布式锁  去redis占坑
        String uuid = UUID.randomUUID().toString();
        Boolean lock = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid,3000, TimeUnit.MINUTES);
        if (lock){
            System.out.println("获取分布式锁成功");
            Map<String, List<Catelog2VO>> dataFromDB = null;
            //加锁成功。。。执行业务
            //设置过期时间,必须和加锁是同步的原子的
            try {
                dataFromDB = getCatelogJsonFromDB();
            }finally {
                String script = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
                //删除锁
                stringRedisTemplate.execute(new DefaultRedisScript<Long>(script,Long.class),Arrays.asList("lock"),uuid);
            }
            return dataFromDB;
        }
        else {
            System.out.println("获取分布式锁失败...等待重试...");
            //加锁失败,开启重试机制
            //休眠二百毫秒
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return  getCatelogJsonFromDBWithRedisLock();  //自旋方式
        }
    }
    public Map<String, List<Catelog2VO>> getCatelogJsonFromDB() {
        synchronized (this){
            //得到锁以后我们应该再去缓存中查看一次,如果没有才继续确定查询
            String catelogJson = stringRedisTemplate.opsForValue().get("catelogJson");
            if (!StringUtils.isEmpty(catelogJson)){
                //缓存不为空直接返回
                Map<String, List<Catelog2VO>> result = JSON.parseObject(catelogJson,
                        new TypeReference<Map<String, List<Catelog2VO>>>() {
                        });
                return result;
            }
            System.out.println("开始查数据库喽....");
            //先将数据库中所有的分类都遍历出来
            List<CategoryEntity> selectList = baseMapper.selectList(null);
            // 一级分类
            List<CategoryEntity> level1Categorys = this.getLevel1Categorys();
            //封装数据
            Map<String, List<Catelog2VO>> map = level1Categorys.stream()
                    .collect(Collectors.toMap(k -> k.getCatId().toString(), v -> {
                        //遍历每一个一级分类,查到一级分类对应的二级分类
                        List<CategoryEntity> catelog2 = getParentCid(selectList, v.getCatId());
                        //封装上面的结果
                        List<Catelog2VO> catelog2VOS = null;
                        if (catelog2 != null) {
                            catelog2VOS = catelog2.stream().map(item2 -> {
                                Catelog2VO catelog2VO =
                                        new Catelog2VO(item2.getParentCid().toString(),
                                                null,
                                                item2.getCatId().toString(),
                                                item2.getName());
                                //找当前二级分类的三级分类封装成vo
                                List<CategoryEntity> catelog3 = getParentCid(selectList, item2.getCatId());
                                if (catelog3 != null) {
                                    List<Catelog2VO.Catelog3VO> catelog3VOS = catelog3.stream().map(item3 -> {
                                        Catelog2VO.Catelog3VO catelog3VO =
                                                new Catelog2VO.Catelog3VO(item3.getParentCid().toString(),
                                                        item3.getCatId().toString(),
                                                        item3.getName());
                                        return catelog3VO;
                                    }).collect(Collectors.toList());
                                    catelog2VO.setCatalog3List(catelog3VOS);
                                }
                                return catelog2VO;
                            }).collect(Collectors.toList());
                        }
                        return catelog2VOS;
                    }));
            //将查到的数据放入缓存,将对象转为json
            String valueJson = JSON.toJSONString(map);
            stringRedisTemplate.opsForValue().set("catelogJson",valueJson,1,TimeUnit.DAYS);
            return map;
        }
    }

Redisson

阻塞锁

Redisson可以对redis进行管理,自动完成很多关于缓存的操作,例如加锁

// 1)锁的自动续期。不用担心业务时间长锁自动过期被删除掉,运行期间自动给锁进行续期30秒
// 2)加锁的业务完成后就不会给锁自动续期,即使不解锁也会自动释放

引入依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.12.0</version>
</dependency>

配置config

@Configuration
@ComponentScan
@EnableCaching
public class MyRedissonConfig {
    @Bean(destroyMethod="shutdown")
    RedissonClient redisson() throws IOException {
        Config config = new Config();
        config.useSingleServer()
                .setAddress("redis://192.168.100.10:6379");
        //根据config创建redissonclient实例
        RedissonClient redissonClient = Redisson.create(config);
        return redissonClient;
    }
}

controller

@Autowired
private RedissonClient redison;
@GetMapping("/hello")
public String hello(){
    //获取一把锁 只要锁的名字一样 就是同一把锁
    RLock lock = redison.getLock("my-lock");
    lock.lock();
    try {
        System.out.println("加锁成功等待执行业务....."+ Thread.currentThread().getId());
        Thread.sleep(30000);
    }catch (Exception e){
        e.printStackTrace();
    }finally {
        System.out.println("释放锁。。。。"+ Thread.currentThread().getId());
        lock.unlock();
    }
    return "hello";
}

读写锁

  • 读+ 读 = 无锁
  • 读+ 写 = 等读完成才能读
  • 写 + 写 = 阻塞状态
  • 写 + 读 = 等写完成才能读
@GetMapping("/read")
    public String read(){
        RReadWriteLock readWriteLock = redison.getReadWriteLock("wr-lock");
        RLock lock = readWriteLock.readLock();
        //读数据
        try {
            lock.lock(30, TimeUnit.SECONDS);
            ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
            Thread.sleep(10000);
            String myMsg = ops.get("myMsg");
            return myMsg;
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            System.out.println("释放锁。。。。"+ Thread.currentThread().getId());
            lock.unlock();
        }
       return "什么也没读到";
    }
    @GetMapping("/write")
    public String write(){
        RReadWriteLock readWriteLock = redison.getReadWriteLock("wr-lock");
        RLock lock = readWriteLock.writeLock();
        try {
            lock.lock(30, TimeUnit.SECONDS);
            ValueOperations<String, String> ops = stringRedisTemplate.opsForValue();
            ops.set("myMsg", UUID.randomUUID().toString());
            Thread.sleep(10000);
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            System.out.println("释放锁。。。。"+ Thread.currentThread().getId());
            lock.unlock();
        }
        return "什么也没读到";
    }

闭锁

只有当条件满足后才能释放锁。

//闭锁
@GetMapping("lockDoor")
public String door() throws InterruptedException {
    RCountDownLatch door = redison.getCountDownLatch("door");
    door.trySetCount(3);
    door.await();
    return "放假啦!!!!";
}
@GetMapping("gogogo/{id}")
public String door(@PathVariable Long id) {
    RCountDownLatch door = redison.getCountDownLatch("door");
    door.countDown();
    return id + "班走喽!!";
}

信号量

可以设置值作为信号量,当有信号量的时候那么就可以进行操作,没有信号量的时候需要等待其他进程释放后在进行操作。

应用场景例如:限流操作,设置一个应用的最大流量是1w

@GetMapping("setPosition")
    public String setPosition() throws InterruptedException {
        RSemaphore park = redison.getSemaphore("park");
        park.trySetPermits(3);
        return "设置了3个车位";
    }
    @GetMapping("park")
    public String park() throws InterruptedException {
        RSemaphore park = redison.getSemaphore("park");
        park.acquire();
        //park.tryAcquire(); 尝试获取信号量,会返回一个布尔值
        return "停车成功喽~~~";
    }
    
    @GetMapping("go")
    public String go() throws InterruptedException {
        RSemaphore park = redison.getSemaphore("park");
        park.release();
        return "车开走喽~~~";
    }
相关实践学习
基于Redis实现在线游戏积分排行榜
本场景将介绍如何基于Redis数据库实现在线游戏中的游戏玩家积分排行榜功能。
云数据库 Redis 版使用教程
云数据库Redis版是兼容Redis协议标准的、提供持久化的内存数据库服务,基于高可靠双机热备架构及可无缝扩展的集群架构,满足高读写性能场景及容量需弹性变配的业务需求。 产品详情:https://www.aliyun.com/product/kvstore &nbsp; &nbsp; ------------------------------------------------------------------------- 阿里云数据库体验:数据库上云实战 开发者云会免费提供一台带自建MySQL的源数据库&nbsp;ECS 实例和一台目标数据库&nbsp;RDS实例。跟着指引,您可以一步步实现将ECS自建数据库迁移到目标数据库RDS。 点击下方链接,领取免费ECS&amp;RDS资源,30分钟完成数据库上云实战!https://developer.aliyun.com/adc/scenario/51eefbd1894e42f6bb9acacadd3f9121?spm=a2c6h.13788135.J_3257954370.9.4ba85f24utseFl
目录
打赏
0
0
0
0
4
分享
相关文章
分布式系统架构8:分布式缓存
本文介绍了分布式缓存的理论知识及Redis集群的应用,探讨了AP与CP的区别,Redis作为AP系统具备高性能和高可用性但不保证强一致性。文章还讲解了透明多级缓存(TMC)的概念及其优缺点,并详细分析了memcached和Redis的分布式实现方案。此外,针对缓存穿透、击穿、雪崩和污染等常见问题提供了应对策略,强调了Cache Aside模式在解决数据一致性方面的作用。最后指出,面试中关于缓存的问题多围绕Redis展开,建议深入学习相关知识点。
182 8
【📕分布式锁通关指南 01】从解决库存超卖开始加锁的初体验
本文通过电商场景中的库存超卖问题,深入探讨了JVM锁、MySQL悲观锁和乐观锁的实现及其局限性。首先介绍了单次访问下库存扣减逻辑的正常运行,但在高并发场景下出现了超卖问题。接着分析了JVM锁在多例模式、事务模式和集群模式下的失效情况,并提出了使用数据库锁机制(如悲观锁和乐观锁)来解决并发问题。 悲观锁通过`update`语句或`select for update`实现,能有效防止超卖,但存在锁范围过大、性能差等问题。乐观锁则通过版本号或时间戳实现,适合读多写少的场景,但也面临高并发写操作性能低和ABA问题。 最终,文章强调没有完美的方案,只有根据具体业务场景选择合适的锁机制。
50 12
分布式系统架构7:本地缓存
这是小卷关于分布式系统架构学习的第10篇文章,主要介绍本地缓存的基础理论。文章分析了引入缓存的利弊,解释了缓存对CPU和I/O压力的缓解作用,并讨论了缓存的吞吐量、命中率、淘汰策略等属性。同时,对比了几种常见的本地缓存工具(如ConcurrentHashMap、Ehcache、Guava Cache和Caffeine),详细介绍了它们的访问控制、淘汰策略及扩展功能。
77 6
云端问道21期方案教学-应对高并发,利用云数据库 Tair(兼容 Redis®*)缓存实现极速响应
云端问道21期方案教学-应对高并发,利用云数据库 Tair(兼容 Redis®*)缓存实现极速响应
云端问道21期实操教学-应对高并发,利用云数据库 Tair(兼容 Redis®)缓存实现极速响应
本文介绍了如何通过云端问道21期实操教学,利用云数据库 Tair(兼容 Redis®)缓存实现高并发场景下的极速响应。主要内容分为四部分:方案概览、部署准备、一键部署和完成及清理。方案概览中,展示了如何使用 Redis 提升业务性能,降低响应时间;部署准备介绍了账号注册与充值步骤;一键部署详细讲解了创建 ECS、RDS 和 Redis 实例的过程;最后,通过对比测试验证了 Redis 缓存的有效性,并指导用户清理资源以避免额外费用。
本地消息表事务:10Wqps 高并发分布式事务的 终极方案,大厂架构师的 必备方案
45岁资深架构师尼恩分享了一篇关于分布式事务的文章,详细解析了如何在10Wqps高并发场景下实现分布式事务。文章从传统单体架构到微服务架构下分布式事务的需求背景出发,介绍了Seata这一开源分布式事务解决方案及其AT和TCC两种模式。随后,文章深入探讨了经典ebay本地消息表方案,以及如何使用RocketMQ消息队列替代数据库表来提高性能和可靠性。尼恩还分享了如何结合延迟消息进行事务数据的定时对账,确保最终一致性。最后,尼恩强调了高端面试中需要准备“高大上”的答案,并提供了多个技术领域的深度学习资料,帮助读者提升技术水平,顺利通过面试。
本地消息表事务:10Wqps 高并发分布式事务的 终极方案,大厂架构师的 必备方案
秒杀抢购场景下实战JVM级别锁与分布式锁
在电商系统中,秒杀抢购活动是一种常见的营销手段。它通过设定极低的价格和有限的商品数量,吸引大量用户在特定时间点抢购,从而迅速增加销量、提升品牌曝光度和用户活跃度。然而,这种活动也对系统的性能和稳定性提出了极高的要求。特别是在秒杀开始的瞬间,系统需要处理海量的并发请求,同时确保数据的准确性和一致性。 为了解决这些问题,系统开发者们引入了锁机制。锁机制是一种用于控制对共享资源的并发访问的技术,它能够确保在同一时间只有一个进程或线程能够操作某个资源,从而避免数据不一致或冲突。在秒杀抢购场景下,锁机制显得尤为重要,它能够保证商品库存的扣减操作是原子性的,避免出现超卖或数据不一致的情况。
87 10
分布式读写锁的奥义:上古世代 ZooKeeper 的进击
本文作者将介绍女娲对社区 ZooKeeper 在分布式读写锁实践细节上的思考,希望帮助大家理解分布式读写锁背后的原理。
115 11
|
3月前
|
Redis作为PHP缓存解决方案的优势、实现方式及注意事项。Redis凭借其高性能、丰富的数据结构、数据持久化和分布式支持等特点,在提升应用响应速度和处理能力方面表现突出
本文深入探讨了Redis作为PHP缓存解决方案的优势、实现方式及注意事项。Redis凭借其高性能、丰富的数据结构、数据持久化和分布式支持等特点,在提升应用响应速度和处理能力方面表现突出。文章还介绍了Redis在页面缓存、数据缓存和会话缓存等应用场景中的使用,并强调了缓存数据一致性、过期时间设置、容量控制和安全问题的重要性。
69 5
redis高并发缓存中间件总结!
本文档详细介绍了高并发缓存中间件Redis的原理、高级操作及其在电商架构中的应用。通过阿里云的角度,分析了Redis与架构的关系,并展示了无Redis和使用Redis缓存的架构图。文档还涵盖了Redis的基本特性、应用场景、安装部署步骤、配置文件详解、启动和关闭方法、systemctl管理脚本的生成以及日志警告处理等内容。适合初学者和有一定经验的技术人员参考学习。
415 7