谷粒随享
学习目标:
一个看起来很简单的应用,背后可能需要数十或数百个服务来支撑,一个请求就要多次服务调用。当请求变慢、或者不能使用时,我们是不知道是哪个后台服务引起的。这时,我们使用 Zipkin 就能解决这个问题。由于业务访问量的增大,业务复杂度增加,以及微服务架构和容器技术的兴起,要对系统进行各种拆分。微服务系统拆分后,我们可以使用 Zipkin 链路,来快速定位追踪有故障的服务点。
Zipkin/SkyWalking 是一款开源的分布式实时数据追踪系统(Distributed Tracking System),能够收集服务间调用的时序数据,提供调用链路的追踪。
Zipkin 其主要功能是聚集来自各个异构系统的实时监控数据,在微服务架构下,十分方便地用于服务响应延迟等问题的定位。
Zipkin 每一个调用链路通过一个 trace id 来串联起来,只要你有一个 trace id ,就能够直接定位到这次调用链路,并且可以根据服务名、标签、响应时间等进行查询,过滤那些耗时比较长的链路节点。
Zipkin 分布式跟踪系统就能非常好地解决该问题,主要解决以下3点问题:
在优化前通过Zipkin链路追踪进行查看接口耗时
在service-util
工具模块pom.xml中新增Zipkin相关依赖
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-tracing-bridge-brave</artifactId>
</dependency>
<dependency>
<groupId>io.zipkin.reporter2</groupId>
<artifactId>zipkin-reporter-brave</artifactId>
</dependency>
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-observation</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-zipkin</artifactId>
<version>2.2.8.RELEASE</version>
</dependency>
<dependency>
<groupId>io.github.openfeign</groupId>
<artifactId>feign-micrometer</artifactId>
<version>12.5</version>
</dependency>
在Nacos配置common.yaml
中新增配置(已完成)
management:
zipkin:
tracing:
endpoint: http://192.168.200.6:9411/api/v2/spans
tracing:
sampling:
probability: 1.0 # 记录速率100%
解决异步任务+多线程导致异步Feign请求无法被链路追踪,故需要未线程池设置装饰器(将当前线程内上下文中trace_id传递到子线程中),故将当日资料中复制到service-util
且设置Spring线程池对象装饰器
//设置解决zipkin链路追踪不完整装饰器对象
taskExecutor.setTaskDecorator(new ZipkinTaskDecorator(zipkinHelper));
访问专辑详情接口、或查看专辑页面进行测试
虽然咱们实现了页面需要的功能,但是考虑到该页面是被用户高频访问的,所以性能需要优化。一般一个系统最大的性能瓶颈,就是数据库的io操作。从数据库入手也是调优性价比最高的切入点。
一般分为两个层面:
重点要讲的是第二个层面:尽量避免直接查询数据库。
解决办法就是:分布式缓存
缓存最常见的3个问题:
缓存穿透: 是指查询一个不存在的数据,由于缓存无法命中,将去查询数据库,但是数据库也无此记录,并且出于容错考虑,我们没有将这次查询的null写入缓存,这将导致这个不存在的数据每次请求都要到存储层去查询,失去了缓存的意义。在流量大时,可能DB就挂掉了,要是有人利用不存在的key频繁攻击我们的应用,这就是漏洞。
缓存雪崩:是指在我们设置缓存时采用了相同的过期时间,导致缓存在某一时刻同时失效,请求全部转发到DB,DB瞬时压力过重雪崩。
缓存击穿: 是指对于一些设置了过期时间的key,如果这些key可能会在某些时间点被超高并发地访问,是一种非常“热点”的数据。这个时候,需要考虑一个问题:如果这个key在大量请求同时进来之前正好失效,那么所有对这个key的数据查询都落到db,我们称为缓存击穿。
与缓存雪崩的区别:
之前,我们学习过synchronized 及lock锁,这些锁都是本地锁。接下来写一个案例,演示本地锁的问题
在service-album
微服务中添加 TestController 控制器:
package com.atguigu.tingshu.album.api;
import com.atguigu.tingshu.album.service.TestService;
import com.atguigu.tingshu.common.result.Result;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* @author atguigu
* @ClassName TestController
* @description: TODO
* @date 2023年08月22日
* @version: 1.0
*/
@RestController
@RequestMapping("api/album/test")
public class TestController {
// 注入服务层方法
@Autowired
private TestService testService;
/**
* 测试分布式锁
* @return
*/
@GetMapping("testLock")
public Result testLock(){
testService.testLock();
return Result.ok();
}
}
TestService接口:
public interface TestService {
/**
* 测试分布式锁
*/
void testLock();
}
TestServiceImpl实现:
package com.atguigu.tingshu.album.service.impl;
import com.atguigu.tingshu.album.service.TestService;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
/**
* @author: atguigu
* @create: 2023-10-07 08:54
*/
@Service
public class TestServiceImpl implements TestService {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Override
public void testLock() {
try {
//1.从Redis缓存中获取key="num"的值 保证redis中存在"num"(手动提前在redis中创建key)
String value = stringRedisTemplate.opsForValue().get("num");
if (StringUtils.isBlank(value)) {
return;
}
//2.对获取到值进行+1操作
int num = Integer.parseInt(value);
stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));
} catch (NumberFormatException e) {
e.printStackTrace();
}
}
}
启动三个微服务端口号分别为 8501、8601、8701 然后通过网关复杂均衡访问控制器
注释掉nacos配置中端口,在本地bootstrap.properties中配置默认端口,避免端口被覆盖。导致端口冲突
测试本地锁:
在linux 系统中安装 ab 压力测试工具 httpd-tools(已安装)
yum install -y httpd-tools
命令语法:ab -n(一次发送的请求数) -c(请求的并发数) 访问路径 注意:将Windows防火墙关闭
ab -n 5000 -c 100 http://192.168.200.1:8500/api/album/test/testLock
发现结果不是5000 ,说明本地锁是锁不住资源的;因此要使用分布式锁!
随着业务发展的需要,原单体单机部署的系统被演化成分布式集群系统后,由于分布式系统多线程、多进程并且分布在不同机器上,这将使原单机部署情况下的并发控制锁策略失效,单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问,这就是分布式锁要解决的问题!
分布式锁主流的实现方案:
每一种分布式锁解决方案都有各自的优缺点:
多个客户端同时获取锁(setnx)
获取成功,执行业务逻辑{从db获取数据,放入缓存,执行完成释放锁(del)}
其他客户端等待重试
/**
* 原理:执行业务方法前先尝试获取锁(setnx存入key val),如果获取锁成功再执行业务代码,业务执行完毕后将锁释放(del key)
*/
@Override
public void testLock() {
//0.先尝试获取锁 setnx key val
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("lock", "lock");
if(flag){
//获取锁成功,执行业务代码
//1.先从redis中通过key num获取值 key提前手动设置 num 初始值:0
String value = stringRedisTemplate.opsForValue().get("num");
//2.如果值为空则非法直接返回即可
if (StringUtils.isBlank(value)) {
return;
}
//3.对num值进行自增加一
int num = Integer.parseInt(value);
stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));
//4.将锁释放
stringRedisTemplate.delete("lock");
}else{
try {
Thread.sleep(100);
this.testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
进行测试:运行结果应该是5000。
可能出现的问题:
问题:setnx刚好获取到锁,业务逻辑出现异常,导致锁无法释放
解决:设置过期时间,自动释放锁。
将上锁代码改为:
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("lock", "lock", 3, TimeUnit.SECONDS);
测试结果应该是5000
问题:可能会释放其他服务器的锁。
场景:如果业务逻辑的执行时间是7s。执行流程如下
最终等于没锁的情况。
解决:setnx获取锁时,设置一个指定的唯一值(例如:uuid);释放前获取这个值,判断是否自己的锁。
修改代码:
@Override
public synchronized void testLock() {
// 声明uuid
String uuid = UUID.randomUUID().toString();
// 将uuid 存储到缓存中
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid, 3, TimeUnit.SECONDS);
if (flag){
try {
//1.从Redis缓存中获取key="num"的值 保证redis中存在"num"(手动提前在redis中创建key)
String value = stringRedisTemplate.opsForValue().get("num");
if (StringUtils.isBlank(value)) {
return;
}
//2.对获取到值进行+1操作
int num = Integer.parseInt(value);
stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));
// 判断uuid 是否与缓存中的数据相等
if (uuid.equals(stringRedisTemplate.opsForValue().get("lock"))){
// 如果相等,删除锁
stringRedisTemplate.delete("lock");
}
} catch (NumberFormatException e) {
e.printStackTrace();
}
}else {
try {
Thread.sleep(100);
this.testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
问题:删除操作缺乏原子性。
场景:
index1执行删除前,lock刚好过期时间已到,被Redis自动释放 在Redis中没有了锁。
index2获取了lock,index2线程获取到了cpu的资源,开始执行方法
index1执行删除,此时会把index2的lock删除
index1 因为已经在方法中了,所以不需要重新上锁。index1有执行的权限。index1已经比较完成了,这个时候,开始执行删除的是index2的锁!
解决方案:使用lua脚本保证删除锁具有原子性操作
/**
* 采用SpringDataRedis实现分布式锁
* 原理:执行业务方法前先尝试获取锁(setnx存入key val),如果获取锁成功再执行业务代码,业务执行完毕后将锁释放(del key)
*/
@Override
public void testLock() {
//0.先尝试获取锁 setnx key val
//问题:锁可能存在线程间相互释放
//Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("lock", "lock", 10, TimeUnit.SECONDS);
//解决:锁值设置为uuid
String uuid = UUID.randomUUID().toString();
Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("lock", uuid, 10, TimeUnit.SECONDS);
if(flag){
//获取锁成功,执行业务代码
//1.先从redis中通过key num获取值 key提前手动设置 num 初始值:0
String value = stringRedisTemplate.opsForValue().get("num");
//2.如果值为空则非法直接返回即可
if (StringUtils.isBlank(value)) {
return;
}
//3.对num值进行自增加一
int num = Integer.parseInt(value);
stringRedisTemplate.opsForValue().set("num", String.valueOf(++num));
//4.将锁释放 判断uuid
//问题:删除操作缺乏原子性。
//if(uuid.equals(stringRedisTemplate.opsForValue().get("lock"))){ //线程一:判断是满足是当前线程锁的值
// //条件满足,此时锁正好到期,redis锁自动释放了线程2获取锁成功,线程1将线程2的锁删除
// stringRedisTemplate.delete("lock");
//}
//解决:redis执行lua脚本保证原子,lua脚本执行会作为一个整体执行
//执行脚本参数 参数1:脚本对象封装lua脚本,参数二:lua脚本中需要key参数(KEYS[i]) 参数三:lua脚本中需要参数值 ARGV[i]
//4.1 先创建脚本对象 DefaultRedisScript泛型脚本语言返回值类型 Long 0:失败 1:成功
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
//4.2设置脚本文本
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.setScriptText(script);
//4.3 设置响应类型
redisScript.setResultType(Long.class);
stringRedisTemplate.execute(redisScript, Arrays.asList("lock"), uuid);
}else{
try {
//睡眠
Thread.sleep(100);
//自旋重试
this.testLock();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
总结:
为了确保分布式锁可用,我们至少要确保锁的实现同时满足以下四个条件:
Redisson是一个在Redis的基础上实现的Java驻内存数据网格(In-Memory Data Grid)。它不仅提供了一系列的分布式的Java常用对象,还提供了许多分布式服务。其中包括(BitSet, Set, Multimap, SortedSet, Map, List, Queue, BlockingQueue, Deque, BlockingDeque, Semaphore, Lock, AtomicLong, CountDownLatch, Publish / Subscribe, Bloom filter, Remote service, Spring cache, Executor service, Live Object service, Scheduler service) Redisson提供了使用Redis的最简单和最便捷的方法。Redisson的宗旨是促进使用者对Redis的关注分离(Separation of Concern),从而让使用者能够将精力更集中地放在处理业务逻辑上。
官方文档地址:https://github.com/Redisson/Redisson/wiki
实现步骤:
在service-util
模块中解开redisson相关依赖(注释放开)
<dependency>
<groupId>org.redisson</groupId>
<artifactId>redisson-spring-boot-starter</artifactId>
</dependency>
在service-util
模块中添加配置类
package com.atguigu.tingshu.common.config.redis;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;
import org.redisson.config.SingleServerConfig;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* redisson配置信息
*/
@Data
@Configuration
@ConfigurationProperties("spring.data.redis")
public class RedissonConfig {
private String host;
private String password;
private String port;
private int timeout = 3000;
private static String ADDRESS_PREFIX = "redis://";
/**
* 自动装配
*/
@Bean
RedissonClient redissonSingle() {
Config config = new Config();
if (StringUtils.isBlank(host)) {
throw new RuntimeException("host is empty");
}
SingleServerConfig serverConfig = config.useSingleServer()
.setAddress(ADDRESS_PREFIX + this.host + ":" + port)
.setTimeout(this.timeout);
if (StringUtils.isNotBlank(this.password)) {
serverConfig.setPassword(this.password);
}
return Redisson.create(config);
}
}
优化代码
package com.atguigu.tingshu.album.service.impl;
import com.alibaba.cloud.commons.lang.StringUtils;
import com.atguigu.tingshu.album.service.TestService;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
/**
* @author atguigu
* @ClassName TestServiceImpl
* @description: TODO
* @date 2023年08月22日
* @version: 1.0
*/
@Service
public class TestServiceImpl implements TestService {
@Autowired
private StringRedisTemplate stringRedisTemplate;
@Autowired
private RedissonClient redissonClient;
@Override
public void testLock() {
try {
//先尝试获取分布式锁 1.创建锁对象 2.调用获取锁方法
RLock lock = redissonClient.getLock("lock");
//获取锁成功后,才执行业务
lock.lock(); //阻塞到获取锁成功为止
//lock.tryLock(5, TimeUnit.SECONDS); //参数1:等待锁获取最大时间 参数2:时间单位 获取锁失败返回false
//lock.tryLock(3, 5, TimeUnit.SECONDS);//参数1:等待锁获取最大时间 参数2:锁过期时间,参数3:时间单位 获取锁失败返回false
try {
//1.从Redis缓存中获取key="num"的值 保证redis中存在"num"(手动提前在redis中创建key)
String value = redisTemplate.opsForValue().get("num");
if (StringUtils.isBlank(value)) {
return;
}
//2.对获取到值进行+1操作
int num = Integer.parseInt(value);
redisTemplate.opsForValue().set("num", String.valueOf(++num));
} finally {
//业务执行完毕释放锁
lock.unlock();
}
} catch (Exception e) {
e.printStackTrace();
}
}
}
分布式锁:
ab -n 5000 -c 100 http://192.168.200.1:8500/api/album/test/testLock
@Autowired
private RedisTemplate redisTemplate;
/**
* 根据专辑ID查询专辑信息-包含专辑标签列表
* 自定义分布式锁
*
* @param id
* @return
*/
@Override
public AlbumInfo getAlbumInfo(Long id) {
try {
//1.优先从缓存中获取数据,命中缓存直接返回,未命中则获取分布式锁
//1.1 创建缓存业务数据Key 前缀:数据ID标识
String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
//1.2 查询Redis缓存获取专辑信息
AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
if (albumInfo != null) {
return albumInfo;
}
//2.尝试获取分布式锁
//2.1 创建锁的Key
String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
//2.2 创建当前线程锁唯一标识UUID
String lockValue = IdUtil.randomUUID();
//2.3 尝试加锁 set k v ex nx
Boolean flag = redisTemplate.opsForValue().setIfAbsent(lockKey, lockValue, 5, TimeUnit.SECONDS);
//3.获取分布式锁成功:执行业务代码
if (flag) {
//3.1 由于前置代码中已有分布式布隆过滤器对查询专辑进行判断是否存在
try {
AlbumInfo albumInfoFromDB = this.getAlbumInfoFromDB(id);
long ttl = RedisConstant.ALBUM_TIMEOUT + RandomUtil.randomInt(0, 500);
redisTemplate.opsForValue().set(dataKey, albumInfoFromDB, ttl, TimeUnit.SECONDS);
return albumInfoFromDB;
} finally {
//4.业务执行完毕,释放锁 采用Lua脚本释放锁
String scriptText = "if redis.call(\"get\",KEYS[1]) == ARGV[1]\n" +
"then\n" +
" return redis.call(\"del\",KEYS[1])\n" +
"else\n" +
" return 0\n" +
"end";
//创建脚本对象-封装lua脚本
DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
redisScript.setScriptText(scriptText);
redisScript.setResultType(Long.class);
redisTemplate.execute(redisScript, Arrays.asList(lockKey), lockValue);
}
} else {
//5.获取分布式锁失败:自旋
Thread.sleep(100);
return this.getAlbumInfo(id);
}
} catch (Exception e) {
log.error("[专辑服务]查询专辑信息异常兜底处理:{}", e);
//兜底处理方案:当前Redis服务不可用(缓存,分布式锁),直接查询数据库
return this.getAlbumInfoFromDB(id);
}
}
/**
* 查询专辑统计信息(从数据中查询获取)
*
* @param id
* @return
*/
@Override
public AlbumInfo getAlbumInfoFromDB(Long id) {
//1.根据专辑ID查询专辑信息
AlbumInfo albumInfo = albumInfoMapper.selectById(id);
//2.根据专辑ID查询标签关系列表
LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
if (CollectionUtil.isNotEmpty(albumAttributeValueList)) {
albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
}
return albumInfo;
}
@Autowired
private RedissonClient redissonClient;
/**
* 根据专辑ID查询专辑信息(包含标签列表)
* 采用Redisson分布式锁
*
* @param id
* @return
*/
@Override
public AlbumInfo getAlbumInfo(Long id) {
try {
//1.优先从分布式缓存Redis中获取业务数据,命中缓存直接返回,未命中缓存尝试获取分布式锁
//1.1 构建业务数据Key 形式=前缀+数据标识ID
String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
//1.2 从缓存中获取业务数据
AlbumInfo albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
//1.3 命中缓存,直接返回结果
if (albumInfo != null) {
return albumInfo;
}
//2.尝试获取分布式锁,获取锁成功执行执行查库逻辑,业务数据放入缓存,释放锁;反之进行自旋
//2.1 构建当前业务ID对应锁Key 形式=前缀+数据标识ID:锁后缀
String lockKey = RedisConstant.ALBUM_INFO_PREFIX + id + RedisConstant.CACHE_LOCK_SUFFIX;
//2.2 创建锁对象
RLock lock = redissonClient.getLock(lockKey);
//2.3 获取分布式锁
lock.lock();
//3.业务逻辑,查询业务执行后释放锁
try {
//3.1 业务执行
albumInfo = this.getAlbumInfoFromDB(id);
int ttl = RandomUtil.randomInt(100, 1000);
redisTemplate.opsForValue().set(dataKey, albumInfo, RedisConstant.ALBUM_TIMEOUT + ttl, TimeUnit.SECONDS);
return albumInfo;
} finally {
//3.2 释放锁采用lua脚本释放
lock.unlock();
}
} catch (Exception e) {
//缓存服务有问题,兜底处理方法
log.error("[专辑服务]查询专辑信息异常:{}", e);
return this.getAlbumInfoFromDB(id);
}
}
/**
* 查询专辑统计信息(从数据中查询获取)
*
* @param id
* @return
*/
@Override
public AlbumInfo getAlbumInfoFromDB(Long id) {
//1.根据专辑ID查询专辑信息
AlbumInfo albumInfo = albumInfoMapper.selectById(id);
//2.根据专辑ID查询标签关系列表
LambdaQueryWrapper<AlbumAttributeValue> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(AlbumAttributeValue::getAlbumId, id);
List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueMapper.selectList(queryWrapper);
if (CollectionUtil.isNotEmpty(albumAttributeValueList)) {
albumInfo.setAlbumAttributeValueVoList(albumAttributeValueList);
}
return albumInfo;
}
专辑详情中多项数据接口需要查询缓存,那么分布式锁的业务逻辑代码就会出现大量的重复。因此,我们可以参考Spring框架事务注解来实现简化代码。只要类上添加了一个注解,那么这个注解就会自带分布式锁的功能。
实现如下:
service-util
模块中添加一个注解package com.atguigu.tingshu.common.cache;
import java.lang.annotation.*;
/**
* 自定义注解:在方法上使用该注解,将方法执行结果放入缓存同时采用分布式锁避免缓存击穿
* <p>
* 元注解:
*
* @Target:注解使用位置
* @Retention: 注解保留阶段(源码,字节码,运行)
* @Inherited: 注解是否可以被继承
* @Documented:产生java文档 javadoc命令生成API文档是否包含注解
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface GuiGuCache {
/**
* 存入Redis中键前缀(业务键,锁的键)
* @return
*/
String prefix() default "none:";
}
package com.atguigu.tingshu.common.cache;
import cn.hutool.core.util.RandomUtil;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.login.GuiGuLogin;
import com.atguigu.tingshu.common.result.ResultCodeEnum;
import com.atguigu.tingshu.common.util.AuthContextHolder;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import io.lettuce.core.RedisClient;
import jakarta.servlet.http.HttpServletRequest;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import org.springframework.web.context.request.RequestAttributes;
import org.springframework.web.context.request.RequestContextHolder;
import org.springframework.web.context.request.ServletRequestAttributes;
import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
/**
* 自定义缓存注解切面类
*
* @author: atguigu
* @create: 2024-05-09 10:46
*/
@Slf4j
@Aspect
@Component
public class GuiGuCacheAspect {
@Autowired
private RedisTemplate redisTemplate;
@Autowired
private RedissonClient redissonClient;
/**
* 环绕通知 对所有业务服务模块从api包下 使用@GuiGuLogin注解方法进行增强,从请求中获取请求头token,查询用户信息,校验,将用户ID存入ThreadLocal
*
* @param pjp 目标方法对象
* @param guiGuCache 作用在方法上自定义注解对象 获取到注解中属性值
* @return
* @throws Throwable
*/
@Around("@annotation(guiGuCache)")
public Object doBasicProfiling(ProceedingJoinPoint pjp, GuiGuCache guiGuCache) throws Throwable {
try {
log.info("[自定义缓存切面]前置逻辑...");
//1.优先从分布式缓存Redis中获取业务数据,命中缓存直接返回,未命中缓存尝试获取分布式锁
//1.1 构建业务数据Key 形式=缓存注解中前缀:方法参数1_方法参数2
//1.1.1 得到缓存注解前缀
String prefix = guiGuCache.prefix();
//1.1.2 获取目标方法参数值
String params = "none";
Object[] args = pjp.getArgs();
if (args != null && args.length > 0) {
params = Arrays.asList(args)
.stream()
.map(arg -> arg.toString())
.collect(Collectors.joining("_"));
}
String dataKey = prefix + params;
//1.2 从Redis缓存中查询业务数据
Object retVal = redisTemplate.opsForValue().get(dataKey);
//1.3 命中缓存返回即可,未命中则执行第二步
if (retVal != null) {
return retVal;
}
//2.尝试获取分布式锁,获取锁成功执行执行查库逻辑,业务数据放入缓存,释放锁;反之进行自旋(tryLock)
//2.1 构建锁的key 形式=业务数据key:锁后缀
String lockKey = dataKey + RedisConstant.CACHE_LOCK_SUFFIX;
//2.2 创建锁对象
RLock lock = redissonClient.getLock(lockKey);
//2.2 获取分布式锁
lock.lock();
//3.执行目标方法,业务逻辑(查询数据库中数据)
//3.1 使用自定义缓存注解方法
try {
retVal = pjp.proceed(pjp.getArgs());
log.info("[自定义缓存]后置逻辑...");
//3.2 将查询结果业务数据放入到缓存中
int ttl = RandomUtil.randomInt(100, 1000);
redisTemplate.opsForValue().set(dataKey, retVal, RedisConstant.ALBUM_TIMEOUT + ttl, TimeUnit.SECONDS);
return retVal;
} finally {
//4.将分布式锁释放
lock.unlock();
}
} catch (Throwable e) {
//缓存服务有问题,兜底处理方法
log.error("[自定义缓存]:兜底处理方法{}", e);
return pjp.proceed(pjp.getArgs());
}
}
}
在这个实现类AlbumInfoServiceImpl中添加注解:
/***
* 根据专辑ID查询专辑信息包含专辑标签列表
* @param id 专辑ID
* @return
*/
@Operation(summary = "根据专辑ID查询专辑信息包含专辑标签列表")
@GetMapping("/albumInfo/getAlbumInfo/{id}")
public Result<AlbumInfo> getAlbumInfo(@PathVariable Long id) {
//AlbumInfo albumInfo = albumInfoService.getAlbumInfo(id);
AlbumInfo albumInfo = albumInfoService.getAlbumInfoFromDB(id);
return Result.ok(albumInfo);
}
/***
* 根据专辑ID查询专辑信息包含专辑标签列表
* @param id 专辑ID
* @return
*/
@Override
@GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX)
public AlbumInfo getAlbumInfoFromDB(Long id) {...}
/**
* 根据专辑ID查询专辑统计信息
*
* @param albumId 专辑ID
* @return
*/
@Override
@GuiGuCache(prefix = RedisConstant.ALBUM_INFO_PREFIX+":stat:")
public AlbumStatVo getAlbumStatVo(Long albumId){...}
在BaseCategoryServiceImpl实现类中添加注解:
/**
* 根据一级分类Id获取分类标签以及标签值
*
* @param category1Id
* @return
*/
@Override
@GuiGuCache(prefix="category:baseAttribute:") //将该方法执行查库结果放入缓存,采用分布式锁杜绝缓存击穿
public List<BaseAttribute> getAttributeByCategory1Id(Long category1Id){}
@Override
@GuiGuCache(prefix = "category:categoryview:")
public BaseCategoryView getCategoryView(Long category3Id){}
/**
* 根据1级分类ID查询该分类下前7个置顶3级分类列表
*
* @param category1Id
* @return
*/
@Override
@GuiGuCache(prefix = "category:topcategory3:")
public List<BaseCategory3> getTopBaseCategory3(Long category1Id){}
/**
* 根据1级分类ID查询包含二级分类以及三级分类
*
* @param category1Id
* @return
*/
@Override
@GuiGuCache(prefix = "category:categoryby1Id:")
public JSONObject getBaseCategoryListByCategory1Id(Long category1Id){}
在UserInfoServiceImpl实现类中添加注解:
@Override
@GuiGuCache(prefix = "user:")
public UserInfoVo getUserInfo(Long userId) {}
布隆过滤器(Bloom Filter),是1970年,由一个叫布隆的小伙子提出的,距今已经五十年了。它实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都比一般的算法要好的多,缺点是有一定的误识别率和删除困难。二进制大家应该都清楚,存储的数据不是0就是1,默认是0。
主要用于判断一个元素是否在一个集合中,0代表不存在某个数据,1代表存在某个数据。
总结: 判断一个元素一定不存在 或者 可能存在! 存在一定的误判率{通过代码调节}
Bit 数组:
0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 | 0 |
---|---|---|---|---|---|---|---|---|
大数据量的时候, 判断一个元素是否在一个集合中。解决缓存穿透问题
**存入过程 **
布隆过滤器上面说了,就是一个二进制数据的集合。当一个数据加入这个集合时,经历如下:
通过K个哈希函数计算该数据,返回K个计算出的hash值
这些K个hash值映射到对应的K个二进制的数组下标
将K个下标对应的二进制数据改成1。
如图所示:
查询过程
布隆过滤器主要作用就是查询一个数据,在不在这个二进制的集合中,查询过程如下:
1、通过K个哈希函数计算该数据,对应计算出的K个hash值
2、通过hash值找到对应的二进制的数组下标
3、判断:如果存在一处位置的二进制数据是0,那么该数据不存在。如果都是1,该数据可能存在集合中。
优点
缺点
添加数据是通过计算数据的hash值,那么很有可能存在这种情况:两个不同的数据计算得到相同的hash值。
例如图中的“张三”和“张三丰”,假如最终算出hash值相同,那么他们会将同一个下标的二进制数据改为1。
这个时候,你就不知道下标为1的二进制,到底是代表“张三”还是“张三丰”。
一、存在误判
假如上面的图没有存 "张三",只存了 "张三丰",那么用"张三"来查询的时候,会判断"张三"存在集合中。
因为“张三”和“张三丰”的hash值是相同的,通过相同的hash值,找到的二进制数据也是一样的,都是1。
误判率:
受三个因素影响: 二进制位的个数m, 哈希函数的个数k, 数据规模n (添加到布隆过滤器中的函数)
已知误判率p, 数据规模n, 求二进制的个数m,哈希函数的个数k {m,k 程序会自动计算 ,你只需要告诉我数据规模,误判率就可以了}
ln: 自然对数是以常数e为底数的对数,记作lnN(N>0)。在物理学,生物学等自然科学中有重要的意义,一般表示方法为lnx。数学中也常见以logx表示自然对数。
二、删除困难
还是用上面的举例,因为“张三”和“张三丰”的hash值相同,对应的数组下标也是一样的。
如果你想去删除“张三”,将下标为1里的二进制数据,由1改成了0。
那么你是不是连“张三丰”都一起删了呀。
在service-search
模块的启动类中添加
package com.atguigu.tingshu;
import com.atguigu.tingshu.common.constant.RedisConstant;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.cache.annotation.EnableCaching;
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
import org.springframework.cloud.openfeign.EnableFeignClients;
import org.springframework.scheduling.annotation.EnableScheduling;
@Slf4j
@SpringBootApplication//(scanBasePackages = {"cn.atguigu", "com.atguigu"}) //默认扫描到启动类所在包,扫描当前模块以及引入的jar
@EnableDiscoveryClient
@EnableFeignClients
@EnableScheduling //开启定时任务
@EnableCaching //开启springCache
public class ServiceAlbumApplication implements CommandLineRunner {
public static void main(String[] args) {
SpringApplication.run(ServiceAlbumApplication.class, args);
}
@Autowired
private RedissonClient redissonClient;
/**
* CommandLineRunner接口springboot提供
* 当springboot应用启动后自动执行一次
*
* @param args
* @throws Exception
*/
@Override
public void run(String... args) throws Exception {
//1.初始化布隆过滤器 参数1:期望数据规模 参数2:误判率
log.info("初始化布隆过滤器");
//1.1 获取布隆过滤器对象
RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
//1.2 初始化布隆过滤器(存在不处理(前提:未扩容),不存在则执行初始化)
if (!bloomFilter.isExists()) {
bloomFilter.tryInit(10000, 0.03);
}
}
}
service-search
模块中SearchServiceImpl上架专辑方法upperAlbum
@Autowired
private RedissonClient redissonClient;
/**
* 构建专辑索引库文档对象,新增专辑到索引库
*
* @param albumId
*/
@Override
public void upperAlbum(Long albumId) {
//....省略代码
//将新增的商品SKUID存入布隆过滤器
//获取布隆过滤器,将新增skuID存入布隆过滤器
RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
bloomFilter.add(albumInfo.getId());
}
service-search
模块中ItemServiceImpl汇总专辑方法getItemInfo()
@Autowired
private RedissonClient redissonClient;
/**
* 汇总专辑详情相关信息
*
* @param albumId
* @return
*/
@Override
public Map<String, Object> getAlbumItem(Long albumId) {
//0.查询布隆过滤器是否包含查询专辑ID
RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
boolean flag = bloomFilter.contains(albumId);
if (!flag) {
throw new GuiguException(404, "访问专辑不存在");
}
//.....省略代码
}
采用定时任务机制进行布隆过滤器扩容
package com.atguigu.tingshu.album.job;
import cn.hutool.core.collection.CollectionUtil;
import com.atguigu.tingshu.album.mapper.AlbumInfoMapper;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.constant.SystemConstant;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.ibatis.ognl.OgnlRuntime;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.util.List;
import java.util.stream.Collectors;
/**
* @author: atguigu
* @create: 2024-05-21 09:21
*/
@Slf4j
@Component
public class BloomFilterResizeJob {
@Autowired
private RedissonClient redissonClient;
@Autowired
private AlbumInfoMapper albumInfoMapper;
/**
* 扩容定时任务:每月1号凌晨3点钟触发扩容
*/
//@Scheduled(cron = "0 0 3 1 * ?")
@Scheduled(cron = "0/5 * * * * ?")
public void resizeBloomFilter() {
//1.判断是否满足扩容条件 布隆过滤器元素数量>期望数据规模进行扩容
//1.1 获取现有布隆过滤对象
RBloomFilter<Long> oldBloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
//1.2 获取现有布隆过滤器各项信息(数量、误判率、数据规模)
long count = oldBloomFilter.count();
long expectedInsertions = oldBloomFilter.getExpectedInsertions();
//TODO 实际运行改为现有数据规模
if (count >= expectedInsertions) {
//2.新建布隆过滤器,初始化期望数据规模是原来的两倍
RBloomFilter<Long> newBloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER + ":new");
newBloomFilter.tryInit(expectedInsertions * 2, oldBloomFilter.getFalseProbability());
//3.将审核通过专辑ID保存新布隆过滤器
//3.1 查询现有审核通过专辑ID
LambdaQueryWrapper<AlbumInfo> queryWrapper = new LambdaQueryWrapper<>();
queryWrapper.eq(AlbumInfo::getStatus, SystemConstant.ALBUM_STATUS_PASS);
queryWrapper.select(AlbumInfo::getId);
List<AlbumInfo> list = albumInfoMapper.selectList(queryWrapper);
if (CollectionUtil.isNotEmpty(list)) {
//3.2 将专辑ID新增到新布隆过滤器
List<Long> albumIdList = list.stream().map(AlbumInfo::getId).collect(Collectors.toList());
for (Long albumId : albumIdList) {
newBloomFilter.add(albumId);
}
}
//4.删除旧的
oldBloomFilter.delete();
//5.对新布隆过滤器改名
newBloomFilter.rename(RedisConstant.ALBUM_BLOOM_FILTER);
}
}
}
先更新DB还是先更新缓存?是更新缓存还是删除缓存?在常规情况下,怎么操作都可以,但一旦存在高并发场景,就需要采用合适的方案。
1、先更新数据库再更新缓存(双写策略1)
线程A:更新数据库(第1s)——> 更新缓存(第10s)
线程B:更新数据库 (第3s)——> 更新缓存(第5s)
并发场景下,这样的情况是很容易出现的,每个线程的操作先后顺序不同,这样就导致请求B的缓存值被请求A给覆盖了,数据库中是线程B的新值,缓存中是线程A的旧值,并且会一直这么脏下去直到缓存失效(设置了过期时间)。
2、先更新缓存再更新数据库(双写策略2)
线程A:更新缓存(第1s)——> 更新数据库(第10s)
线程B: 更新缓存(第3s)——> 更新数据库(第5s)
和前面一种情况相反,缓存中是线程B的新值,而数据库中是线程A的旧值。
前两种方式之所以会在并发场景下出现异常,本质上是因为更新缓存和更新数据库是两个操作,我们没有办法控制并发场景下两个操作之间先后顺序,也就是先开始操作的线程先完成自己的工作。
3、先删除缓存再更新数据库
通过这种方式,我们很惊喜地发现,前面困扰我们的并发场景的问题确实被解决了!两个线程都只修改数据库,不管谁先,数据库以之后修改的线程为准。
但这个时候,我们来思考另一个场景:两个并发操作,一个是更新操作,另一个是查询操作,更新操作删除缓存后,查询操作没有命中缓存,先把老数据读出来后放到缓存中,然后更新操作更新了数据库。于是,在缓存中的数据还是老的数据,导致缓存中的数据是脏的。很显然,这种状况也不是我们想要的。
修改数据业务逻辑
1.删除缓存
2.更新数据库
3.睡眠一段时间(500ms)
4.再次删除缓存
加了个睡眠时间,主要是为了确保请求 A (读)在睡眠的时候,请求 B (写)能够在这这一段时间完成「从数据库读取数据,再把缺失的缓存写入缓存」的操作,然后请求 B 睡眠完,再删除缓存。
所以,请求 B 的睡眠时间就需要大于请求A「从数据库读取数据 + 写入缓存」的时间。
但是具体睡眠多久其实是个玄学,很难评估出来,所以这个方案也只是尽可能保证一致性而已,极端情况下,依然也会出现缓存不一致的现象。因此,还是不太建议这种方案。
对于蓝色的文字,“删除缓存 10”必须在“回写缓存10”后面,那如何才能保证一定是在后面呢?让请求 A 的最后一次删除,等待 500ms。
依赖“锁”的机制,避免出现并发读写。弊端:性能低
读数据方法(查询操作)
@Autowired
private RedissonClient redissonClient;
/**
* 读数据方法
*
* @param id
* @return
*/
@Override
public Long read(Long id) {
//1.创建读写锁对象 锁名称前缀:数据标识
RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("myLock:" + id);
RLock lock = readWriteLock.readLock();
//2.获取读锁成功 执行读操作 读锁持有期间,不允许并发发写,允许并发读
try {
lock.lock(5, TimeUnit.SECONDS);
log.info(Thread.currentThread().getName() + "执行读数据操作");
return id;
} finally {
//3.释放读锁
//lock.unlock();
}
}
修改数据方法(修改操作)
/**
* 写数据方法
*
* @param id
*/
@Override
public void write(Long id) {
//1.创建读写锁对象 锁名称前缀:数据标识
RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("myLock:" + id);
//2.获取写锁成功 执行写操作 写锁持有期间,不允许并发读,并发写
RLock lock = readWriteLock.writeLock();
lock.lock(5, TimeUnit.SECONDS);
try {
log.info(Thread.currentThread().getName() + "执行写入操作");
} finally {
//3.释放写锁
//lock.unlock();
}
}
「先更新数据库,再删缓存」的策略的第一步是更新数据库,那么更新数据库成功,就会产生一条变更日志,记录在 binlog 里。
于是我们就可以通过订阅 binlog 日志,拿到具体要操作的数据,然后再执行缓存删除,阿里巴巴开源的 Canal 中间件就是基于这个实现的。
Canal 模拟 MySQL 主从复制的交互协议,把自己伪装成一个 MySQL 的从节点,向 MySQL 主节点发送 dump 请求,MySQL 收到请求后,就会开始推送 Binlog 给 Canal,Canal 解析 Binlog 字节流之后,转换为便于读取的结构化数据,供下游程序订阅使用。
下图是 Canal 的工作原理:
所以,如果要想保证「先更新数据库,再删缓存」策略第二个操作能执行成功,我们可以使用「消息队列来重试缓存的删除」,或者「订阅 MySQL binlog 再操作缓存」,这两种方法有一个共同的特点,都是采用异步操作缓存。
验证MySQL是否开启BinLog确保Value为 ON(已完成)
show variables like '%log_bin%';
在MySQL新增用户用于监听Binlog日志,在Canal容器中要使用该用户(已完成)
#创建用户
CREATE USER canal IDENTIFIED BY '123456';
#给用户授权
GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
#如果是MySQL8.X以上需要对加密方式进行设置
ALTER USER 'canal'@'%' IDENTIFIED WITH mysql_native_password BY '123456';
#刷新生效
FLUSH PRIVILEGES;
查询MySQL主节点状态;完成重置
SHOW MASTER STATUS;
reset master;
采用Docker方式创建Canal服务端。以下为创建Canal容器命令,要修改要监听主MySQL数据库IP跟端口用户名以及密码确保正确(不需要执行)
docker run -p 11111:11111 --name canal \
-e canal.destinations=tingshuTopic \
-e canal.instance.master.address=192.168.200.6:3306 \
-e canal.instance.dbUsername=canal \
-e canal.instance.dbPassword=123456 \
-e canal.instance.connectionCharset=UTF-8 \
-e canal.instance.tsdb.enable=true \
-e canal.instance.gtidon=false \
-e canal.instance.filter.regex=.*\\..* \
-d canal/canal-server:v1.1.5
重新创建canal容器,促使canal重新跟MySQL建立连接,从MySQL最新BinLog偏移量开始读取数据
Canal日志在容器
/home/admin/canal-server/logs/tingshuTopic 日志路径
目前不支持:JDK17
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<groupId>com.atguigu</groupId>
<artifactId>service-cdc</artifactId>
<version>1.0-SNAPSHOT</version>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.3.6.RELEASE</version>
</parent>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!--web 需要启动项目-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>top.javatool</groupId>
<artifactId>canal-spring-boot-starter</artifactId>
<version>1.2.1-RELEASE</version>
</dependency>
<dependency>
<groupId>javax.persistence</groupId>
<artifactId>persistence-api</artifactId>
<version>1.0</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>
</dependencies>
</project>
file->settings
提供启动类
package com.atguigu.tingshu;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @author: atguigu
* @create: 2023-11-01 16:39
*/
@SpringBootApplication
public class CDCApplicaiton {
public static void main(String[] args) {
SpringApplication.run(CDCApplicaiton.class, args);
}
}
application.yml
spring:
profiles:
active: dev
application-dev.yaml
#canal配置
canal:
destination: tingshuTopic #Canal服务端发送数据的话题名称跟上面容器里参数destinations的一样
server: 192.168.200.6:11111
spring:
redis:
host: 192.168.200.6
port: 6379
提供Java实体类类监听变更后的数据,注意属性上使用@Column注解进行映射
package com.atguigu.tingshu.model;
import lombok.Data;
import javax.persistence.Column;
/**
*
* @author: atguigu
* @create: 2023-11-01 16:22
*/
@Data
public class CDCEntity {
@Column(name = "id")
private Long id;
}
监听到变更业务处理类
package com.atguigu.tingshu.handler;
import com.atguigu.tingshu.model.CDCEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import top.javatool.canal.client.annotation.CanalTable;
import top.javatool.canal.client.handler.EntryHandler;
/**
* @author: atguigu
* @create: 2023-12-23 10:53
*/
@Slf4j
@Component
@CanalTable("user_info") ////监听变更表
public class UserCdcHandler implements EntryHandler<CDCEntity> {
@Autowired
private RedisTemplate redisTemplate;
/**
* 监听到数据修改
*
* @param before 变更前数据
* @param after 变更后数据
*/
@Override
public void update(CDCEntity before, CDCEntity after) {
log.info("监听到数据修改,ID:{}", after.getId());
String key = "userInfoVo:" + after.getId();
redisTemplate.delete(key);
}
/**
* 监听到删除操作
*
* @param cdcEntity 删除前数据
*/
@Override
public void delete(CDCEntity cdcEntity) {
log.info("监听到数据删除,ID:{}", cdcEntity.getId());
String key = "userInfoVo:" + cdcEntity.getId();
redisTemplate.delete(key);
}
}
提供Redis自定义模板对象设置key,val序列化器
package com.atguigu.tingshu.config;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
/**
* @author: atguigu
* @create: 2023-12-23 10:59
*/
@Configuration
public class RedisConfig {
@Bean
@Primary
public RedisTemplate<Object, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
RedisTemplate<Object, Object> redisTemplate = new RedisTemplate<>();
redisTemplate.setConnectionFactory(redisConnectionFactory);
//String的序列化方式
StringRedisSerializer stringRedisSerializer = new StringRedisSerializer();
// 使用GenericJackson2JsonRedisSerializer 替换默认序列化(默认采用的是JDK序列化)
GenericJackson2JsonRedisSerializer genericJackson2JsonRedisSerializer = new GenericJackson2JsonRedisSerializer();
//序列号key value
redisTemplate.setKeySerializer(stringRedisSerializer);
redisTemplate.setValueSerializer(genericJackson2JsonRedisSerializer);
redisTemplate.setHashKeySerializer(stringRedisSerializer);
redisTemplate.setHashValueSerializer(genericJackson2JsonRedisSerializer);
redisTemplate.afterPropertiesSet();
return redisTemplate;
}
}
tips:如果启动canal客户端长时间未监听到变更数据,做以下操作