免费个人网站建设报价_网站建设合同 完整版_百度广告代运营_软考培训机构排名

基于mysql binlog，监听数据变化

mysql开启binlog配置

如果是新建docker 容器mysql,启动命令：

docker run --name mysql-binlog -p 3306:3306 -e MYSQL_ROOT_PASSWORD=123456 -d mysql:8.0.34  --log-bin=mysql-bin --binlog-format=ROW

也可以修改my.ini配置
数据库账户需要拥有较高的权限，如：管理员权限

一、引入依赖

        <dependency><groupId>com.zendesk</groupId><artifactId>mysql-binlog-connector-java</artifactId><version>0.27.1</version></dependency>

二、监听器

import com.alibaba.fastjson.JSONObject;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;
import lombok.extern.slf4j.Slf4j;import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;@Component
@Slf4j
public class MysqlBinLogClient implements ApplicationRunner {@Overridepublic void run(ApplicationArguments args)  {new Thread(() -> {//项目启动完成连接bin-logtry {connectMysqlBinLog();} catch (IOException e) {throw new RuntimeException(e);}}).start();}// 要监听的数据库private String databaseName = "lobster";// 要监听的表private String tableName = "demo";/*** 连接mysqlBinLog*/public void connectMysqlBinLog() throws IOException {log.info("监控BinLog服务已启动");BinaryLogClient client = new BinaryLogClient("192.168.0.194", 3306,"root", "123456");
//        client.setServerId(1); //和自己之前设置的server-id保持一致Map<Long, String> tableIdToTableInfo = new HashMap<>();client.registerEventListener(event -> {EventData data = event.getData();if (data instanceof TableMapEventData) {//只要连接的MySQL发生的增删改的操作，则都会进入这里，无论哪个数据库TableMapEventData tableMapEventData = (TableMapEventData) data;if (databaseName.equals(tableMapEventData.getDatabase())){//可以通过转成TableMapEventData类实例的tableMapEventData来获取当前发生变更的数据库if (tableName.equals(tableMapEventData.getTable())) {tableIdToTableInfo.put(tableMapEventData.getTableId(), databaseName + ":" + tableName);}}}//表数据新增时触发if (data instanceof WriteRowsEventData) {WriteRowsEventData writeData = (WriteRowsEventData) data;writeData.getTableId();if ((databaseName + ":" + tableName).equals(tableIdToTableInfo.get(writeData.getTableId()))){List<Serializable[]> rows = writeData.getRows();for (Serializable[] row : rows) {System.out.println("Insert:");System.out.println(JSONObject.toJSONString(row));}}}//表数据发生修改时触发if (data instanceof UpdateRowsEventData) {System.out.println("Update:");System.out.println(data.toString());//表数据发生插入时触发} else if (data instanceof DeleteRowsEventData) {System.out.println("Delete:");System.out.println(data.toString());}});try {client.connect();} finally {client.disconnect();}}
}

三、监听效果