数据变化监听实现

前言

今天纯粹是无聊随便看看,然后发现有个好东西,可能可以用于某些数据变化主动监听场景搞事情.废话不多说,直接上货.

一、Binlog4j是什么

轻量级 Mysql Binlog 客户端，提供宕机续读，高可用集群等特性

二、项目特性

集群模式，通过集群部署的方式，保证服务高可用。
宕机续读，避免宕机期间造成数据丢失
数据转换，基于泛型封装 binlog Event 的序列化数据
兼容传统项目与 Spring Boot / Cloud 项目
兼容 Spring Boot 2.x 与 Spring Boot 3.x 版本

三、使用步骤

1.引入库

<dependency><groupId>com.gitee.Jmysy</groupId><artifactId>binlog4j-core</artifactId><version>latest.version</version>
</dependency>
<dependency><groupId>com.gitee.Jmysy</groupId><artifactId>binlog4j-spring-boot-starter</artifactId><version>latest.version</version>
</dependency>

2.使用

通过 BinlogClient 创建 binlog 客户端，IBinlogEventHandler 用于接受 binlog 事件通知，该接口允许使用泛型，数据将遵循驼峰规则进行封装。

public class BootStrap {public static void main(String[] args) {BinlogClientConfig clientConfig = new BinlogClientConfig();clientConfig.setHost("127.0.0.1");clientConfig.setPort(3306);clientConfig.setUsername("root");clientConfig.setPassword("taoren@123");clientConfig.setServerId(1990);IBinlogClient binlogClient = new BinlogClient(clientConfig);binlogClient.registerEventHandler("database", "table", new IBinlogEventHandler() {@Overridepublic void onInsert(BinlogEvent event) {System.out.println("插入数据:{}", event.getData());}@Overridepublic void onUpdate(BinlogEvent event) {System.out.println("修改数据:{}", event.getData());}@Overridepublic void onDelete(BinlogEvent event) {System.out.println("删除数据:{}", event.getData());}});binlogClient.connect();}
}

3、高级特性使用

通过 Persistence 配置为 true 启用宕机续读功能，Binlog4j 会将 binlog 的 filename 与 position 记录到 redis, 所以同时你需要设置 Redis 配置.
先在 application.yml 中填写 BinlogClient 配置

spring:binlog4j:redis-config:host: 127.0.0.1port: 6379password: taoren@123client-configs:master:username: rootpassword: taoren@123host: 127.0.0.1port: 3306serverId: 1990

public class BootStrap {public static void main(String[] args) {RedisConfig redisConfig = new RedisConfig();redisConfig.setHost("127.0.0.1");redisConfig.setPort(6379);redisConfig.setPassword("taoren@123");BinlogClientConfig clientConfig = new BinlogClientConfig();clientConfig.setHost("127.0.0.1");clientConfig.setPort(3306);clientConfig.setUsername("root");clientConfig.setPassword("taoren@123");clientConfig.setServerId(1990); // Client 编号clientConfig.setRedisConfig(redisConfig); // Redis 配置clientConfig.setPersistence(true); // 启用持久化 (宕机重启后, 从上次读取的位置开始)clientConfig.setMode(BinlogClientMode.cluster); // 高可用集群BinlogClient binlogClient = new BinlogClient(clientConfig);binlogClient.registerEventHandler("database", "table", new IBinlogEventHandler<User>() {@Overridepublic void onInsert(BinlogEvent<User> event) {System.out.println("插入数据:{}", event.getData());}@Overridepublic void onUpdate(BinlogEvent<User> event) {System.out.println("修改数据:{}", event.getData());}@Overridepublic void onDelete(BinlogEvent<User> event) {System.out.println("删除数据:{}", event.getData());}});binlogClient.connect();}
}

单表监测
使用 @BinlogSubscriber 注解，指定 IBinlogEventHandler 需要注册到哪个客户端，并且指定监听的 database 与 table

@BinlogSubscriber(clientName = "master", database = "pear-admin", table ="sys_user")
public class UserEventHandler implements IBinlogEventHandler<User> {@Overridepublic void onInsert(BinlogEvent<User> event) {System.out.println("插入数据：" + event.getData());}@Overridepublic void onUpdate(BinlogEvent<User> event) {System.out.println("修改数据:" + event.getData());}@Overridepublic void onDelete(BinlogEvent<User> event) {System.out.println("删除数据:" + event.getData());}}

多表监测
@BinlogSubscriber 注解 database 与 table 属性支持 Pattern 匹配，IBinlogEventHandler 在不指定泛型的情况下，event.getData () 为 Map<String, Object> 类型，用于兼容不同表的数据结构

@BinlogSubscriber(clientName = "master", database = ".*", table ="sys.*")
public class UserEventHandler implements IBinlogEventHandler {@Overridepublic void onInsert(BinlogEvent event) {System.out.println("数据库：" + event.getDatabase());System.out.println("数据表：" + event.getTable());System.out.println("新数据：" + event.getData());}@Overridepublic void onUpdate(BinlogEvent event) {System.out.println("数据库：" + event.getDatabase());System.out.println("数据表：" + event.getTable());System.out.println("原数据：" + event.getOriginalData());System.out.println("新数据：" + event.getData());}@Overridepublic void onDelete(BinlogEvent event) {System.out.println("数据库：" + event.getDatabase());System.out.println("数据表：" + event.getTable());System.out.println("新数据：" + event.getData());}}