谷粒随享

第6章详情优化

学习目标：

微服务分布式链路追踪组件应用（基础组件）
针对于相对高并发业务场景进行性能优化
- 采用分布式缓存Redis
- 缓存击穿（分布式锁）
- 缓存穿透（分布式布隆过滤器）
- 缓存雪崩（分布式集群）
- 缓存一致性（基于MySQL-BinLog数据最终一致）
- 采用异步任务+线程池
- 代码优化：利用AOP将冗余代码抽取-提供自定义缓存注解

1、链路追踪Zipkin

一个看起来很简单的应用，背后可能需要数十或数百个服务来支撑，一个请求就要多次服务调用。当请求变慢、或者不能使用时，我们是不知道是哪个后台服务引起的。这时，我们使用 Zipkin 就能解决这个问题。由于业务访问量的增大，业务复杂度增加，以及微服务架构和容器技术的兴起，要对系统进行各种拆分。微服务系统拆分后，我们可以使用 Zipkin 链路，来快速定位追踪有故障的服务点。

Zipkin/SkyWalking 是一款开源的分布式实时数据追踪系统（Distributed Tracking System），能够收集服务间调用的时序数据，提供调用链路的追踪。

Zipkin 其主要功能是聚集来自各个异构系统的实时监控数据，在微服务架构下，十分方便地用于服务响应延迟等问题的定位。

Zipkin 每一个调用链路通过一个 trace id 来串联起来，只要你有一个 trace id ，就能够直接定位到这次调用链路，并且可以根据服务名、标签、响应时间等进行查询，过滤那些耗时比较长的链路节点。

Zipkin 分布式跟踪系统就能非常好地解决该问题，主要解决以下3点问题：

1. 动态展示服务的链路；
1. 分析服务链路的瓶颈并对其进行调优；
1. 快速进行服务链路的故障发现。

在优化前通过Zipkin链路追踪进行查看接口耗时

在service微服务父工程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%

访问专辑详情接口、或查看专辑页面进行测试

http://localhost:8502/doc.html#/web-api/%E4%B8%93%E8%BE%91%E8%AF%A6%E6%83%85%E7%AE%A1%E7%90%86/getItem

通过Zipkin管理页面查看接口耗时

2、详情优化

虽然咱们实现了页面需要的功能，但是考虑到该页面是被用户高频访问的，所以性能需要优化。一般一个系统最大的性能瓶颈，就是数据库的io操作。从数据库入手也是调优性价比最高的切入点。

一般分为两个层面：

一是提高数据库sql本身的性能
- 单表
- 全值匹配我最爱
- 最左前缀要遵守
- 带头大哥不能死
- 中间兄弟不能断
- 索引列上少计算
- LIKE符号写最右
- 多表
- 小表（驱动表）驱动大表（被驱动表，关联字段，查询条件，上建立索引）
- 关联字段类型一致
- 表编码一致
二是尽量避免直接查询数据库。

重点要讲的是第二个层面：尽量避免直接查询数据库。

解决办法就是：分布式缓存

2.1 缓存常见问题

缓存最常见的3个问题：

缓存穿透
缓存雪崩
缓存击穿

缓存穿透: 是指查询一个不存在的数据，由于缓存无法命中，将去查询数据库，但是数据库也无此记录，并且出于容错考虑，我们没有将这次查询的null写入缓存，这将导致这个不存在的数据每次请求都要到存储层去查询，失去了缓存的意义。在流量大时，可能DB就挂掉了，要是有人利用不存在的key频繁攻击我们的应用，这就是漏洞。

解决1 ：空结果也进行缓存，但它的过期时间会很短，最长不超过五分钟，但是不能防止随机穿透。
解决2 ：使用布隆过滤器来解决随机穿透问题。

缓存雪崩:是指在我们设置缓存时采用了相同的过期时间，导致缓存在某一时刻同时失效，请求全部转发到DB，DB瞬时压力过重雪崩。

解决1：原有的失效时间基础上增加一个随机值，比如1-5分钟随机，这样每一个缓存的过期时间的重复率就会降低，就很难引发集体失效的事件。
解决2：如果单节点宕机，可以采用集群部署方式防止雪崩

缓存击穿: 是指对于一些设置了过期时间的key，如果这些key可能会在某些时间点被超高并发地访问，是一种非常“热点”的数据。这个时候，需要考虑一个问题：如果这个key在大量请求同时进来之前正好失效，那么所有对这个key的数据查询都落到db，我们称为缓存击穿。

