郑州网站建设有限公司,网站建设前端后端,烟台定制网站建设报价,广州网络推广培训机构缓存
缓存技术是一种可以大幅度提高系统性能的技术#xff0c;我们可以在某些适用的场景下使用缓存来大幅度的提高系统性能
读缓存的基本流程#xff1a;
请求向缓存中查数据
if (命中) {返回缓存中的数据
} else {从数据库中取出数据将该数据在缓存中再存储一份返回缓存中…缓存
缓存技术是一种可以大幅度提高系统性能的技术我们可以在某些适用的场景下使用缓存来大幅度的提高系统性能
读缓存的基本流程
请求向缓存中查数据
if (命中) {返回缓存中的数据
} else {从数据库中取出数据将该数据在缓存中再存储一份返回缓存中的数据
}本地缓存
我们在单体系统应用中可以使用本地缓存来进行系统的缓存需求我们可以在模块中自定义一个HashMap将所需要的信息以键值对的方式存储进去按照缓存的查找逻辑进行操作也可以有很高的性能甚至于说这种方式减少了一层中间件的IO甚至比使用Redis的效率还要高但问题就在于本地缓存非常不适合于在分布式系统中实现因为分布式系统中有几个节点就需要几个本地缓存而我们也不确定我们会被负载均衡到哪个节点中。
Redis
Redis是一种缓存中间件其解决了分布式系统难以使用本地缓存的问题
!-- 引入redis--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependency在application.yml中进行配置
spring:redis:host: xxx.xxx.xxx.xxx必须要配置的是redis的地址其默认端口是6379若配置了密码也必须对密码进行配置
SpringBoot帮助我们对redis的使用进行了十分简化的配置其提供了两种实现方式
返回MapObject, Object的RedisTemplate返回MapString, Object的StringRedisTemplate
使用测试 AutowiredStringRedisTemplate stringRedisTemplate;Testpublic void testStringRedisTemplate() {ValueOperationsString, String testValue stringRedisTemplate.opsForValue();testValue.set(test, Hello World_ UUID.randomUUID().toString());String test testValue.get(test);System.out.println(test);// 输出Hello World_6ebc06c7-245f-4698-a5dd-c5e832ba7b7f}真实业务中的使用
注意我们使用String作为值进行存储的情况下我们会把JSON类型的字符串作为Value故我们在使用时要注意Json与对象的转换这样做是为了方便我们在全平台进行操作。序列化与反序列化 // 真正的业务逻辑Overridepublic MapString, ListCatelog2Vo getCatalogJson() {// 试图获取一下缓存String catalogJSON redisTemplate.opsForValue().get(catalogJSON);// 如果缓存中不存在数据if (StringUtils.isEmpty(catalogJSON)) {// 从数据库中取出数据MapString, ListCatelog2Vo catalogJsonFromDb getCatalogJsonFromDb();// 序列化catalogJSON JSON.toJSONString(catalogJsonFromDb);// 将数据放入缓存redisTemplate.opsForValue().set(catalogJSON, catalogJSON);return catalogJsonFromDb;}// 反序列化// TypeReference是我们要转换的类型MapString, ListCatelog2Vo stringListMap JSON.parseObject(catalogJSON, new TypeReferenceMapString, ListCatelog2Vo() {});return stringListMap;}// 从数据库中读取数据的逻辑public MapString, ListCatelog2Vo getCatalogJsonFromDb(){...}注意我们在较老版本的letture中可能会出现DriectOutOfMemory异常这个异常是由于老版本的letture在与Netty交互时不能有效的释放内存我们必须使用更高版本的letture或者使用jedis客户端来避免这个问题
注意Redis默认的内存会使用Xmx的内存我们也不能一味的只增大这个容量因为只要内存不能有效释放这个早晚会满
在使用Redis时甚至可以得到1300的TPS而没使用时只有7TPS
Redis三大问题
缓存穿透
一直查询一个在数据库中不存在的数据导致每次查询都进入数据库不经过缓存使用大量的查询很有可能导致数据库崩溃
解决方案将null存入缓存并加入短暂的过期时间致使不存在的数据也会被缓存拦截但要注意这个null的时间必须短暂不然会导致查不到存在数据的情况。
缓存雪崩
Redis中的数据在同一时间大量过时失效导致大量的数据进入数据库致使数据库崩溃
解决方案我们应该给缓存的过期时间再加一个1-5分钟的随机数让他们不会在同一时间失效
缓存击穿
一个高频热点Key失效后被同时超高频率的访问导致大量数据同时进入数据库致使数据库崩溃
解决方案加锁让请求一个一个来
三大Redis问题的解决方案
使用synchronized加锁避免缓存击穿的发生 // 从数据库中读取数据的逻辑// 使用sychronized进行加锁避免缓存击穿public MapString, ListCatelog2Vo getCatalogJsonFromDb() {ListCategoryEntity level1Categories getLevel1Categories();// 使用当前对象作为锁意为只有拿到当前对象的线程才可以执行下面的代码// 若缓存中已经存储了则直接从缓存中取出避免大量线程直接进来没走上面Redis的情况synchronized (this) {String catalogJSON redisTemplate.opsForValue().get(catalogJSON);if (!StringUtils.isEmpty(catalogJSON)) {MapString, ListCatelog2Vo stringListMap JSON.parseObject(catalogJSON, new TypeReferenceMapString, ListCatelog2Vo() {});return stringListMap;}System.out.println(查询了数据库);ListCategoryEntity categoryEntities baseMapper.selectList(null);MapString, ListCatelog2Vo parentCid level1Categories.stream().collect(Collectors.toMap(key - key.getCatId().toString(),value - {ListCategoryEntity level2Categories getParentCid(categoryEntities, value.getCatId());ListCatelog2Vo catelogVos null;if (level2Categories ! null) {catelogVos level2Categories.stream().map(item - {ListCategoryEntity level3Categories getParentCid(categoryEntities, item.getCatId());ListCatelog2Vo.Catelog3Vo level3Vos null;if (level3Categories ! null) {level3Vos level3Categories.stream().map(level3Item - {Catelog2Vo.Catelog3Vo catelog3Vo new Catelog2Vo.Catelog3Vo(item.getCatId().toString(), level3Item.getCatId().toString(), level3Item.getName());return catelog3Vo;}).collect(Collectors.toList());}Catelog2Vo catelog2Vo new Catelog2Vo(value.getCatId().toString(), level3Vos, item.getCatId().toString(), item.getName());return catelog2Vo;}).collect(Collectors.toList());}return catelogVos;}));return parentCid;}}但注意上面这种处理逻辑在数据库查完之后就释放锁但此时Redis中还没有存储故下一个线程仍然会查数据库造成锁失效
我们解决这种情况的方式就是将存储Redis的操作也放在synchronized代码块中
类似于下面这样 synchronized(this) {....................// 序列化catalogJSON JSON.toJSONString(parentCid);// 将数据放入缓存redisTemplate.opsForValue().set(catalogJSON, catalogJSON);return parentCid;}此处还要注意我们这里使用synchronized做的是本地锁不适用于分布式系统多个模块不共享锁每个模块都有一个this对象若要在分布式系统下进行开发还需要进一步使用分布式锁
分布式锁
分布式锁的基本原理我们在redis中存储一对键值对我们让所有的服务都向Redis中Set这个键值对哪个服务Set成功了那个键值对就代表这个服务获取到了锁
Redis的Set操作可以在后面携带参数NX意思为如果该key不存在我们会进行存储如果存在则不存储 // 使用分布式锁public MapString, ListCatelog2Vo getCatalogJsonFromDbWithRedisLock() {ListCategoryEntity level1Categories getLevel1Categories();/*** 令程序试图向Redis中添加数据若添加成功则继续进行后续的查数据库操作* 若添加失败则证明已经有其他线程拿到了锁我们必须进行重试再继续试图拿到锁然后向下进行这个时候就有缓存了*/Boolean lock redisTemplate.opsForValue().setIfPresent(lock, 111);if (lock) {MapString, ListCatelog2Vo catalogJsonFromDb getCatalogJsonFromDb();// 业务执行结束后释放锁redisTemplate.delete(lock);return catalogJsonFromDb;} else {// 若没有拿到锁重试取锁MapString, ListCatelog2Vo catalogJsonFromDbWithRedisLock getCatalogJsonFromDbWithRedisLock();return catalogJsonFromDbWithRedisLock;}}另外若我们在delete之前发生了断电系统崩溃等情况就会导致死锁
解决死锁的方式一般为给我们的lock键设置一个过期时间注意这个过期时间的设置不能分两行写而应该在一行里以一个原子操作完成
将加锁方法修改为
Boolean lock redisTemplate.opsForValue().setIfPresent(lock, 111, 300, TimeUnit.SECONDS);这样就实现了原子操作令加锁不会被中断也添加了过期时间
但是这样还是有问题我们在加锁后结束了线程时我们的锁会被其他线程删除此时我们还有解决方案
给我们添加锁的方法添加的Value添加为UUID让我们每个线程都只能删除自己对应的锁 String uuidString UUID.randomUUID().toString();Boolean lock redisTemplate.opsForValue().setIfPresent(lock, uuidString, 300, TimeUnit.SECONDS);if (lock) {MapString, ListCatelog2Vo catalogJsonFromDb getCatalogJsonFromDb();// 业务执行结束后释放锁if (uuidString.equals(redisTemplate.opsForValue().get(key))) {redisTemplate.delete(lock);}这样就可以做到尽可能的删除自己的锁了但是还有问题
因为我们设置了数据过期时间而数据很有可能在我们获取了key之后过期这样你删除的就又是别人的Key了
根本原因就是我们的判断和删锁操作必须是一个原子操作不可以中断
我们可以使用Redis的lua脚本实现这个操作 String script if redis.call(\get\,KEYS[1]) ARGV[1]\n then\n return redis.call(\del\,KEYS[1])\n else\n return 0\n end;// 这里需要传入一个他的默认RedisScript这个RedisScript的泛型就是这个脚本执行的返回值类型这里是int型// 这个对象的内容传入脚本和返回值类型的反射类型// 第二个参数传我们Key的集合这里使用Arrays.asList()进行转换// 第三个参数传入我们要对比的uuidString// 这样我们就通过lua脚本把Redis的对比和删除操作设置为原子操作了redisTemplate.execute(new DefaultRedisScriptInteger(script, Integer.class), Arrays.asList(lock), uuidString);使用自定义锁的完整操作举例如下 // 真正的业务逻辑Overridepublic MapString, ListCatelog2Vo getCatalogJson() {// 试图获取一下缓存String catalogJSON redisTemplate.opsForValue().get(catalogJSON);// 如果缓存中不存在数据
// String catalogJSON null;if (StringUtils.isEmpty(catalogJSON)) {System.out.println(缓存不命中查询数据库..............);// 从数据库中取出数据MapString, ListCatelog2Vo catalogJsonFromDb getCatalogJsonFromDbWithRedisLock();return catalogJsonFromDb;}System.out.println(缓存命中直接返回...........);// 反序列化// TypeReference是我们要转换的类型MapString, ListCatelog2Vo stringListMap JSON.parseObject(catalogJSON, new TypeReferenceMapString, ListCatelog2Vo() {});return stringListMap;}// 使用分布式锁public MapString, ListCatelog2Vo getCatalogJsonFromDbWithRedisLock() {ListCategoryEntity level1Categories getLevel1Categories();/*** 令程序试图向Redis中添加数据若添加成功则继续进行后续的查数据库操作* 若添加失败则证明已经有其他线程拿到了锁我们必须进行重试再继续试图拿到锁然后向下进行这个时候就有缓存了*/String uuidString UUID.randomUUID().toString();Boolean lock redisTemplate.opsForValue().setIfAbsent(lock, uuidString, 300, TimeUnit.SECONDS);if (lock) {System.out.println(获取分布式锁成功.......);// 业务执行结束后释放锁
// if (uuidString.equals(redisTemplate.opsForValue().get(key))) {
// redisTemplate.delete(lock);
// }MapString, ListCatelog2Vo catalogJsonFromDb;// 使用try-finally块来保证锁的释放try {catalogJsonFromDb getCatalogJsonFromDb();} finally {String script if redis.call(\get\,KEYS[1]) ARGV[1]\n then\n return redis.call(\del\,KEYS[1])\n else\n return 0\n end;// 这里需要传入一个他的默认RedisScript这个RedisScript的泛型就是这个脚本执行的返回值类型这里是int型// 这个对象的内容传入脚本和返回值类型的反射类型// 第二个参数传我们Key的集合这里使用Arrays.asList()进行转换// 第三个参数传入我们要对比的uuidString// 这样我们就通过lua脚本把Redis的对比和删除操作设置为原子操作了Long lock1 redisTemplate.execute(new DefaultRedisScriptLong(script, Long.class), Arrays.asList(lock), uuidString);}return catalogJsonFromDb;} else {System.out.println(获取分布式锁失败.......等待重试);// 若没有拿到锁重试取锁MapString, ListCatelog2Vo catalogJsonFromDbWithRedisLock getCatalogJsonFromDbWithRedisLock();return catalogJsonFromDbWithRedisLock;}}Redisson
我们在使用分布式锁的过程中不可避免的要对各种各样的场景进行判断操作而我们使用原生的Redis进行分布式锁的处理其一有可能会发生我们某种情况没有处理好系统上线后发生严重错误的情况其二这种原生操作会对我们的开发效率有很大的限制故我们引入一个Redis中推荐的第三方技术来处理分布式锁相关的问题
Redisson是一种操作Redis的客户端和letture和jedis类似
引入引入Redisson的依赖注意Redisson和SpringBoot依赖有版本关系
!-- 引入redission--dependencygroupIdorg.redisson/groupIdartifactIdredisson/artifactIdversion3.23.3/version/dependency我们可以使用配置文件或者对象文件使用Config获取JSON或YAML文件、Config进行配置程序化配置操作
这里使用Config进行配置操作
创建一个Config类进行配置
Configuration
public class MyRedissonConfig {Bean(destroyMethod shutdown)public RedissonClient redissonClient() throws IOException {Config config new Config();// 集群模式的配置方式
// config.useClusterServers().addNodeAddress(127.0.0.1:7001, 127.0.0.1:7002);// 单点模式的配置方式config.useSingleServer().setAddress(redis://192.168.202.142:6379);RedissonClient redissonClient Redisson.create(config);return redissonClient;}
}进行一点点测试 AutowiredRedissonClient redissonClient;Testpublic void testRedisson() {System.out.println(redissonClient);}可重入锁
Redisson锁最终都继承了JUC锁在单体系统中JUC可以完全替代Redisson换句话说Redisson就是分布式系统中的JUC
一个简单的示例 ResponseBodyGetMapping(/hello)public String hello() {// 创建一个锁RLock lock redisson.getLock(my-lock);// 加锁阻塞式lock.lock();try {System.out.println(加锁成功执行业务......... Thread.currentThread().getId());Thread.sleep(30000);}catch (Exception e){} finally {lock.unlock();System.out.println(释放锁.......... Thread.currentThread().getId());}return hello;}注意在这里是不会出现死锁问题的因为redisson给锁设置了过期时间这个过期时间是30S如果业务超长Redisson的看门狗机制会给锁自动续期
阻塞式加锁会执行一个while(true)死循环在循环中想要出去就只能获取到锁同时Redisson也支持在加锁时自动添加一个过期时间lock.lock(10, TimeUnit.SECONDS)这样在10秒之后不管业务是否执行都会让锁过期
如果我们没有指定过期时间Redisson会使用看门狗的默认时间30 * 1000ms
而我们看门狗机制的默认续期时间是 WatchDog / 3 也就是看门狗时间的三分之一在20s的时候进行续期
另外一种方式是
boolean res lock.trylock(100, 10, TimeUnit.SECONDS);
if (res) {try {} finally {lock.unlock();}
}这种方式的作用是尝试进行加锁如果加锁成功返回true
公平锁
公平锁是指我们的线程按照进入等待的顺序获取锁谁先来的就让谁先获得锁
RLock fairLock redisson.getFairLock(any-lock);
fairLock.lock();读写锁
我们在进行读写操作的时候一般是允许同时读但不允许同时写和边写边读故这里读写锁就诞生了 GetMapping(/write)ResponseBodypublic String writeValue() {String s ;// 创建读写锁RReadWriteLock lock redisson.getReadWriteLock(rw-lock);// 获取这个读写锁的写锁RLock rLock lock.writeLock();rLock.lock();try {s UUID.randomUUID().toString();Thread.sleep(30000);redisTemplate.opsForValue().set(writeValue, s);} catch (Exception e) {e.printStackTrace();} finally {rLock.unlock();}return s;}GetMapping(/read)ResponseBodypublic String readValue() {String s ;RReadWriteLock lock redisson.getReadWriteLock(rw-lock);RLock rLock lock.readLock();rLock.lock();try {s redisTemplate.opsForValue().get(writeValue);} catch (Exception e) {e.printStackTrace();} finally {rLock.unlock();}return s;}最关键的问题就是如果写锁存在读锁就必须等待这也就意味着如果写操作没有完成读操作就无法获取到数据这也就保证了我们每次读取获得的都是最新的数据
写 写阻塞式等待写 读读等待写读 写写也要等待读读 读共享锁相当于无锁
信号量
Redisson也可以获取信号量这个操作会向Redis中插入一个数值在这个数值耗尽之前都是允许进入的。 GetMapping(/park)ResponseBodypublic String park() throws InterruptedException {// 声明一个信号量RSemaphore park redisson.getSemaphore(park);// 加上信号量锁park.acquire();return ok ;}GetMapping(/go)ResponseBodypublic String go() throws InterruptedException {RSemaphore park redisson.getSemaphore(park);park.release();return ok ;}信号量最常用的操作就是分布式限流通过设置信号量大小、以及tryAcquire()操作来进行限流获取到是true再进行操作获取不到就直接跳转一个提示页面或者流量过大的提示。 GetMapping(/park)ResponseBodypublic String park() throws InterruptedException {// 声明一个信号量RSemaphore park redisson.getSemaphore(park);// 加上信号量锁// .tryAcquire()方法会尝试获取锁若成功会返回true不成功直接返回falseboolean b park.tryAcquire();if (b) {return ok b;}return ok b;}GetMapping(/go)ResponseBodypublic String go() throws InterruptedException {RSemaphore park redisson.getSemaphore(park);park.release();return ok ;}
闭锁
闭锁机制是一种要求另一个程序调用一定次数之后才允许另一个程序进行调用的锁其实现类似于一个班全部的学生走完之后才允许锁门。实现方式 GetMapping(/lockDoor)ResponseBodypublic String lockDoor() throws InterruptedException {RCountDownLatch door redisson.getCountDownLatch(door);door.trySetCount(5);// 要求闭锁全部完成之后再进行door.await();return 放假了......;}GetMapping(/gogogo/{id})ResponseBodypublic String gogogo(PathVariable(id) Long id) throws InterruptedException {RCountDownLatch door redisson.getCountDownLatch(door);door.countDown();// 要求闭锁全部完成之后再进行return id 班的人都走了;}总结
Redisson通过看门狗机制和LUA脚本保证了对Redis的操作是一个原子操作以及我们系统中断时也能保证锁的释放同时也能保证自己的锁不会去解别人的锁这就避免了死锁问题的出现
锁的粒度问题
一般来讲我们以每一条数据作为一把锁例如product-11-lock 这样的每一条数据一把锁这样可以保证我们一个业务只会影响自己的业务不会影响到其他内容否则就可能出现 A 请求被 B 阻塞的问题就很抽象
实际业务改写就变成下面的样子 public MapString, ListCatelog2Vo getCatalogJsonFromDbWithRedissonLock() {ListCategoryEntity level1Categories getLevel1Categories();RLock lock redisson.getLock(catalogJson-lock);lock.lock();MapString, ListCatelog2Vo catalogJsonFromDb;// 使用try-finally块来保证锁的释放try {catalogJsonFromDb getCatalogJsonFromDb();} finally {lock.unlock();}return catalogJsonFromDb;}Redis数据一致性问题
我们为了数据的读取速度使用Redis存储一些热点数据但是问题也随之出现若我们取出数据并存入Redis之后数据库中进行了是数据的修改我们的Redis和数据库的数据就不一致了这会导致我们无法读取到正确的数据。
对于Redis的与数据库的一致性问题我们引出两种解决方式
双写模式我们每次修改数据库都对Redis进行一次重新写入失效模式我们在修改数据库后对Redis中对应的键进行删除等待下次读数据的时候再重新写入
但这两种模式都有一定的问题 双写模式 我们的线程如果在写缓存的过程中发生了切换很有可能导致脏数据的出现 A — 写数据库 —线程切换 B — 写数据库 — B写缓存 — A写缓存 我们本该在缓存中读取到B的数据但在这种情况下却读取到了A的数据 但这种问题也只是暂时的不一致问题从整体上来看我们的数据还是一致的因为Redis中的数据我们会给他设置一个过期时间当数据过期之后再被查询的时候还是会查到最新的数据也就是说其也会最终显示正确的数据也可以保证最终一致性但其无法保证实时一致性若对实时一致性有强烈要求我们就需要使用加锁的方式将一个线程锁在一起使用 失效模式 也是发生了线程切换的场景 A写数据库写了一半 — 线程切换B读缓存 — B读数据库尚未更新缓存 — 线程切换— A继续写数据库 — A删缓存 — 线程切换B更新缓存 这样就又出现了脏数据情况不过这也是暂时不一致的情况其也还是能保证最终一致性
对于这个问题我们首先要考虑以下的场景
我们要优先考虑业务场景对于用户维度的情况其是否可能会发生同时读写的情况修改个人信息和读个人信息会同时发生吗我们的业务是否真的一定需要解决实时的一致性问题另外我们对于基础数据也可以使用canal订阅binlog的方式进行处理实在不行我们就加一个读写锁最受不了的解决方案
简单介绍一下CannalCannal是一种中间件其会把自己伪装成MySql的一个从数据库MySql在更新的时候会把更新的信息传递给从数据库Cannal就把存储在binlog中的更新日志进行更新并进一步的再对Redis进行修改 工作情况最近在干嘛为什么减弱项目方向加强技术方向项目相关的太低效、唐杰哥那边无法提供高效的技能提升 云视讯问题很多会无法参加表达很想参加各种会但是都参加不了包括GIS、软评、今天的智慧xxx都错过了
