day11

  重建布隆过滤器；Canal客户端解决一致性问题

common/service-util/src/main/java/com/atguigu/tingshu/common/config/redis/RedissonConfig.java

@@ -42,6 +42,8 @@ public class RedissonConfig {
     RedissonClient redissonSingle() {
         Config config = new Config();
 
+        //修改默认看门狗/释放锁锁时间   默认加锁成功后：锁有效10s  底层看门狗执行定时任务延迟时间:10/3s
+        config.setLockWatchdogTimeout(10*1000);
         if (StringUtils.isBlank(host)) {
             throw new RuntimeException("host is  empty");
         }

common/service-util/src/main/java/com/atguigu/tingshu/common/interceptor/FeignInterceptor.java

@@ -14,7 +14,7 @@ public class FeignInterceptor implements RequestInterceptor {
     public void apply(RequestTemplate requestTemplate){
         //  获取请求对象
         RequestAttributes requestAttributes = RequestContextHolder.getRequestAttributes();
-        //异步编排 与 MQ消费者端 为 null
+        //异步编排 定时任务 与 MQ消费者端 为 null
         if(null != requestAttributes) {
             ServletRequestAttributes servletRequestAttributes = (ServletRequestAttributes)requestAttributes;
             HttpServletRequest request = servletRequestAttributes.getRequest();

service/service-album/src/main/java/com/atguigu/tingshu/ServiceAlbumApplication.java

@@ -41,7 +41,7 @@ public class ServiceAlbumApplication implements CommandLineRunner {
         RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
         //2.如果布隆过滤器不存在则创建，完成初始化
         if (!bloomFilter.isExists()) {
-            bloomFilter.tryInit(100000L, 0.03);
+            bloomFilter.tryInit(10000L, 0.03);
         }
     }
 }

service/service-album/src/main/java/com/atguigu/tingshu/album/api/TestController.java

@@ -1,6 +1,8 @@
 package com.atguigu.tingshu.album.api;
 
+import com.atguigu.tingshu.album.service.TestService;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cache.annotation.CacheEvict;
 import org.springframework.cache.annotation.Cacheable;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -14,11 +16,12 @@ import org.springframework.web.bind.annotation.RestController;
  */
 @Slf4j
 @RestController
-@RequestMapping("/api/test")
+@RequestMapping("/api/album/test")
 public class TestController {
 
     /**
      * SpringCache 将方法上缓存注解参数作为key 方法结果作为Value
+     *
      * @param id
      * @return
      */
@@ -31,11 +34,30 @@ public class TestController {
 
     /**
      * 情况缓存
+     *
      * @param id
      */
     @CacheEvict(value = "userCache", key = "#id")
     @GetMapping("/update/{id}")
-    public void updateUser(@PathVariable("id") Long id){
+    public void updateUser(@PathVariable("id") Long id) {
         log.info("根据ID更新取业务数据");
     }
+
+    @Autowired
+    private TestService testService;
+
+    @GetMapping("/read/{id}")
+    public String read(@PathVariable long id) {
+        String result = testService.read(id);
+        return "读数据成功：" + result;
+    }
+
+
+    @GetMapping("/write/{id}")
+    public String write(@PathVariable long id) {
+        testService.write(id);
+        return "写数据成功";
+    }
+
+
 }

service/service-album/src/main/java/com/atguigu/tingshu/album/service/TestService.java

@@ -0,0 +1,7 @@
+package com.atguigu.tingshu.album.service;
+
+public interface TestService {
+    String read(long id);
+
+    void write(long id);
+}

service/service-album/src/main/java/com/atguigu/tingshu/album/service/impl/AlbumInfoServiceImpl.java

@@ -278,6 +278,7 @@ public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo
         albumInfo.setId(id);
         baseMapper.updateById(albumInfo);
 
+
         //2.更新专辑标签关系表
         //2.1 根据专辑ID删除原有专辑标签关系
         albumAttributeValueMapper.delete(
@@ -295,6 +296,7 @@ public class AlbumInfoServiceImpl extends ServiceImpl<AlbumInfoMapper, AlbumInfo
                     }).collect(Collectors.toList());
             //批量保存
             albumAttributeValueService.saveBatch(albumAttributeValueList);
+
         }
 
         //4.对专辑中文本内容进行审核 异步审核

service/service-album/src/main/java/com/atguigu/tingshu/album/service/impl/TestServiceImpl.java

@@ -0,0 +1,59 @@
+package com.atguigu.tingshu.album.service.impl;
+
+import com.atguigu.tingshu.album.service.TestService;
+import lombok.extern.slf4j.Slf4j;
+import org.redisson.api.RLock;
+import org.redisson.api.RReadWriteLock;
+import org.redisson.api.RedissonClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author: atguigu
+ * @create: 2025-03-21 11:31
+ */
+@Slf4j
+@Service
+public class TestServiceImpl implements TestService {
+
+    @Autowired
+    private RedissonClient redissonClient;
+
+
+    @Override
+    public String read(long id) {
+        //1.获取读写锁对象
+        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("lock:" + id);
+        //2.获取读锁
+        RLock lock = readWriteLock.readLock();
+        lock.lock(5, TimeUnit.SECONDS);
+        try {
+            //3.执行读数据业务
+            log.info("读锁加锁成功,执行读数据业务");
+            return "result:" + id;
+        } finally {
+            //4.释放读锁  TODO 故意不释放锁，等5后自动释放
+            //lock.unlock();
+        }
+    }
+
+    @Override
+    public void write(long id) {
+        //1.获取读写锁对象
+        RReadWriteLock readWriteLock = redissonClient.getReadWriteLock("lock:" + id);
+        RLock lock = readWriteLock.writeLock();
+
+        //2.获取写锁
+        lock.lock(5, TimeUnit.SECONDS);
+
+        //3.执行写数据业务
+        try {
+            log.info("写锁加锁成功,执行写数据业务");
+        } finally {
+            //4.释放写锁 TODO 故意不释放锁，等5后自动释放
+            //lock.unlock();
+        }
+    }
+}

service/service-album/src/main/java/com/atguigu/tingshu/album/task/RebuildBloomFilterTask.java

@@ -0,0 +1,91 @@
+package com.atguigu.tingshu.album.task;
+
+import cn.hutool.core.collection.CollUtil;
+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.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;
+
+/**
+ * @author: atguigu
+ * @create: 2025-03-21 09:21
+ */
+@Slf4j
+@Component
+public class RebuildBloomFilterTask {
+
+    @Autowired
+    private AlbumInfoMapper albumInfoMapper;
+
+    @Autowired
+    private RedissonClient redissonClient;
+
+    /**
+     * 重建布隆过滤器 TODO 后续改为xxl-job 定时+手动调度
+     * 周期:每月1号凌晨2点执行
+     */
+    //@Scheduled(cron = "0/10 * * * * ?")
+    @Scheduled(cron = "0 0 2 1 * ?")
+    public void rebuildBloomFilter() {
+        //1.处理扩容重建情况：布隆过滤器中元素数量大于期望数据规模，造成误判率就会上升
+        //1.1 获取"旧"的布隆过滤器对象 得到：期望数据规模、现有元素数量
+        RBloomFilter<Long> oldBloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
+        long expectedInsertions = oldBloomFilter.getExpectedInsertions();
+        long count = oldBloomFilter.count();
+        double falseProbability = oldBloomFilter.getFalseProbability();
+        log.info("[专辑服务]触发重建布隆过滤器任务：现有期望数据规模：{}，现有实际数据量：{}", expectedInsertions, count);
+        //1.2 如果满足实际元素数量大于期望数据量，需要扩容
+        if (count >= expectedInsertions) {
+            log.info("[专辑服务]布隆过滤器扩容");
+            //1.2.1 创建/初始化"新"的布隆过滤器 名称="旧":new  期望数据规模：原来的2倍
+            RBloomFilter<Long> newBloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER + ":new");
+            newBloomFilter.tryInit(expectedInsertions * 2, falseProbability);
+            //1.2.2 将过审专辑ID加入到"新"布隆过滤器内
+            this.addDataToBloomFilter(newBloomFilter);
+            //1.2.3 删除"旧"布隆过滤器 删除配置以及产生位图
+            oldBloomFilter.delete();
+            //1.2.4 对"新"布隆过滤器重命名 改为"旧"的名称，此时数据已经新的了
+            newBloomFilter.rename(RedisConstant.ALBUM_BLOOM_FILTER);
+        } else {
+            log.info("[专辑服务]布隆过滤器重建");
+            //2.不满足扩容条件：重建布隆过滤器.存在一些专辑被删除，被下架，被删除专辑ID仍然存在于布隆过滤器中。
+            //2.1 删除原有布隆过滤器，重新初始化新的
+            oldBloomFilter.delete();
+            RBloomFilter<Long> newBloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
+            newBloomFilter.tryInit(expectedInsertions, falseProbability);
+            //2.2 将过审专辑ID加入到布隆过滤器中
+            this.addDataToBloomFilter(newBloomFilter);
+        }
+    }
+
+
+    /**
+     * 将审核通过专辑ID加入到布隆过滤器
+     *
+     * @param bloomFilter
+     */
+    private void addDataToBloomFilter(RBloomFilter<Long> bloomFilter) {
+        //1.根据审核条件查询过审专辑ID 得到专辑ID列表
+        List<AlbumInfo> albumInfoList = albumInfoMapper
+                .selectList(
+                        new LambdaQueryWrapper<AlbumInfo>().eq(AlbumInfo::getStatus, SystemConstant.ALBUM_STATUS_PASS)
+                                .select(AlbumInfo::getId)
+                );
+        if (CollUtil.isNotEmpty(albumInfoList)) {
+            //2.循环将专辑ID加入到布隆过滤器
+            albumInfoList
+                    .stream()
+                    .map(AlbumInfo::getId)
+                    .forEach(bloomFilter::add);
+        }
+    }
+}

service/service-album/src/main/resources/application.yml

@@ -1,3 +1,5 @@
 spring:
   cache:
     type: redis # 缓存类型
+server:
+  port: 8501

service/service-album/src/test/java/com/atguigu/tingshu/RedissonLockTest.java

@@ -7,6 +7,8 @@ import org.redisson.api.RedissonClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
 
+import java.util.concurrent.TimeUnit;
+
 /**
  * @author: atguigu
  * @create: 2025-03-19 10:27
@@ -37,13 +39,13 @@ public class RedissonLockTest {
         //if (flag) {
             try {
                 log.info(Thread.currentThread().getName()+"加锁成功,执行业务逻辑,查询数据库");
-                //TimeUnit.SECONDS.sleep(100);
-                lock.lock();
-                try {
-                    System.out.println(Thread.currentThread().getName()+"再次加锁成功,执行业务");
-                } finally {
-                    lock.unlock();
-                }
+                TimeUnit.SECONDS.sleep(100);
+                //lock.lock();
+                //try {
+                //    System.out.println(Thread.currentThread().getName()+"再次加锁成功,执行业务");
+                //} finally {
+                //    lock.unlock();
+                //}
 
 
                 //tod 执行业务代码

service/service-cdc/pom.xml

@@ -0,0 +1,51 @@
+<?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>

service/service-cdc/src/main/java/com/atguigu/tingshu/CDCApplicaiton.java

@@ -0,0 +1,16 @@
+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);
+    }
+}

service/service-cdc/src/main/java/com/atguigu/tingshu/config/RedisConfig.java

@@ -0,0 +1,38 @@
+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;
+    }
+}

service/service-cdc/src/main/java/com/atguigu/tingshu/handler/AlbumInfoCdcHandler.java

@@ -0,0 +1,38 @@
+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("album_info") ////监听变更表
+public class AlbumInfoCdcHandler implements EntryHandler<CDCEntity> {
+
+    @Autowired
+    private RedisTemplate redisTemplate;
+
+    /**
+     * 监听用户表更新/逻辑删除，删除Redis缓存用户信息
+     *
+     * @param before
+     * @param after
+     */
+
+    @Override
+    public void update(CDCEntity before, CDCEntity after) {
+        log.info("监听到专辑表更新/逻辑删除，删除Redis缓存专辑信息");
+        Long id = after.getId();
+        String redisKey = "album:info:" + id;
+        redisTemplate.delete(redisKey);
+    }
+
+}

service/service-cdc/src/main/java/com/atguigu/tingshu/handler/UserCdcHandler.java

@@ -0,0 +1,38 @@
+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;
+
+    /**
+     * 监听用户表更新/逻辑删除，删除Redis缓存用户信息
+     *
+     * @param before
+     * @param after
+     */
+
+    @Override
+    public void update(CDCEntity before, CDCEntity after) {
+        log.info("监听到用户表更新/逻辑删除，删除Redis缓存用户信息");
+        Long id = after.getId();
+        String redisKey = "user:userinfo:" + id;
+        redisTemplate.delete(redisKey);
+    }
+
+}

service/service-cdc/src/main/java/com/atguigu/tingshu/model/CDCEntity.java

@@ -0,0 +1,17 @@
+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;
+}