与缓存雪崩的区别：

击穿是一个热点key失效
雪崩是很多key集体失效

解决：锁

2.2 缓存击穿解决方案

2.2.1 本地锁的局限性

之前，我们学习过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 ，说明本地锁是锁不住资源的；因此要使用分布式锁！

2.2.2 分布式锁

随着业务发展的需要，原单体单机部署的系统被演化成分布式集群系统后，由于分布式系统多线程、多进程并且分布在不同机器上，这将使原单机部署情况下的并发控制锁策略失效，单纯的Java API并不能提供分布式锁的能力。为了解决这个问题就需要一种跨JVM的互斥机制来控制共享资源的访问，这就是分布式锁要解决的问题！

分布式锁主流的实现方案：

基于数据库实现分布式锁
基于缓存（ Redis等）
基于Zookeeper

每一种分布式锁解决方案都有各自的优缺点：

性能：Redis最高
可靠性：Zookeeper最高(CAP定理-CP)

2.2.3 使用redis脚本实现分布式锁

2.2.3.1 使用setnx 命令实现

多个客户端同时获取锁（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刚好获取到锁，业务逻辑出现异常，导致锁无法释放

解决：设置过期时间，自动释放锁。

2.2.3.2 使用setex给锁设置过期时间

首先想到通过expire设置过期时间（缺乏原子性：如果在setnx和expire之间出现异常，锁也无法释放）
在set时指定过期时间（推荐）

将上锁代码改为：

Boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("lock", "lock", 3, TimeUnit.SECONDS);

测试结果应该是5000

问题：可能会释放其他服务器的锁。

场景：如果业务逻辑的执行时间是7s。执行流程如下

index1业务逻辑没执行完，3秒后锁被自动释放。
index2获取到锁，执行业务逻辑，3秒后锁被自动释放。
index3获取到锁，执行业务逻辑
1. index1业务逻辑执行完成，开始调用del释放锁，这时释放的是index3的锁，导致index3的业务只执行1s就被别人释放。

最终等于没锁的情况。

解决：setnx获取锁时，设置一个指定的唯一值（例如：uuid）；释放前获取这个值，判断是否自己的锁。

2.3.3.3 使用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值确实和uuid相等
index1执行删除前，lock刚好过期时间已到，被Redis自动释放在Redis中没有了锁。
index2获取了lock,index2线程获取到了cpu的资源，开始执行方法
index1执行删除，此时会把index2的lock删除

index1 因为已经在方法中了，所以不需要重新上锁。index1有执行的权限。index1已经比较完成了，这个时候，开始执行删除的是index2的锁！

解决方案：使用lua脚本保证删除锁具有原子性操作

2.3.3.4 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();
        }
    }
}

总结：

为了确保分布式锁可用，我们至少要确保锁的实现同时满足以下四个条件：

互斥性。在任意时刻，只有一个客户端能持有锁。
不会发生死锁。即使有一个客户端在持有锁的期间崩溃而没有主动解锁，也能保证后续其他客户端能加锁。
解铃还须系铃人。加锁和解锁必须是同一个客户端，客户端自己不能把别人加的锁给解了。
加锁和解锁必须具有原子性。

2.2.4 使用 Redisson 解决分布式锁

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

2.3 分布式锁整合业务

2.3.1 自定义分布式锁整合业务

@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;
}

2.3.2 使用redisson整合分布式锁

@Autowired
private RedissonClient redissonClient;

/**
 * 根据专辑ID查询专辑信息-包含专辑标签列表
 * Redisson分布式锁
 *
 * @param id 1(100线程) 2（100线程）
 * @return
 */
