?一、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過濾
實現業務邏輯