service/service-cdc/src/main/resources/application-dev.yaml

@@ -0,0 +1,8 @@
+#canal配置
+canal:
+  server: 192.168.200.6:11111
+  destination: tingshuTopic #Canal服务端发送数据的话题名称跟上面容器里参数destinations的一样
+spring:
+  redis:
+    host: 192.168.200.6
+    port: 6379

service/service-cdc/src/main/resources/application.yml

@@ -0,0 +1,3 @@
+spring:
+  profiles:
+    active: dev

service/service-cdc/src/main/resources/logback-spring.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+
+    <contextName>logback</contextName>
+
+    <!-- 日志的输出目录 -->
+    <property name="log.path" value="D://work//tingshu_work//logs" />
+
+    <!--控制台日志格式：彩色日志-->
+    <!-- magenta:洋红 -->
+    <!-- boldMagenta:粗红-->
+    <!-- cyan:青色 -->
+    <!-- white:白色 -->
+    <!-- magenta:洋红 -->
+    <property name="CONSOLE_LOG_PATTERN"
+              value="%yellow(%date{yyyy-MM-dd HH:mm:ss}) %highlight([%-5level]) %green(%logger) %msg%n"/>
+
+    <!--文件日志格式-->
+    <property name="FILE_LOG_PATTERN"
+              value="%date{yyyy-MM-dd HH:mm:ss} [%-5level] %thread %file:%line %logger %msg%n" />
+
+    <!--编码-->
+    <property name="ENCODING" value="UTF-8" />
+
+    <!-- 控制台日志 -->
+    <appender name="CONSOLE" class="ch.qos.logback.core.ConsoleAppender">
+        <!-- 临界值过滤器 -->
+        <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
+            <level>INFO</level>
+        </filter>
+        <encoder>
+            <pattern>${CONSOLE_LOG_PATTERN}</pattern>
+            <charset>${ENCODING}</charset>
+        </encoder>
+    </appender>
+
+    <!-- 文件日志 -->
+    <appender name="FILE" class="ch.qos.logback.core.FileAppender">
+        <file>${log.path}//log.log</file>
+        <append>true</append>
+        <encoder>
+            <pattern>%date{yyyy-MM-dd HH:mm:ss} %msg%n</pattern>
+            <charset>${ENCODING}</charset>
+        </encoder>
+    </appender>
+
+
+    <!-- 开发环境 -->
+    <springProfile name="dev">
+        <!-- com.atguigu日志记录器：业务程序INFO级别  -->
+        <logger name="com.atguigu" level="INFO" />
+        <logger name="top.javatool" level="WARN" />
+        <!-- 根日志记录器：INFO级别  -->
+        <root level="INFO">
+            <appender-ref ref="CONSOLE" />
+        </root>
+    </springProfile>
+
+</configuration>

