CompletableFuture 异步线程池顾名思义就是，管理一些线程的资源池，提供限制和管理资源的方式，维护一些基本信息。Executor框架Java5后面引进的一种框架，有队列，拒绝策略等东西由三部分组成：任务Task：定义做什么Runnable不返回结果、Callable返回结果可抛异常任务的执行Executor（ThreadPoolExecutor关键类继承于此）：谁来做怎么做异步计算的结果Future：如何获取结果运行流程：先决定用Runnable、Callable哪个接口通过调用 ExecutorService 的 submit() 方法来提交任务，具体的执行则由其实现类（如 ThreadPoolExecutor）来完成。给一个结果Future，可以主进程干别的获取结果Future.get，计算完成直接取走，计算未完成线程堵塞ThreadPoolExecutor核心参数image-20251010161133982ThreadPoolExecutor 3 个最重要的参数：corePoolSize：核心线程数：线程池中保持活动状态的线程数量，即使是空闲的（常驻店员）maximumPoolSize：最大线程数：线程池能够容纳同时执行的最大线程数（常驻店员+临时工最大值）只有workQueue和corePoolSize都满了时候才招临时workQueue：用于保存等待执行任务的阻塞队列（等候区）ThreadPoolExecutor其他常见参数 :unit : keepAliveTime 参数的时间单位。keepAliveTime:非核心线程在没有作业的时候的等待时间threadFactory :用于创建新线程，可以用它来为线程池中的线程命名。handler :拒绝策略：当任务队列已满且线程数达到maximumPoolSize时，用于处理新提交任务的策略。线程池创建一、通过ThreadPoolExecutor 构造函数创建没啥说的，上面参数一个一个写就行了构造函数。二、通过 Executors 工具类创建（不推荐）Executors提供了几个常用的静态方法创建线程池：Executors.newFixedThreadPool(int nThreads): 创建一个固定大小的线程池。Executors.newSingleThreadExecutor(): 创建一个只有一个线程的线程池。Executors.newCachedThreadPool(): 创建一个可缓存的线程池，线程数根据任务量动态调整。Executors.newScheduledThreadPool(int corePoolSize): 创建一个可以执行定时任务的线程池。[11]为什么不推荐用Executors 创建前两个：使用了无界的LinkedBlockingQueue作为任务队列。如果任务的生产速度远快于消费速度，队列中的任务会持续堆积，最终可能导致服务器内存溢出（OOM）。后两个：允许创建的线程数量几乎是无限的。当有大量并发请求时，它会不断创建新线程，这可能耗尽CPU和内存资源，导致系统崩溃。image-20251010164219604工作过程image-20251010165637571线程池拒绝策略CallerRunsPolicy, 使用线程池的调用者所在的线程去执行被拒绝的任务，除非线程池被停止或者线程池的任务队列已有空缺。AbortPolicy, 直接抛出干不了的异常DiscardPolicy, 不做任何处理，静默拒绝提交的任务。DiscardOldestPolicy, 抛弃最老的任务，然后执行该任务。还有自定义拒绝方式CompletableFuture为什么不用Future本身了呢？虽然他可以异步计算，但是获取结果的时候必须使用Future.get()的方式阻塞调用线程，或者轮询询问是否完成。我的项目创建CompletableFuture使用的是通过自定义线程池，根据runnable构建执行任务。也就是下面这种ompletableFuture.runAsync(() -> { redisUtil.zset("user_video_history:" + uid, vid); // 添加到/更新观看历史记录 videoStatsService.updateStats(vid, "play", true, 1); }, taskExecutor);而不是使用了默认内置线程池ForkJoinPool.commonPool()，因为这个线程池的处理线程数量是电脑的CPU核数-1，不如自定义1.有并行加载数据的，比如同时加载用户信息和视频信息，加快加载速度2.有异步通知消息的，就是如果有人点赞了异步去通知，作用就是用户不需要等结果，还有就是网络IO发消息这种操作也是比较耗时间的，异步不会堵塞主线程。3.有异步缓存更新的，符合读写分离这种设计理念实际使用上，项目在：并行加载：聊天系统同时加载聊天记录和用户资料异步通知：异步通知消息的，就是如果有人点赞了异步去通知异步更新：除了缓存写的时候，还有就是在于更新视频到OSS上
常规代码首先Controller层、Service层不必多说，常规流程的代码，这里说一下实现的功能有什么：1、根据弹幕id集合获取弹幕列表2、删除弹幕DanmuWebSocketServer技术考量1、首先因为DanmuWebSocketServer 类的实例不是由 Spring 容器管理的标准 Bean，想拿到数据库和缓存的内容只能定义成静态字段获取2、Session 是 中代表一个客户端与服务器之间完整、双向、持久化 WebSocket 连接的 Java 对象。基于SpringBoot容器。ServerEndpoint一、连接时执行 @OnOpen public void onOpen(Session session, @PathParam("vid") String vid) { if (videoConnectionMap.get(vid) == null) { Set<Session> set = new HashSet<>(); set.add(session); videoConnectionMap.put(vid, set); } else { videoConnectionMap.get(vid).add(session); } sendMessage(vid, "当前观看人数" + videoConnectionMap.get(vid).size()); // System.out.println("建立连接，当前观看人数: " + videoConnectionMap.get(vid).size()); }1、@OnOpen：连接就执行2、 videoConnectionMap.put(vid, set)：连接注册3、sendMessage(vid, "当前观看人数"...：广播观看人数的消息二、收到消息时触发1、@OnMessage2、Token验证：目的是做到只有登录用户才能发弹幕3、持久化弹幕到数据库4、（序列化后）广播弹幕String dmJson = JSON.toJSONString(danmu); sendMessage(vid, dmJson);三、连接关闭执行1、@OnClose2、在Map中移除当前的连接记录3、如果没人了就直接移除这个视频，有人就更新在线人数if (videoConnectionMap.get(vid).size() == 0) { videoConnectionMap.remove(vid); } else { sendMessage(vid, "当前观看人数" + ...)重点发送信息 public void sendMessage(String vid, String text) { Set<Session> set = videoConnectionMap.get(vid); // 使用并行流往各客户端发送数据 set.parallelStream().forEach(session -> { try { session.getBasicRemote().sendText(text); } catch (Exception e) { e.printStackTrace(); } }); } }session.getBasicRemote().sendText(text)实现发送并且动态渲染广播机制服务器里面存在一个通讯录// 对每个视频存储该视频下的session集合 private static final Map<String, Set<Session>> videoConnectionMap = new ConcurrentHashMap<>();有用户连接就会把session存到对应视频的集合里面然后像每个连接都发一遍消息就是广播// 步骤1：根据视频ID，获取所有观看该视频的客户端连接 Set<Session> set = videoConnectionMap.get(vid); // 步骤2：使用并行流遍历所有连接，并向每个连接发送消息 set.parallelStream().forEach(session -> { try { // 步骤3：通过session对象发送文本消息 session.getBasicRemote().sendText(text);
Netty项目的实时通信就是基于Netty实现的。以下是Netty部分的笔记：引子首先说NIOimage-20251016212853121BIO：同步阻塞 IO （个人理解就是悲观锁）--数据的读写必须堵塞在一个线程里进行NIO： 同步非阻塞（个人理解就是多线程）--- 面向缓冲的，基于通道的IO方法AIO：异步非阻塞，就是NIO2代，基于事件和回调机制例如你有1000个IO操作，如果使用NIO，是1000个线程反复调用read。如果使用多路复用，可以1个线程反复调用select，在数据就绪后，再调用read。还是能节约资源的[!NOTE]异步 IO 是基于事件和回调机制实现的，也就是应用操作之后会直接返回，不会堵塞在那里，当后台处理完成，操作系统会通知相应的线程进行后续的操作。Netty是什么？Netty是一个基于NIO的CS框架，可以快速简单的开发网络应用程序；简化了TCP和UDP套接字服务器等网络编程。应用场景首先就是常用的框架比如RocketMQ、Elasticsearch实现自己的HTTP服务器，类Tomcat可以自己实现一个实时通信系统，我的项目就是基于Netty实现的通信各部件1、Channel：通道，所有数据的操作以及状态变更都在这个通道上，每个用户的WebSocket连接就是一个Channel。2、EventLoop:事件循环，单线程执行器负责处理一个或多个Channel上的所有I/O事件，永不停止。​ 单线程处理多个Channel：IO多路复用，同redis的单线程监听的大量客户端 连接，也就是NIO​ 永不停止：也就是说一直循环等待，永远不停止，只要有被分配到他头上的 Channel来了操作，就是他来执行。3、EventLoopGroup：事件循环组，一组事件循环的集合，用来管理和分配，项目中有BOSS和WORKER这两个4、Pipeline：通道管道，是依附在Channel上的处理器链，也就是责任链模式的工作机制启动流程（详细见实时通信的实现逻辑）(入站)客户端发消息，服务器收消息:WebSocketHandler+ChatHandler+IMServerIMServer：客户端发起WebSocket连接到端口Boss接受连接，创建ChannelBoss将这个新的Channel注册到work上的一个EventLoop上开始执行这个Channel上的Pipeline处理数据读取过来责任链传过来的数据JSON 反序列化成对象，并且再次拿到UserId封装到里面持久化数据：chatDetailedMapper.insert(chatDetailed);完整存 redisUtil.zset("chat_detailed_zset")消息id缓存准备推送数据：推送数据：获取发送者自己的Channel集合（多端同步）获取接收者所有的Channel集合（消息转发）出站：推送消息到对应的EventLoop上，启动出站Pipelinefor (Channel channel : from) { channel.writeAndFlush(IMResponse.message("whisper", map)); }(出站)服务端发消息，客户端收消息：WebSocketHandler业务触发：定位到目标用户的全部在线 Channel。写入数据：承接上面的推送writeAndFlush()启动出站Pipeline流：消息从 Pipeline 的尾部（靠近应用层）开始向前（靠近网络层）流动，执行出站处理器。编码：IMResponse 对象编码-->WebSocketFrame 转换为网络可传输的原始字节流。发送：EventLoop 线程将这些字节流写入操作系统的 Socket 发送缓冲区，数据通过网络发送给客户端。触发时机：聊天消息推送触发，评论通知推送，点赞通知推送总结：整个过程是围绕 EventLoop 和 Pipeline 循环往复：收消息（入站）： EventLoop 读数据 → Pipeline 从头到尾处理（解码、认证、业务）。发消息（出站）： 业务逻辑触发 → Pipeline 从尾到头处理（业务编码、协议编码） → EventLoop 写数据。心跳机制说心跳机制先说一下 TCP长连接和短连接长连接就是在S和C之间建立连接一直持续，短连接就是读写完成之后就关闭连接为什么需要心跳机制？在长连接的情况下，如果一方断开连接了，CS之间没有交互的话，是不知道对方已经断开了的，所以引入心跳机制：如果处于Idle状态（连接但无交互），C或S会发送一个特殊的数据包给对方，接收方接收到这个特殊数据报文，也返回一个特殊的数据报文，这就是一个PING-PONG 交互。Netty的心跳机制TCP本身有心跳机制，但是不太灵活，所以一般在应用层实现，对应到Netty就是核心类IdleStateHandler
TCP三次握手一次握手:客户端发送带有 SYN（SEQ=x） 标志的数据包 -> 服务端，然后客户端进入 SYN_SEND 状态，等待服务端的确认；二次握手:服务端发送带有 SYN+ACK(SEQ=y,ACK=x+1) 标志的数据包 –> 客户端,然后服务端进入 SYN_RECV 状态；三次握手:客户端发送带有 ACK(ACK=y+1) 标志的数据包 –> 服务端，然后客户端和服务端都进入ESTABLISHED 状态，完成 TCP 三次握手。为什么三次握手第一次握手：Client 什么都不能确认； Server 确认了对方发送正常，自己接收正常第二次握手：Client 确认了：自己发送、接收正常，对方发送、接收正常； Server 确认了：对方发送正常，自己接收正常第三次握手：Client 确认了：自己发送、接收正常，对方发送、接收正常； Server 确认了：自己发送、接收正常，对方发送、接收正常三次握手就能确认双方收发功能都正常，缺一不可。image-20251028110654478含义SYN:同步。 用于发起连接请求ACK:确认，表示确认号字段有效SEQ=x： 报文段的序列号。它表示该报文段所携带的数据的第一个字节在整个字节四次挥手第一次挥手：客户端发送一个 FIN（SEQ=x） 标志的数据包->服务端，用来关闭客户端到服务端的数据传送。然后客户端进入 FIN-WAIT-1 状态。第二次挥手：服务端收到这个 FIN（SEQ=X） 标志的数据包，它发送一个 ACK （ACK=x+1）标志的数据包->客户端 。然后服务端进入 CLOSE-WAIT 状态，客户端进入 FIN-WAIT-2 状态。第三次挥手：服务端发送一个 FIN (SEQ=y)标志的数据包->客户端，请求关闭连接，然后服务端进入 LAST-ACK 状态。第四次挥手：客户端发送 ACK (ACK=y+1)标志的数据包->服务端，然后客户端进入TIME-WAIT状态，服务端在收到 ACK (ACK=y+1)标志的数据包后进入 CLOSE 状态。此时如果客户端等待 2MSL 后依然没有收到回复，就证明服务端已正常关闭，随后客户端也可以关闭连接了为什么要四次挥手?TCP 是全双工通信，可以双向传输数据。任何一方都可以在数据传送结束后发出连接释放的通知，待对方确认后进入半关闭状态。当另一方也没有数据再发送的时候，则发出连接释放通知，对方确认后就完全关闭了 TCP 连接。举个例子：A 和 B 打电话，通话即将结束后。第一次挥手：A 说“我没啥要说的了”第二次挥手：B 回答“我知道了”，但是 B 可能还会有要说的话，A 不能要求 B 跟着自己的节奏结束通话第三次挥手：于是 B 可能又巴拉巴拉说了一通，最后 B 说“我说完了”第四次挥手：A 回答“知道了”，这样通话才算结束。image-20251028110709814
Docker运行笔记准备把项目优化，使用Docker打包环境配置，这样双机运行更方便一些省略双机的docker安装过程IDEA实现打包镜像1、安装docker插件2、在项目下创建Dockerfile文件（与pom.xml同目录）#基础镜像JRE8 FROM anapsix/alpine-java:8_server-jre_unlimited #时区设置 RUN ln -sf /usr/share/zoneinfo/Asia/Shanghai /etc/localtime #目录 WORKDIR /app #端口 EXPOSE 7070 #jar实际名称 COPY ./target/backend-0.0.1-SNAPSHOT.jar app.jar # 运行Spring Boot应用 CMD java -jar app.jar3、配置docker连接4、构建镜像注意：如果要在本机上使用docker运行，数据库配置需要改动，如果依旧使用127.0.0.1，docker容器不能访问，遂改成${DB_HOST:127.0.0.1}，并且在运行的时候，指定DB_HOST的值为当前电脑ip即可。
存储引擎InnoDB和MyISAM区别InnodBmyisam事务支持不支持索引聚集索引非聚集索引（见B+树部分）锁最小是行锁最小是表锁count不保存行数保存，查行数很快恢复支持redo不支持索引索引是什么？就是数据库的目录为什么索引能提高速度？速度很快，其次是数据唯一性有保障（可以创建唯一索引），加速排序和分组（因为索引本身是排好序的）聚集索引和非聚集索引聚集索引：数据的物理顺序和列值的逻辑顺序相同（主键）查询的时候，因为聚集索引 B+ 树的叶子节点存储了完整数据，直接查就行非聚集索引：就是顺序没关系（除了聚集就全是非聚集）查询的时候，因为非聚集索引 B+ 树的叶子节点存储的是 索引值 和 对应主键值1、需要先在这个树找到该行的主键值2、使用主键值，去聚集索引 B+ 树二次查找这个过程就叫做「回表」索引的数据结构是树，因为树天生有二分查找的特性，查询速度高索引失效索引失效就是违反了数据库使用索引的原则，转变成了全表扫描查询条件类1、Like通配符前置LIKE '%关键字这个没什么说的，就是索引是左侧顺序匹配的，相当于词典，只给右边的字母显然不能找到索引。2、最左前缀原则在使用复合索引的时候，只有带着最左边的索引成分的时候，索引才能正常使用。索引失效的时候，查询的性能就变成了全表扫描，效率慢。CREATE INDEX idx_col ON table (a, b, c);a是最左边WHERE a = 1 AND b = 2可以，因为有a，索引成立WHERE b = 2 AND c = 3不可以，因为没有a，索引失效3、否定条件比如 ！= not in，范围太大，使用索引成本高于直接进行全表扫描。操作索引类1、对索引进行运算where DATE(time) = '2025-1-1'因为这相当于对索引列的所有数据都计算一遍，等同于全表扫描。2、索引隐形转换比如id是int，结果和string类型比较，内部转换，同1。MYSQL索引为什么使用 B+树？平衡二叉树AVL：AVL实现平衡主要是靠旋转，旋转是非常耗时的，所以不能用（增加和删除）红黑树：和AVL相比，红黑树不追求完美的平衡，只是大致的平衡。对比AVL，查询略降，删除大提高因为红黑树同时引入了颜色，当插入或删除数据时，只需要进行 O(1)次数的旋转以及变色就能保证基本的平衡，不需要像 AVL 树进行 O(lgn)次数的旋转。总的来说， 红黑树的统计性能高于 AVL。红黑树应用：HashMap链表+红黑树解决哈希冲突 ，TreeMap 红黑树存储排序键值对在“高且瘦”的树中，你每次比较都可能需要访问一个新的节点。如果每个节点都位于磁盘的不同页（Page）上，那么查找一个数据就可能需要很多次磁盘I/O。比如树高20，就需要最多20次I/O，这无法接受。B树：矮胖树，一个节点可以有多个子节点，每个节点都会存储数据本身，节点内的键值是排好序的，节点之间也满足排序关系。B+树直接看和B树的区别数据存储：B+树的Data只存在于叶子节点中，非叶子节点只有索引。（B树什么节点都存数据）叶子结点：B+树中只有叶子节点有指针，叶子节点之间通过指针链接起来形成一个有序链表（B树叶子节点是相互独立的）查找：B+树什么查找都得走到叶子节点，查找路径是稳定的（B树可能在非叶子节点就命中）Data，可能是行的全部数据（如 Innodb 的聚簇索引），也可能只是行的主键（如 Innodb 的辅助索引），或者是行所在的地址（如 MyIsam 的非聚簇索引）。(重要)由于B+树非叶子节点不存Data，所以每个节点能存的索引更多，树更矮。因此，B+树优势：1、因为树的结构更矮胖，磁盘IO次数少（读写）2、因为叶子节点是通过双向链表连接的，查询到一条数据就可以沿着链表的上下直接顺序查询，IO路径规律，范围查询效率更高SELECT * FROM users WHERE age BETWEEN 20 AND 30IO路径规律——>磁头不需要大动3、因为Data都存在于叶子节点里面，最差的情况就是一般情况，查询性能稳定。4、因为双向链表和Data存于叶子，全表遍历效率高，只需要遍历链表就行SQL优化慢SQL优化1、定位慢查询在配置文件启用，long_query_time = 1：设置阈值；low_query_log = 1：开启慢查询日志。然后执行 SHOW FULL PROCESSLIST；看status和time2、分析慢查询情况定位之后，对这个语句执行加上EXPLAIN；看type：类型，all是全表，其他次之看rows：行数3、解决
锁表级锁和行级锁的区别qubie表级锁行级锁范围锁整张表锁当前字段实现简单，加锁快困难，加锁慢与引擎无关有关，因引擎谁有MYISAM、InnoDBInnoDB（默认行）InnoDB的几种表级锁表锁表级读锁：阻止其他的对表的写表级写锁：阻止其他对表的读写元数据锁MDL作用：防止执行CRUD的时候表结构变更数据库本身会加MDL，不用我们来操作写锁获取优先级高于读锁对一张表进行 CRUD 操作时，加的是 MDL 读锁；对一张表做结构变更操作的时候，加的是 MDL 写锁；那它在什么时候释放？-->在事务提交之后意向锁作用：为了快速判断表里是否有行锁为什么要判断？因为读锁和行锁是读写互斥，写写互斥的在加上行共享锁之前，先在表级别加上一个“意向共享锁”在加上行独占锁之前，先在表级别加上一个“意向独占锁”这里面谈到普通的select语句是不算里面的，这类是通过MVCC实现的一致性，无锁InnoDB的几种行级锁锁的兼容性分类共享锁S：读锁，事务在读取记录的时候获取共享锁，允许多个事务同时获取锁排它锁X：写锁，事务在修改记录的时候获取排它锁，不允许多个事务.....。锁的范围分类记录锁：锁定单个记录。间隙锁：锁定一个id数字集合（范围），不包括记录本身。前开后开临键锁：记录锁+间隙锁，锁定一个范围，包含记录本身，主要目的是为了解决幻读问题。前开后闭什么时候加行级锁？普通的 select 语句是不会对记录加锁的，MVCC实现一致性首先记住，加锁的对象都是索引！！！，基本单位就是默认的next-key lock唯一索引等值查询情况查询的记录 存在，定位到记录之后，退化成 X记录锁查询的记录 不存在，定位到第一个大于查询记录的记录之后，退化成 间隙锁，锁住这个记录和上一条记录的范围思考：不沿用next-key lock的原因是因为这个场景下幻读只需要间隙锁就能解决了为什么 id = 5 记录上的主键索引的锁不可以是 next-key lock？如果是 next-keylock，就意味着其他事务无法删除 id = 5 这条记录，但是这次的案例是查询 id = 2 的记录，只要保证前后两次查询 id = 2 的结果集相同，就能避免幻读的问题了。唯一索引范围查询情况先说明范围查询的整体机制：会对每一个扫描到的索引加next-key lock，下面细分情况实际在 Innodb 存储引擎中，会用一个特殊的记录来标识最后一条记录，该特殊的记录的名字叫supremum pseudo-record大于情况：往后推，前加一个包含自己的，后面以此递归最后扫到supremum加 (xxxx, +∞]大于等于情况：等值的地方变成记录锁小于或小于等于情况：条件值不在表里，终止扫描记录的Next-Key锁退化成间隙锁小于情况：条件值在表里，终止扫描记录的Next-Key锁退化成间隙锁小于等于情况：条件值在表里，终止扫描记录的Next-Key锁不退化成间隙锁举例：有个表有4个数据项idname10A20B30C40D情况5：10满足加(Previous, 10]，20满足加(10, 20]，30满足加(20, 30]，到边界了，终止扫描SELECT * FROM products WHERE id <= 30;情况4：10满足加(Previous, 10]，20满足加(10, 20]，30不满足，退化，加(20, 30)SELECT * FROM products WHERE id < 30;情况1：定位到30，加(20, 30]，40满足加(30, 40]，接着扫描，加锁(40, +∞]SELECT * FROM products WHERE id > 20;情况2：20，临键锁退化，变成记录锁锁20，接着同情况1SELECT * FROM products WHERE id >= 20;情况3：10满足加(Previous, 10]，20满足加(10, 20]，然后到30，不满足，退化，加(20, 30)SELECT * FROM products WHERE id <(<=) 25;什么时候加表级锁？当执行UPDATE、DELETE的语句的时候，WHERE没有命中索引或者索引失效的时候，就会对全表扫描加锁。事务四大特性原子性：事务是最小的执行单位，不能分割 --- undo log保证的隔离性：并发访问数据库的时候，事务彼此之间是不可见的（取决于事务隔离级别）比如先后扣款，先扣款事务还没做完，后扣款事务来了看到的也是原数量持久性：一个事务被提交之后，他对数 Q`据库的改变是永久的，即使数据库发生故障。 --- redo log保证的一致性：执行事务前后，数据保持一致（指变化总和）并发事务带来的问题（问题）脏读一个事务读取数据并且修改了这个数据，这个时候第二个事务读取了这个未提交的数据，但是第一个事务突然回滚，导致改后的数据没有到数据库里面，最后第二个事务读到了脏数据丢失修改一个事务在读取一个数据的时候，另一个事务读取这个数据，第一个事务修改这个数据，第二个数据也修改这个数据，但是数据最后相当于只修改了一次，丢失了一次修改不可重复读指一个事务在事务内多次读一个数据，但是在中间间隙中来了第二个事务，修改了数据，导致第一个事务在事务内前后两次读到的数据不一样幻读意思就是一个事务读了几条数据，在他第二次读之前中间来了第二个事务插入了几条数据，导致第一个事务再读的时候像出现了幻觉一样多了几条数据不可重复读、幻读的区别实际上幻读就是不可重复读特殊情况，不可重复读强调某一条数据被修改了，幻读强调对数据行数的修改上（变多或者变少）之所以区分是因为解决他们的方案不一样标准事务隔离级别（解决）是用来平衡隔离性和并发性能的东西级别越高，数据一致性越好，但并发性能可能越低。读取未提交：允许读取未提交的数据变更，啥都防不住读取已提交：允许读取并发事务已经提交的数据，可以阻止脏读可重复读：对同一个字段的多次读取都是一样的，除非是自身修改，阻止了脏读和不可重复读，至于幻读，仍有可能（InnoDB大幅减少）可串行化：最高隔离，依次执行，全都防住READ-UNCOMMITTEDREAD-COMMITTEDREPEATABLE-READSERIALIZABLEMysql的默认隔离级别Mysql InnoDB引擎 的默认隔离级别是可重复读。InnoDB怎么解决大部分幻读的快照读：普通的SELECT的语句，通过MVCC实现，事务启动时创建一个数据快照，后续的快照读都读取这个版本的数据，这样避免了看到其他事务插进来的行（幻读）和修改的行（不可重复读）当前读：除了普通查询之外的操作读取到的都是最新数据而不是快照数据，InnoDB用临键锁next-key lock防止其他事物在这个范围内插入新的记录（幻读）（像 SELECT ... LOCK IN SHARE MODE（加S锁），SELECT ... FOR UPDATE，INSERT， UPDATE， DELETE（加X锁） 这些操作）正因为这样的机制，InnoDB在可重复读（RR）级别的性能，在只读或者读多写少的情况下，和读取已提交（RC）的性能差不多。在写密集的时候，RR的间隙锁可能会导致它比RC出现更多的锁等待Mysql的隔离级别怎么实现的由锁和 MVCC 共同实现的读取未提交：啥也不用干（实际上InnoDB根本不支持，设置也会自动变成读取已提交）读取已提交：通过Read View实现的，实际上是高级的版本号可重复读：通过Read View实现的，实际上是高级的版本号可串行化：完全依赖于锁Read View是如何工作的？是什么东西？他有四个字段：m_ids是一个活跃事务id列表min_trx_id活跃事务中id最小的事务max_trx_id是活跃事务最大id的下一个事务的id值，也就是最大事务id+1creator_trx_id是创建这个Read View的事务id活跃事务：运行中且未提交的事务怎么解决并发的RR可重复读实现这时候一个事务来进行增删改查等操作，会先创建一个Read VIew来判断在做后续的操作的时候，其他事务自己能不能可见，这里有个先后顺序，也就是说先创建快照，再执行查询。有几种情况：（trx _id是其他事务的id）如果记录的trx_id小于Read View中的min_trx_id，说明这个事务是在这view版本创建的时候就已经结束了。可见如果记录的trx_id大于等于Read View中的max_trx_id，说明了这个事务是在View 快照之后才开始的事务。不可见如果记录的trx_id在min_trx_id和max_trx_id之间，先判断trx_id在不在m_ids中在：表示生成该版本的记录的活跃事务还没提交，所以不可见（还在忙活，可能回滚）不在：表示生成该版本的记录的活跃事务已经提交，所以可见（忙活完了，间隙进来的） 在创建快照和执行查询的间隙进来的image-20251008212707053RC读以提交实现创建快照是每次执行SELECT都创建一个read view快照，有不可重复读的情况，因为没有解决那个间隙问题。InnoDB 对 MVCC 的实现undolog和readview的配合：当 Read View 判断当前数据不可见时，它必须通过 Undo Log 提供的版本链，去找到一个它认为可见的历史版本。可重复读：启动事务时生成一个 Read View，整个事务期间都用这个，记录相应的undo log用链表连起来，读只能读到启动之前的记录读提交：事务读取的时候创建Read ViewMVCC是什么上面这个版本链条控制并发事务同时访问一个记录的解决方案就是MVCC，下面就是思想，readview+undolog是实现读操作的时候使用快照读取。快照读取基于事务开始的时候数据库的状态来创建，因此事务不会读取其他事务未提交的修改，具体情况：读取的时候，事务查找符合条件的数据行，选择符合事务开始时间的版本读取，如果有多个版本，不晚于开始时间的最新版本。写操作的时候生成一个新的数据版本，并且把修改后的数据写入数据库，具体情况：写的时候，事务会为要修改的数据行创建一个新的版本，并且把修改后的数据写入新版本，新版本的数据有当前事务的版本号，以便其他事务能读取相应版本的数据；旧版本的数据也留着，以便其他事务快照读。事务提交和回滚的时候事务提交的时候，做的修改成为数据库的最新版本，并且其他事务可见。事务回滚的时候，做的修改撤销，并且其他事务不可见
谈谈对IOC的理解（是什么）IOC也就是控制反转，要谈谈IOC，就要先说一下控制反转是什么。传统的开发方式就是A依赖于B，那就是A 手动new一个B的对象出来；而IOC的方式是不需要new对象，而是通过IOC容器也就是Spring来帮助我们实例化对象，需要什么对象，就去IOC容器里取出来就行。所以说控制反转的意思就是“把对象创建管理什么的 权利交给外部环境IOC”。（为什么）那为什么要使用IOC？解决了什么问题？对象之间的耦合度变低，也可以说依赖程度；资源容易管理，比如使用spring容器很容易实现一个单例，而传统模式得自己写。没有IOC是这样的：Service 层想要使用 Dao 层的具体实现的话，需要通过 new 关键字在UserServiceImpl 中手动 new 出 IUserDao 的具体实现类 UserDaoImpl（不能直接 new 接口类）。放图...有IOC是这样的：放图...谈谈对AOP的理解（是什么）aop就是面向切面编程，主要的思想就是把横切关注的地方从核心业务逻辑分离出来，切面Aspect：对关注点封装的类。通知类型有哪些？Before前置通知：在方法调用之前触发After后置通知：在方法调用之后触发AfterReturning返回通知：在方法调用完成，返回返回值之后触发AfterThrowing异常通知：在方法调用中抛出异常后触发Around环绕通知：可以在任意前后触发，还可以直接拿到目标对象，要执行的方法（为什么）为什么要用AOP？解决了什么问题？比如日志记录、事务管理、权限控制、接口限流等，如果一直重复实现这些行为，代码会特别冗余，难以维护。// 日志注解 @Target({ElementType.PARAMETER,ElementType.METHOD}) @Retention(RetentionPolicy.RUNTIME) @Documented public @interface Log { /** * 描述 */ String description() default ""; /** * 方法类型 INSERT DELETE UPDATE OTHER */ MethodType methodType() default MethodType.OTHER; } // 日志切面 @Component @Aspect public class LogAspect { // 切入点，所有被 Log 注解标注的方法 @Pointcut("@annotation(cn.annotation.Log)") public void webLog() { } /** * 环绕通知 */ @Around("webLog()") public Object doAround(ProceedingJoinPoint joinPoint) throws Throwable { // 省略具体的处理逻辑 } // 省略其他代码 }@Log(description = "method1",methodType = MethodType.INSERT) public CommonResponse<Object> method1() { // 业务逻辑 xxService.method1(); // 省略具体的业务处理逻辑 return CommonResponse.success(); }（怎么实现的？）spring AOP就是基于动态代理实现的，我详细聊聊Spring AOP 的实现机制Spring AOP 的实现机制A.如何实现剥离？一共有三个对象，真实对象（Target）、代理对象（Proxy）、增强逻辑（Advice）工作流程：接收到方法调用。执行增强逻辑（如 @Before 的通知）。将调用转发给真实对象。执行增强逻辑（如 @After 的通知）。返回结果。B.机制一：JDK动态代理（基于接口）条件：至少实现了一个接口为什么至少实现了一个接口？因为JDK动态代理生成的类已经继承了Proxy类，由于单继承，所以它只能通过实现接口的方式实现代理原理：利用了Java原生的API Proxy InvocationHandler运行的时候动态生成一个代理类，这个类和真实对象是兄弟关系，实现同一个接口通过实现InvocationHandler插入增强的逻辑调用代理对象，走的增强逻辑C.机制二：CGLIB动态代理（基于继承）条件：无（没实现接口只能用它）原理：利用字节码运行的时候操作字节码，生成一个代理类，这个类和真实对象是父子关系子类会重写目标中所有非final的方法重写的方法中加入增强逻辑为什么不能重写final方法？因为final方法不能重写public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {// 1. 执行前置增强 (如 @Before)beforeAdvice();try {// 2. 调用真实目标对象的方法result = method.invoke(targetObject, args);// 3. 执行后置增强 (如 @AfterReturning)afterReturningAdvice(result);} catch (Exception e) {// 4. 执行异常增强 (如 @AfterThrowing)afterThrowingAdvice(e);throw e;} finally {// 5. 执行最终增强 (如 @After)afterFinallyAdvice();}D.总结JDK动态代理限制在于只能代理接口定义的方法，CGLIB动态代理限制在于不能代理final类、方法。Spring的默认：智能选择，有接口JDK动态代理，没有接口就CGLIBSpringBoot2.x以上版本：默认CGLIB-->为了避免不必要的报错(比如注入实现类)什么场景我们会强制使用CGLIB？一、希望代理没在接口定义的方法 二、希望代理类就是目标类的类型@Autowired直接按照实现类类型注入所以一般都用@resource按类型如果使用JDK代理，生成的代理对象只实现了接口，并不是实现类类型，会导致注入失败。而CGLIB代理生成的代理对象是目标类的子类，可以同时匹配接口类型和具体类类型。
三大日志binlog二进制日志Server层的日志，逻辑日志，用来全量数据备份和主从复制。写入时机在每完成一次更新操作之后，生成一条binlog刷盘时机什么时候刷盘由sync_binlog参数决定0：不去强制要求，由系统自行判断何时写入磁盘；1：每次事务提交后，都要将binlog写入磁盘；N：每N个事务，才会将binlog写入磁盘。方案STATEMENT：记录数据变更的SQL语句本身好：日志文件体积小，一SQL记一次坏：可能存在数据不一致，对于动态函数---NOW()UUID()LIMIT等依赖上下文，会导致主从数据不一致ROW：记录数据最终被修改成什么样好：数据一致性最高，绝对一致主从坏：日志文件体积大，更新多少行就记多少MIXED：折中判断特点追加写，写满一个文件，就创建一个新的文件继续写，不覆盖，存全量日志--->恢复，主从复制注意的是只记录库、表结构改变和数据修改的日志redolog重做日志存储引擎层面的，物理日志，用来保障事务的持久性和崩溃恢复Mysql有Buffer Pool缓冲池，跟redis一样。如果在Buffer Pool中的页的修改还没持久化到磁盘中，这个时候宕机了，事务更改就消失了引入WAL技术，就是先写到日志，再写到磁盘上image-20251010110214968写入时机当执行一条更新语句（如 UPDATE）时，先在 Buffer Pool 中查找对应的数据页。如果不在，就从磁盘加载到 Buffer Pool。接着，修改 Buffer Pool 中的这个数据页。此时，这个被修改了但还没刷回磁盘的数据页，就是所谓的 “脏页”。标记脏页的同时，InnoDB把这个页面做了什么物理修改记录记录到Redo Log Buffer中每一次数据变更都会产生对应的redo log例如：在表空间xx文件、第xx页、偏移量xx的位置写入了xxx数据刷盘时机事务提交时，先把 Redo Log Buffer 中对应的所有日志记录，强制刷盘，写到redolog中， 此时，现在已经算是被持久化了。（由参数调控）trx_commit = 1：每次事务提交时，都必须立刻将 redo log 从内存刷入磁盘。（默认）trx_commit = 0：不保证事务提交时刷盘，而是大约每秒钟由后台线程将 redo log 从内存刷入磁盘一次。不安全trx_commit = 2：每次事务提交时，只把 redo log 从内存写入到操作系统的页缓存（Page Cache），不直接写入磁盘文件。然后大约每秒钟，操作系统会把它的缓存统一刷到磁盘。 折中因为即使此时Buffer Pool 中的脏页还没有写回数据库，崩溃的时候，也可以依靠Redo log来进行重做，恢复到事务最后一次提交的状态这就是WAL思想，先写日志，后写数据两阶段提交不一致问题保障binlog和redolog的数据一致性，不一致的问题：先写 Redo Log，后写 Binlog，但在写 Binlog 前宕机：Redo Log 已经持久化，数据可以恢复。Binlog 没有记录。后果： 主库数据是完整的，但从库（通过 Binlog 复制）会缺失这条数据，导致主从数据不一致。先写 Binlog，后写 Redo Log，但在写 Redo Log 前宕机：Binlog 已经记录，从库会复制这条数据。Redo Log 没有完全持久化，崩溃恢复时主库会回滚这条数据（WAL 思想失败）。后果： 从库多出一条数据，主从数据不一致。阶段一：Prepare 准备阶段数据变更： 事务执行所有 SQL 语句，修改 Buffer Pool 中的数据页，并生成相应的 Redo Log 记录。Redo Log 刷盘：存储引擎将 Redo Log Buffer 中的数据刷入磁盘。标记 XID： 在这条 Redo Log 记录上，会标记一个事务 ID (XID)，并将其状态设置为 "Prepare"（准备状态）。状态： Redo Log 已持久化到磁盘，但事务在 InnoDB 内部仍未最终提交。如果此时宕机，恢复时 Redo Log 会发现自己处于 "Prepare" 状态，等待 Binlog 的最终判断。阶段二：Commit 提交阶段Step 2.1：写入 Binlog（Server 层操作）写入 Binlog：Server 层开始处理，将这个事务对应的 Binlog 事件写入到 Binlog 文件中。binlog刷盘： 此时，Binlog 也被强制刷盘。记录 XID： 在 Binlog 事件中，同样会记录这个事务的 XID。状态： Redo Log（Prepare 状态）和 Binlog 都已持久化到磁盘。Binlog 成为崩溃恢复的最终裁决者。Step 2.2：写入 Redo Log Commit 标记（InnoDB 层操作）Redo Log 提交： 事务执行层通知 InnoDB 存储引擎，事务已完成 Binlog 写入。最终标记： InnoDB 只需要在 Redo Log 上添加一个 "Commit" 标记（最终完成状态），表明该事务已彻底提交。这个操作通常只在内存中修改状态，并不强制刷盘。状态： 事务彻底完成。数据修改对外部可见。为什么 Step 2.2 不强制刷盘？因为即使在 Step 2.2 标记完成前宕机，我们回到 Step 2.1 的状态：Redo Log (Prepare) + Binlog (已记录)。崩溃恢复时，MySQL 依然会根据 Binlog 来完成提交。所以，Step 2.2 的标记只是为了事务的正常结束流程，在崩溃恢复中不再是决定性的步骤。崩溃恢复时恢复2PC 的价值体现在崩溃恢复时。MySQL 重启时，会根据 Redo Log 和 Binlog 的状态，执行以下判断逻辑：Redo Log 状态Binlog 状态恢复操作未 Prepare无记录回滚 ：事务视为失败，Redo Log 没有任何记录，直接清理。Prepare 状态有对应 XID 的记录提交 ：事务视为成功，强制完成 Redo Log 的 Commit 标记（以 Binlog 为准）。Prepare 状态无对应 XID 的记录回滚 ：事务视为失败，因为 Binlog 缺失，所以从库无法复制，主库需要回滚。Committed 状态有记录跳过：事务已完成，正常运行。undolog回滚日志逻辑日志，顾名思义，用来回滚的，也就是保障事务的原子性的作用。记录修改之前的状态Undo Log 的作用作用一：保障事务的**原子性，回滚实现机制：当一个事务执行 ROLLBACK 命令，或者事务执行过程中遭遇异常中断（如死锁被终止、应用程序崩溃）时，系统会利用 Undo Log 记录的逻辑信息，将数据恢复到事务开始前的状态。回滚流程： 沿着当前数据行上的 Roll_Pointer 指针，找到最新的 Undo Log 记录，执行其中的反向操作，从而撤销本次事务的所有修改。作用二：实现多版本并发控制 (MVCC)见事务篇刷盘在事务没提交之前，先把更新前的数据放到undo log 中，详细就是：插入：记录插入的主键值就行--------回滚的时候只删除这个主键值对应的记录就行删除：记录这个数据项的所有字段---回滚的时候把内容组成的记录重新插入到表中更新：记录主键值和被修改的旧值---回滚的时候把列更新为旧值就行随 Redo Log 刷盘而持久化。因为undolog本身就算是一个修改操作，修改了异同redo log 记录了此次事务「完成后」的数据状态，记录的是更新之后的值（持久性）undo log记录了此次事务「开始前」的数据状态，记录的是更新之前的值（原子性）
当前项目在更新数据的时候，是单纯的先更数据库再删除缓存，没有加入删除失败重试的机制，通过选型，选择了Canal进行监听Mysql的操作，并且接管删除缓存和重试操作。首先就是得知道Canal的原理，为什么能实现以上流程。Canal的原理要了解Canal的原理，首先得知道Mysql主从复制的原理Mysql主从复制第一步： 数据库发生变化，Master 写到 binlog。第二步： Slave 进程连接 Master，把 binlog 里的内容 复制 到自己的 relay log 中。第三步：Slave 另一个进程 读取 relay log，并把它里面的操作 执行 在自己的数据库上。通俗易懂来说就是 「抄作业」Canal实现监听的机制第一步：canal模拟slave的交互协议，伪装自己是slave，向master发送dump协议（其实就是不断传输的一个动作）第二步：master收到dump请求，建立连接，开始推送binlog给canal第三步：canal解析binlog（字节流）Canal实现同步的机制还在学习......Canal实现监听安装Canal安装完成修改配置# position info canal.instance.master.address=127.0.0.1:3306 canal.instance.master.journal.name=binlog.000001 canal.instance.master.position=157配置以上信息及账号密码启动Canal这里发生报错了，一顿排查之后发现是因为我的数据库的binlog不叫binlog，叫DESKTOP-xxxx-bin.000001，修改配置文件之后启动成功SHOW MASTER STATUS;image-20251019165338280Spring配置 <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.4</version> </dependency>canal: # 部署到208上，port默认为11111 server: 127.0.0.1:11111 # 实例名，与之前复制example之后那个文件夹名称一致 destination: redis # 设置canal消息日志打印级别 logging: level: top.javatool.canal.client: warn创建账号CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%'; -- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ; FLUSH PRIVILEGES;创建监听类import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import org.springframework.stereotype.Component; import javax.annotation.PostConstruct; import java.net.InetSocketAddress; import java.util.List; @Component public class CanalListener { /** * 解析数据 * * @param beforeColumns 修改、删除后的数据 * @param afterColumns 新增、修改、删除前的数据 * @param dbName 数据库名字 * @param tableName 表大的名字 * @param eventType 操作类型（INSERT,UPDATE,DELETE） * @param timestamp 消耗时间 */ private static void dataDetails(List<CanalEntry.Column> beforeColumns, List<CanalEntry.Column> afterColumns, String dbName, String tableName, CanalEntry.EventType eventType, long timestamp) { System.out.println("数据库：" + dbName); System.out.println("表名：" + tableName); System.out.println("操作类型:" + eventType); if (CanalEntry.EventType.INSERT.equals(eventType)) { System.out.println("这是一条新增的数据"); } else if (CanalEntry.EventType.DELETE.equals(eventType)) { System.out.println("删除数据：" + afterColumns); } else { System.out.println("更新数据：更新前数据--" + afterColumns); System.out.println("更新数据：更新后数据--" + beforeColumns); } System.out.println("操作时间：" + timestamp); } @PostConstruct public void run() throws Exception { CanalConnector conn = CanalConnectors.newSingleConnector(new InetSocketAddress("124.111.11.111", 11111), "example", null, null); while (true) { conn.connect(); conn.subscribe(".*\\..*"); // 回滚到未进行ack的地方 conn.rollback(); // 获取数据 每次获取一百条改变数据 Message message = conn.getWithoutAck(100); //获取这条消息的id long id = message.getId(); int size = message.getEntries().size(); if (id != -1 && size > 0) { // 数据解析 analysis(message.getEntries()); } else { //暂停1秒防止重复链接数据库 Thread.sleep(1000); } // 确认消费完成这条消息 conn.ack(message.getId()); // 关闭连接 conn.disconnect(); } } /** * 数据解析 */ private void analysis(List<CanalEntry.Entry> entries) { for (CanalEntry.Entry entry : entries) { // 解析binlog CanalEntry.RowChange rowChange = null; try { rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("解析出现异常 data:" + entry.toString(), e); } if (rowChange != null) { // 获取操作类型 CanalEntry.EventType eventType = rowChange.getEventType(); // 获取当前操作所属的数据库 String dbName = entry.getHeader().getSchemaName(); // 获取当前操作所属的表 String tableName = entry.getHeader().getTableName(); // 事务提交时间 long timestamp = entry.getHeader().getExecuteTime(); for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) { dataDetails(rowData.getBeforeColumnsList(), rowData.getAfterColumnsList(), dbName, tableName, eventType, timestamp); } } } } } 问题问题1：这里面遇到了一个问题，就是即使导入完坐标了，但是还是没有protocol这个类，导致很多报错解决1：发现是由于版本比帖子的版本高，canal取消了这个，解决办法是再导入一个包<dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.protocol</artifactId> <version>1.1.6</version> </dependency>问题2：运行之后，Springboot类反倒是启动不起来了，一直卡住了。解决2：发现原来是上面的代码有问题，详细状况： 1、@PostConstruct 方法默认在主线程运行，run方法有个无限循环。 2、Spring容器初始化之后，会调用带上面注解的方法，然后主线程被永久阻塞。 3、用一个新线程异步启动run方法new Thread。 成果数据库：cixingji 表名：user 操作类型:UPDATE 更新数据：更新前数据--[index: 0 等等一堆Canal实现同步Redis两种策略经过了解和搜索，发现有两种同步方式：1.Kafka手动处理Json消息解析此方法的最大的坏处在于，只要存在多个表，那内部逻辑就是一大堆的If else的循环，非常耦合。后续想要同步到其他组件比如ElasticSearch，又需要写新的逻辑 // 步骤 1: 【必须】手动添加 if/else 链来分发到不同的 ES 逻辑 if ("commodity".equals(table)) { // 步骤 2: 【必须】实现转换逻辑，将 Map 转换为 CommodityDocument for (Map<String, String> row : message.getData()) { CommodityDocument doc = convertMapToCommodityDocument(row); // 转换方法 esService.index("commodity_index", doc); } } else if ("order".equals(table)) { // 步骤 3: 【必须】实现转换逻辑，将 Map 转换为 OrderDocument for (Map<String, String> row : message.getData()) { OrderDocument doc = convertMapToOrderDocument(row); // 转换方法 esService.index("order_index", doc); }2.引入CanalStarter及EntryHandler实现类这种方法的实现起来还算可以，并没有特别复杂，主要在于耦合度很低，扩展性也很强。原理就是Handler来自动化字段反射和类型转换，以及Handler匹配字段和实体类通过注解其他同上Spring配置<dependency> <groupId>top.javatool</groupId> <artifactId>canal-spring-boot-starter</artifactId> <version>1.2.1-RELEASE</version> </dependency>创建Handler类@CanalTable("user") @Component @AllArgsConstructor @Slf4j public class NmsUserHandler implements EntryHandler<User> { private final RedisTemplate<Object,Object> redisTemplate; @Override public void insert(User nmsUser) { log.info("[新增]"+nmsUser.toString()); redisTemplate.opsForValue().set("user:"+nmsUser.getUsername(),nmsUser); Object o = redisTemplate.opsForValue().get("user:ns0"); System.out.println(o); } @Override public void update(User before, User after) { log.info("[更新]"+after.toString()); redisTemplate.opsForValue().set("user:"+after.getUsername(),after); Object o = redisTemplate.opsForValue().get("user:snn7"); System.out.println(o); } @Override public void delete(User nmsUser) { log.info("[删除]"+nmsUser.getUsername()); redisTemplate.delete("user:"+nmsUser.getUsername()); } }这里遇到了问题：项目的RedisTemplate的泛型是<String,Object> ，不符合这个同步的代码解决：在RedisConfig加入一个新的方法，叫别的名字，泛型返回值改成<Object,Object>，即可使用配置同步的表@AllArgsConstructor @NoArgsConstructor @Data @Table(name = "user") // 标明对应的数据库表名 public class User implements Serializable { private static final long serialVersionUID = 1L; @Column(name="uid") @TableId private Integer uid; @Column(name="username") private String username; @Column(name="password") private String password; //及其他 } 也就是说在实体类的上面加上@Table(name = "表名")，在每个属性上面加入 @Column(name="属性")效果2025-10-20 21:17:05.316 INFO 20616 --- [xecute-thread-1] c.c.backend.im.handler.NmsUserHandler : [更新]User{uid=1, username='cixingji', nickname='ccvxv', avatar='https://cube.elemecdn.com/9/c2/f0ee8a3c7c9638a54940382568c9dpng.png', background='https://tinypic.host/images/2023/11/15/69PB2Q5W9D2U7L.png', gender=2, description='这个人很懒，什么都没留下~', exp=0, coin=0.0, vip=0, state=0, role=1, auth=0, authMsg='', createDate=Thu Jul 17 21:00:34 CST 2025, deleteDate=null} nullCanal实现删重实现canal进行redis删除失败的时候，进行重试删除。两种策略策略1：即时重试，不需要添加新组件。策略2：异步补偿-->死信队列。需要添加MQ。策略1：// ... 注入 RedisTemplate ... @Service @Slf4j public class RedisService { @Retryable( value = {RedisConnectionFailureException.class, RedisBusyException.class}, // 指定触发重试的异常 maxAttempts = 3, // 最大重试次数 backoff = @Backoff(delay = 1000) // 延迟1秒后重试 ) public void delete(String key) { log.warn("Attempting to delete Redis key: {}", key); // 如果连接Redis失败，会抛出异常，触发重试 redisTemplate.delete(key); } // 增加一个 Recover 方法，用于最终失败后的处理（例如发送告警） @Recover public void recoverDelete(RedisException e, String key) { log.error("Final attempt to delete key {} failed! Error: {}", key, e.getMessage()); // 【关键】：这里是即时重试失败后的最终处理点 // 比如：发送邮件告警，或者写入一个本地的重试表 }策略1只能实现redis偶尔的抖动，网络小故障。准备使用策略2，可以应用于redis长时间宕机。为什么叫做死信/异补死信队列：消息在消费过程中，故障了或者失败了异步 ： 交给一个独立的服务实现。补偿 ： 独立的服务持续监听重试队列，等待Redis或ES恢复之后，再进行补偿的“重试”。一、引入Kafka1.配置<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>2.2.6.RELEASE</version> </dependency>spring: kafka: bootstrap-servers: localhost:9092 consumer: auto-offset-reset: latest max-poll-records: 100 max-partition-fetch-bytes: 10000002.二、定义补偿消息体DTO@Data @NoArgsConstructor @AllArgsConstructor public class SyncFailDTO implements Serializable { private String tableName; // 哪个表失败了 private String operationType; // INSERT, UPDATE, DELETE private String key; // Redis key (用于删除操作) private Object data; // 失败时要同步的完整数据对象（例如 User 对象） private Long timestamp = System.currentTimeMillis(); }三、修改 Canal Handler（生产者） // 引入 Kafka 模板 private final KafkaTemplate<String, Object> kafkaTemplate; // 注入泛型正确的 RedisTemplate<Object, Object> private final RedisTemplate<Object,Object> redisTemplate; // 定义补偿 topic 名称 private static final String COMPENSATION_TOPIC = "canal_redis_compensation"; private void handleRedisOperation(String opType, String key, User user) { try { log.info("[尝试 Redis 同步] Op:{} Key:{}", opType, key); // 核心逻辑：执行 Redis 操作 if ("DELETE".equals(opType)) { redisTemplate.delete(key); } else { // INSERT 或 UPDATE redisTemplate.opsForValue().set(key, user); } } catch (Exception e) { // 如果 Redis 操作失败（例如连接异常、宕机等），则进入补偿流程 log.error("[Redis 失败] 发送补偿消息到 Kafka. Key: {}", key, e); // 1. 构造失败消息体 SyncFailDTO failMessage = new SyncFailDTO("user", opType, key, user); // 发送完整的 User 对象，以便重试时可以正确 SET // 2. 发送到补偿队列 (使用 key 作为 Kafka key，确保同一 key 的消息有序) kafkaTemplate.send(COMPENSATION_TOPIC, key, failMessage); } }四、创建补偿消费者服务（消费者）public class RedisCompensationConsumer { // 注入 RedisTemplate (泛型需与 Canal Handler 中保持一致) private final RedisTemplate<Object,Object> redisTemplate; // 【可选】引入 Spring Retry 或自行实现重试机制 // 监听步骤 3 中定义的 topic @KafkaListener(topics = "canal_redis_compensation", groupId = "redis-compensation-group") public void listen(SyncFailMessage message) { String key = message.getKey(); String opType = message.getOperationType(); log.warn("[补偿开始] 尝试重新同步 key: {}，操作: {}", key, opType); try { // 核心逻辑：再次尝试 Redis 操作 if ("DELETE".equals(opType)) { redisTemplate.delete(key); } else { // INSERT 或 UPDATE // 这里的 data 需要进行类型转换，Spring Kafka 默认是 Map，需要反序列化回 User // 简化：如果之前发送的就是 User 对象（依赖正确的 JsonSerializer/Deserializer 配置） redisTemplate.opsForValue().set(key, message.getData()); } log.info("[补偿成功] key: {}", key); } catch (Exception e) { // 【关键】：如果补偿失败，则需要延迟或重复消费 // 1. Kafka 默认重试：如果抛出异常，Kafka 可能会根据配置自动重试几次。 // 2. 延迟重试：更高级的方案是，将此消息发送到一个**延迟队列**或**真正的死信队列（DLQ）**，等待更长时间再重试，防止对宕机的Redis造成持续压力。 log.error("[补偿失败] key: {}，将等待 Kafka 重试或进入下一步 DLQ 流程。", key, e); // 抛出异常，让 Kafka 消费者机制介入重试 (或配置手动 Ack/Nack) throw new RuntimeException("Redis compensation failed, waiting for Kafka retry.", e); } } }Canal实现同步ES引入ES通用工具类已有创建要同步的Document要同步哪个表的哪些字段，这里面就只需要有这些字段// UserDocument.java @Data @NoArgsConstructor @AllArgsConstructor public class UserDocument { @Id // 假设使用 userId 作为 ES 文档 ID private Long id; private String username; private String nickname; private String phone; // ... 仅包含需要ES搜索的字段 }在Handler类加入ES逻辑@CanalTable("user") @Component @AllArgsConstructor @Slf4j public class NmsUserHandler implements EntryHandler<User> { private final RedisTemplate<Object, Object> redisTemplate; private final EsService esService; // 【新增】注入 ES Service // --- 辅助方法：将 DB 实体转换为 ES Document --- private UserDocument convertToDocument(User user) { // 【核心逻辑】：实现数据转换和字段筛选 return new UserDocument( user.getUserId(), user.getUsername(), user.getNickname(), user.getPhone() ); } @Override public void insert(User user) { // Redis 逻辑... redisTemplate.opsForValue().set("user:" + user.getUsername(), user); // 【新增 ES 逻辑】 UserDocument doc = convertToDocument(user); esService.index("user_index", doc.getId().toString(), doc); } @Override public void update(User before, User after) { // Redis 逻辑... redisTemplate.opsForValue().set("user:" + after.getUsername(), after); // 【新增 ES 逻辑】 UserDocument doc = convertToDocument(after); esService.index("user_index", doc.getId().toString(), doc); // ES 索引是幂等的，可以直接用 index/upsert } @Override public void delete(User user) { // Redis 逻辑... redisTemplate.delete("user:" + user.getUsername()); // 【新增 ES 逻辑】 esService.delete("user_index", user.getUserId().toString()); } }效果Kafka学习（Canal）各部分组件如下各部分Producer生产者，负责创建和发送消息Consumer 消费者，负责阅读和处理消息Broker Kafka服务器，负责存储，接收，发送消息Topic 是在Broker中的一类数据的分类Partition Partition是Topic的物理切分，也就是分区，一个Partition相当于一个独立的通道一个独立的线程工作流程每条消息的的结构：Key+ValueKey是分区路由，用来保证同一个Partition的顺序性。在不指定Key的情况下，分区策略就会变成：早期版本是轮询（均匀的发送到Topic的所有Partition），新版本是粘性分区（生产者会尝试把连续的消息一直发给同一个Partition，直到满再换）指定Key，一般都是拿主键id来保障顺序性。1、生产者发送消息，把消息发送给Broker，并且指定Topic，指定Key。2、Broker接收到消息之后，根据Topic和Key，把消息追加写到指定Partition的末尾；消息写入成功后，就会拿到一个Offset偏移量，也就是这个消息在Partition的位置编号。3、消费者读取消息，向Broker请求订阅某个Topic，消费者是主动向Broker询问：有没有新消息。（订阅以Topic为单位）4、消费者处理完这批消息之后，告诉Broker：我已经看完了编号为xx的消息了。这个xx就是新的Offset，Broker会记录这个记录，重复3。