一、Windows下开启MySQL binLog日志
首先要开启MySQL的BinLog 管理
show variables like '%log_bin%';
如果发现log_bin是OFF，打开mysql文件夹下面的my.ini，修改一下
在 [mysqld] 下面加
# 开启bin-log
log-bin=mysql-bin # 开启binlog功能
binlog-format=ROW # 设置binlog格式
server_id=1 # 设置服务ID号
保存之后重启服务
二、java代码实现
pom引入jar
<!--mysql监听-->
<dependency>
<groupId>com.github.shyiko</groupId>
<artifactId>mysql-binlog-connector-java</artifactId>
<version>0.21.0</version>
</dependency>
java代码
代码逻辑
监听表一定要有完整性标识字段，否则无法实现（业务数据会产生完整性标识，用于比对数据）
业务上删除必须是逻辑删除，物理删除都要被监听
package com.dahua.data.xdjaencrypt.business.controller;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* mysql bin log 日志监听
*/
@Component
@Slf4j
public class MySQLBinaryLogConfig {
//数据库表，需要监听的表
private static final List<String> TABLE_NAME = new ArrayList<String>(Arrays.asList("user", "user2", "table3"));
//监控数据库
private static final List<String> DATABASE = new ArrayList<String>(Arrays.asList("test"));
{
System.out.println("启动监听：启动中....");
getThread().start();
System.out.println("启动监听：成功...");
}
public Thread getThread() {
BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306, "root", "dahuacloud");
client.setServerId(1);
Map<Long, Object> concurrentHashMap = new ConcurrentHashMap<Long, Object>();
return new Thread(() -> {
client.registerEventListener(event -> {
String database = null;//监控数据库
String table = null; //监控表
final EventData data = event.getData();
System.out.println("数据监听开始..." + data);
if (data instanceof TableMapEventData) {
//把tableName 和tableId 关联上
TableMapEventData tableMapEventData = (TableMapEventData) data;
database = tableMapEventData.getDatabase();
System.out.println("监控数据库" + database);
table = tableMapEventData.getTable();
concurrentHashMap.put(tableMapEventData.getTableId(), tableMapEventData.getTable());
log.info("数据表:{},data:{},database:{}", table, data.toString(), database);
} else if (data instanceof UpdateRowsEventData) {
UpdateRowsEventData tableMapEventData = (UpdateRowsEventData) data;
if (TABLE_NAME.contains(concurrentHashMap.get(tableMapEventData.getTableId()))) {
System.out.println("业务操作,把after后的数据重新生成完整性标识，进行比对，匹配不入库，不匹配的数据入库");
System.out.println("修改:" + data);
}
;
} else if (data instanceof WriteRowsEventData) {
WriteRowsEventData tableMapEventData = (WriteRowsEventData) data;
if (TABLE_NAME.contains(concurrentHashMap.get(tableMapEventData.getTableId()))) {
System.out.println("业务操作,row中的数据重新生成完整性标识进行比对，匹配不入库，不匹配的数据入库");
System.out.println("添加:" + data);
}
} else if (data instanceof DeleteRowsEventData) {
DeleteRowsEventData tableMapEventData = (DeleteRowsEventData) data;
if (TABLE_NAME.contains(concurrentHashMap.get(tableMapEventData.getTableId()))) {
System.out.println("业务操作,所有物理删除的都要入库，所以业务定义删除的时候要，只支持逻辑删除，不支持物理删除");
System.out.println("删除:" + data);
}
}
if (!StringUtils.isAllBlank(table, database) && DATABASE.contains(database) && TABLE_NAME.contains(table)) {
log.info("<<<<<< 收到MySQL binLog 日志推送 >>>>>>>");
//开始编写具体的逻辑
log.info("监控数据库:{},监控表{}，操作类型{}", database, table);
}
});
try {
client.connect();
} catch (IOException e) {
e.printStackTrace();
}
});
}
}
各个监听产生的数据
修改监听-----------------------------------------------------------------------------
数据监听RotateEventData{binlogFilename='mysql-bin.000001', binlogPosition=6460}
数据监听FormatDescriptionEventData{binlogVersion=4, serverVersion='5.6.48-log', headerLength=19, dataLength=92, checksumType=CRC32}
数据监听QueryEventData{threadId=2, executionTime=0, errorCode=0, database='test', sql='BEGIN'}
数据监听TableMapEventData{tableId=70, database='test', table='user', columnTypes=3, 3, 3, 15, columnMetadata=0, 0, 0, 765, columnNullability={2}, eventMetadata=null}
数据监听UpdateRowsEventData{tableId=70, includedColumnsBeforeUpdate={0, 1, 2, 3}, includedColumns={0, 1, 2, 3}, rows=[
{before=[17, 0, 8, 88899], after=[17, 0, 8, 88810]}
]}
数据监听QueryEventData{threadId=2, executionTime=0, errorCode=0, database='test', sql='COMMIT'}
删除监听-----------------------------------------------------------------------------
数据监听QueryEventData{threadId=2, executionTime=0, errorCode=0, database='test', sql='BEGIN'}
数据监听TableMapEventData{tableId=70, database='test', table='user', columnTypes=3, 3, 3, 15, columnMetadata=0, 0, 0, 765, columnNullability={2}, eventMetadata=null}
数据监听DeleteRowsEventData{tableId=70, includedColumns={0, 1, 2, 3}, rows=[
[15, 0, 4, add]
]}
数据监听QueryEventData{threadId=2, executionTime=0, errorCode=0, database='test', sql='COMMIT'}
新增监听-----------------------------------------------------------------------------
数据监听QueryEventData{threadId=2, executionTime=0, errorCode=0, database='test', sql='BEGIN'}
数据监听TableMapEventData{tableId=70, database='test', table='user', columnTypes=3, 3, 3, 15, columnMetadata=0, 0, 0, 765, columnNullability={2}, eventMetadata=null}
数据监听WriteRowsEventData{tableId=70, includedColumns={0, 1, 2, 3}, rows=[
[18, 2, 3, 4]
]}
数据监听QueryEventData{threadId=2, executionTime=0, errorCode=0, database='test', sql='COMMIT'}
```
通过数据可以看出新增删除修改都是被监听多次
并且表名称不是每次都能获取到，所有要有tableid，关联表名称
通过tableid过滤
实现业务逻辑