service/service-user/src/main/java/com/atguigu/tingshu/user/api/UserInfoApiController.java

@@ -1,6 +1,7 @@
 package com.atguigu.tingshu.user.api;
 
 import com.atguigu.tingshu.common.cache.GuiGuCache;
+import com.atguigu.tingshu.common.login.GuiGuLogin;
 import com.atguigu.tingshu.common.result.Result;
 import com.atguigu.tingshu.user.service.UserInfoService;
 import com.atguigu.tingshu.vo.user.UserInfoVo;
@@ -38,6 +39,7 @@ public class UserInfoApiController {
      * @param needCheckBuyStateTrackIds
      * @return
      */
+    @GuiGuLogin
     @Operation(summary = "根据专辑ID+声音ID列表查询得到声音购买情况")
     @PostMapping("/userInfo/userIsPaidTrack/{userId}/{albumId}")
     public Result<Map<Long, Integer>> userIsPaidTrack(

service/service-user/src/main/java/com/atguigu/tingshu/user/service/impl/UserInfoServiceImpl.java

@@ -134,12 +134,15 @@ public class UserInfoServiceImpl extends ServiceImpl<UserInfoMapper, UserInfo> i
      */
     @Override
     public void updateUser(UserInfoVo userInfoVo) {
+        //1.TODO 删除缓存
         Long userId = AuthContextHolder.getUserId();
         UserInfo userInfo = baseMapper.selectById(userId);
         //注意：只允许修改部分信息
         userInfo.setNickname(userInfoVo.getNickname());
         userInfo.setAvatarUrl(userInfoVo.getAvatarUrl());
         baseMapper.updateById(userInfo);
+        //3.TODO 睡眠一段时间 确保并发读线程执行完毕，将"脏"数据写回Redis
+        //4.TODO 再删除一次缓存
     }
 
     /**