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

本文涉及的产品
云数据库 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
相关文章
|
7天前
|
消息中间件 Java Linux
2024年最全BATJ真题突击:Java基础+JVM+分布式高并发+网络编程+Linux(1),2024年最新意外的惊喜
2024年最全BATJ真题突击:Java基础+JVM+分布式高并发+网络编程+Linux(1),2024年最新意外的惊喜
|
12天前
|
缓存 NoSQL Java
分布式项目中锁的应用(本地锁-_redis【setnx】-_redisson-_springcache)-fen-bu-shi-xiang-mu-zhong-suo-de-ying-yong--ben-de-suo--redissetnx-springcache-redisson(一)
分布式项目中锁的应用(本地锁-_redis【setnx】-_redisson-_springcache)-fen-bu-shi-xiang-mu-zhong-suo-de-ying-yong--ben-de-suo--redissetnx-springcache-redisson
66 0
|
3天前
|
存储 缓存 算法
高并发架构设计三大利器:缓存、限流和降级
软件系统有三个追求:高性能、高并发、高可用,俗称三高。本篇讨论高并发,从高并发是什么到高并发应对的策略、缓存、限流、降级等。
42 1
|
12天前
|
存储 NoSQL Java
探索Java分布式锁:在高并发环境下的同步访问实现与优化
【4月更文挑战第17天】Java分布式锁是解决高并发下数据一致性问题的关键技术,通过Redis、ZooKeeper、数据库等方式实现。它确保多节点共享资源时的同步访问,防止数据不一致。优化策略包括锁超时重试、续期、公平性和性能优化。合理设计分布式锁对支撑大规模分布式系统至关重要。
|
12天前
|
存储 缓存 NoSQL
缓存、分布式缓存和持久化
这篇内容介绍了缓存的概念和Redis的作用,以口袋与公文包的比喻解释了缓存如何提高数据访问速度。Redis是一个内存中的高级缓存系统,能提升系统响应速度。接着讨论了为何需要分布式缓存,通过多个“篮子”(Redis节点)解决单点故障和性能瓶颈,保证高可用性和数据安全性。最后提到了Redis的两种持久化机制——RDB(定期数据快照)和AOF(记录写操作日志),分别用照片备份和实时同步来比喻,说明它们在数据丢失风险和恢复速度上的权衡。
|
12天前
|
存储 NoSQL Java
分布式锁,Redission,其它实现问题讲解,以及面试题回答案例
分布式锁,Redission,其它实现问题讲解,以及面试题回答案例
139 1
|
12天前
|
XML NoSQL Java
Java单体项目和分布式项目中的锁
Java单体项目和分布式项目中的锁 Java单体项目和分布式项目中的锁
34 2
|
12天前
|
缓存 NoSQL Redis
分布式项目中锁的应用(本地锁-_redis【setnx】-_redisson-_springcache)-fen-bu-shi-xiang-mu-zhong-suo-de-ying-yong--ben-de-suo--redissetnx-springcache-redisson(二)
分布式项目中锁的应用(本地锁-_redis【setnx】-_redisson-_springcache)-fen-bu-shi-xiang-mu-zhong-suo-de-ying-yong--ben-de-suo--redissetnx-springcache-redisson
41 0
|
1天前
|
存储 缓存 监控
利用Redis构建高性能的缓存系统
在现代Web应用中,性能优化是提升用户体验和响应速度的关键。Redis作为一款开源的内存数据结构存储系统,因其出色的性能、丰富的数据结构和灵活的使用方式,成为了构建高性能缓存系统的首选工具。本文将探讨Redis在缓存系统中的应用,分析其优势,并通过实例展示如何结合Redis构建高效、可靠的缓存系统,以应对高并发、大数据量等挑战。
|
5天前
|
缓存 NoSQL Redis
【后端面经】【缓存】36|Redis 单线程:为什么 Redis 用单线程而 Memcached 用多线程?-- Redis多线程
【5月更文挑战第21天】Redis启用多线程后,主线程负责接收事件和命令执行,IO线程处理读写数据。请求处理流程中,主线程接收客户端请求,IO线程读取并解析命令,主线程执行后写回响应。业界普遍认为,除非必要,否则不建议启用多线程模式,因单线程性能已能满足多数需求。公司实际场景中,启用多线程使QPS提升约50%,或选择使用Redis Cluster以提升性能和可用性。
11 0

热门文章

最新文章