@Override
public AlbumInfo getAlbumInfo(Long id) {
    try {
        //1.优先从缓存中获取数据，命中返回直接返回，未命中获取分布式锁
        //1.1 创建缓存业务数据Key 形式：前缀+数据ID标识
        String dataKey = RedisConstant.ALBUM_INFO_PREFIX + id;
        //1.2 查询缓存中专辑业务数据
        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 创建锁对象
        RLock lock = redissonClient.getLock(lockKey);
        //2.3 执行加锁方法
        lock.lock();

        //3.执行业务（查询数据库业务）
        try {
            //3.1 只要线程调用lock方法最终都会获取锁成功，避免后来线程获取锁成功造成多次查库，再查一次缓存
            albumInfo = (AlbumInfo) redisTemplate.opsForValue().get(dataKey);
            if (albumInfo != null) {
                return albumInfo;
            }
            //3.2 执行查询数据库方法 将查询结果放入缓存
            albumInfo = this.getAlbumInfoFromDB(id);
            long ttl = RedisConstant.ALBUM_TIMEOUT + RandomUtil.randomInt(0, 500);
            redisTemplate.opsForValue().set(dataKey, albumInfo, ttl, TimeUnit.SECONDS);
            return albumInfo;
        } finally {
            //4.释放锁
            lock.unlock();
        }
    } 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;
}

2.4 AOP与分布式锁整合

专辑详情中多项数据接口需要查询缓存，那么分布式锁的业务逻辑代码就会出现大量的重复。因此，我们可以参考Spring框架事务注解来实现简化代码。只要类上添加了一个注解，那么这个注解就会自带分布式锁的功能。

实现如下：

在service-util 模块中添加一个注解
提供切面类，采用环绕通知对自定义注解进行增强

2.4.1 自定义注解

package com.atguigu.tingshu.common.cache;

import java.lang.annotation.*;

/**
 * 自定义缓存注解，用于将目标方法（查库）结果放入缓存，采用分布式锁避免缓存击穿
 * 元注解：
 *
 * @Target 注解使用位置
 * @Retention 注解保留到哪个阶段  SOURCE CLASS RUNTIME
 * @Inherited 注解是否可以继承
 * @Documented 是否产生Java文档 通过javadoc删除文档中是否包含该注解
 */
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@Documented
public @interface GuiGuCache {


    /**
     * 目标方法执行查库结果放入到Redis缓存Key的前缀
     *
     * @return
     */
    String prefix() default "data:";


}

2.4.2 自定义切面类

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 jakarta.servlet.http.HttpServletRequest;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.language.Nysiis;
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-04-19 09:14
 */
@Slf4j
@Component
@Aspect
public class GuiGuCacheAspect {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private RedissonClient redissonClient;


    /**
     * 采用环绕通知，对所有模块中方法使用@GuiGuCache注解方法进行增强
     *
     * @param pjp 切入点目标方法对象
     * @return
     * @throws Throwable
     */
    @Around("@annotation(guiGuCache)")
    public Object doBasicProfiling(ProceedingJoinPoint pjp, GuiGuCache guiGuCache) throws Throwable {
        try {
            //一、前置通知逻辑
            //1.优先从缓存中获取数据，命中返回直接返回，未命中获取分布式锁
            //1.1 定义业务数据Key 形式：自定义缓存前缀+目标方法参数（多个参数采用_拼接）
            String params = "none";
            //1.1.1 获取目标方法参数数组
            Object[] args = pjp.getArgs();
            if (args != null && args.length > 0) {
                //1.1.2 将数组转为List集合
                params = Arrays.asList(args)
                        .stream()
                        .map(arg -> arg.toString())
                        .collect(Collectors.joining("_"));
            }
            String dataKey = guiGuCache.prefix() + params;
            //1.2 查询Redis获取业务数据(Redis中数据类型String存储)
            Object result = redisTemplate.opsForValue().get(dataKey);
            //1.3 命中缓存则返回，否则执行第二步获取分布式锁
            if (result != null) {
                return result;
            }
            //2.获取分布式锁-最终所有来到该位置线程终究会获取锁成功
            //2.1 创建锁Key 形式：前缀+数据标识+锁后缀
            String lockKey = dataKey + RedisConstant.CACHE_LOCK_SUFFIX;
            //2.2 创建锁对象
            RLock lock = redissonClient.getLock(lockKey);
            //2.3 执行加锁方法
            lock.lock();

            //二、执行目标方法（被增强方法执行）
            try {
                //3.执行业务（查询数据库业务,结果放入Redis缓存中）
                //3.1 只要线程调用lock方法最终都会获取锁成功，避免后来线程获取锁成功造成多次查库，再查一次缓存
                result = redisTemplate.opsForValue().get(dataKey);
                if (result != null) {
                    return result;
                }
                //3.2 执行查询数据库方法 将查询结果放入缓存
                result = pjp.proceed(args);
                //三、后置通知逻辑
                long ttl = RedisConstant.ALBUM_TIMEOUT + RandomUtil.randomInt(0, 500);
                redisTemplate.opsForValue().set(dataKey, result, ttl, TimeUnit.SECONDS);
                return result;
            } finally {
                //4.释放锁
                lock.unlock();
            }
        } catch (Throwable e) {
            log.error("[自定义缓存切面]查询业务数据异常兜底处理：{}", e);
            //兜底处理方案：当前Redis服务不可用（缓存，分布式锁），直接查询数据库
            return pjp.proceed(pjp.getArgs());
        }
    }
}

2.4.3 测试分布式锁

在这个实现类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) {}

3、布隆过滤器

3.1 什么是布隆过滤器

布隆过滤器（Bloom Filter），是1970年，由一个叫布隆的小伙子提出的，距今已经五十年了。它实际上是一个很长的二进制向量和一系列随机映射函数。布隆过滤器可以用于检索一个元素是否在一个集合中。它的优点是空间效率和查询时间都比一般的算法要好的多，缺点是有一定的误识别率和删除困难。二进制大家应该都清楚，存储的数据不是0就是1，默认是0。

主要用于判断一个元素是否在一个集合中，0代表不存在某个数据，1代表存在某个数据。

总结： 判断一个元素一定不存在或者可能存在！ 存在一定的误判率{通过代码调节}

Bit 数组：

0	0	0	0	0	0	0	0	0

3.1.1 使用场景

大数据量的时候, 判断一个元素是否在一个集合中。解决缓存穿透问题

3.1.2 原理

**存入过程 **

布隆过滤器上面说了，就是一个二进制数据的集合。当一个数据加入这个集合时，经历如下：

通过K个哈希函数计算该数据，返回K个计算出的hash值

这些K个hash值映射到对应的K个二进制的数组下标

将K个下标对应的二进制数据改成1。

如图所示：

查询过程

布隆过滤器主要作用就是查询一个数据，在不在这个二进制的集合中，查询过程如下：

1、通过K个哈希函数计算该数据，对应计算出的K个hash值

2、通过hash值找到对应的二进制的数组下标

3、判断：如果存在一处位置的二进制数据是0，那么该数据不存在。如果都是1，该数据可能存在集合中。

3.1.3 布隆过滤器的优缺点

优点

由于存储的是二进制数据，所以占用的空间很小
它的插入和查询速度是非常快的，时间复杂度是O（K），空间复杂度：O (M)。
- K: 是哈希函数的个数
- M: 是二进制位的个数
保密性很好，因为本身不存储任何原始数据，只有二进制数据

缺点

添加数据是通过计算数据的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。

那么你是不是连“张三丰”都一起删了呀。

3.2 初始化布隆过滤器

在service-search 模块的启动类中添加

package com.atguigu.tingshu;

import com.atguigu.tingshu.common.constant.RedisConstant;
import jakarta.annotation.PostConstruct;
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.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;

@Slf4j
@SpringBootApplication//(scanBasePackages = "com.atguigu.tingshu")
@EnableDiscoveryClient
@EnableFeignClients
@EnableCaching //开启springCache缓存
public class ServiceAlbumApplication {

    public static void main(String[] args) {
        SpringApplication.run(ServiceAlbumApplication.class, args);
    }

    @Autowired
    private RedissonClient redissonClient;


    /**
     * 构造方法执行后（对象实例化）
     * 启动类执行完毕后，springboot启动后执行一次
     */
    @PostConstruct
    public void initBloomFilter() {
        log.info("[专辑服务]启动项目后初始化布隆过滤器");
        RBloomFilter<Long> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        //参数1：数据规模（业务增长扩容） 参数2：误判率
        bloomFilter.tryInit(100000, 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, "访问专辑不存在");
    }
    //.....省略代码
}

4、数据一致性方案

4.1 Canal服务端

「先更新数据库，再删缓存」的策略的第一步是更新数据库，那么更新数据库成功，就会产生一条变更日志，记录在 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; 
   
```

采用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

4.2 Canal客户端

目前不支持：JDK17

创建独立新模块，service-cdc在java项目中引入依赖，JDK环境要设置为1.8

<?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>

确保该模块环境JDK8

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客户端长时间未监听到变更数据，做以下操作

重置MySQL binglog

#重置MySQL主节点Binglog
reset master;
   
SHOW MASTER STATUS;

重新创建canal容器

第6章 详情优化.md 52 KB Permalink History Raw

第6章 详情优化