1.首先mysql要開啟binlog白胀,修改mysql的配置文件:
server_id=2023 #主節(jié)點的編號熟菲,保證唯一即可
log_bin=mysql-bin
binlog_format=row
2.使用mysql-binlog-connector-java來監(jiān)聽MySQL的二進制日志(binlog)事件毡琉,引入依賴谜嫉。
<dependency>
<groupId>com.github.shyiko</groupId>
<artifactId>mysql-binlog-connector-java</artifactId>
<version>0.29.2</version>
</dependency>
定義配置:
mysql:
binlog:
host: localhost
port: 3306
schema: heg_hotel
username: root
password: 123456
tables: t_user,t_order
3.代碼中配置:
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
import com.stephen.listener.CustomerBinlogListener;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
/**
* @Description 開啟binlog監(jiān)聽工具
* @Author jack
* @Date 2024/8/22 14:35
*/
@Configuration
public class BinlogConfig {
@Value("${mysql.binlog.host}")
private String host;
@Value("${mysql.binlog.port}")
private Integer port;
@Value("${mysql.binlog.username}")
private String userName;
@Value("${mysql.binlog.password}")
private String password;
@Value("${mysql.binlog.schema:#{null}}")
private String schema;
@Value("${mysql.binlog.tables}")
private String tables;
@Bean
public CustomerBinlogListener customerBinlogListener() {
//考慮到多個數(shù)據(jù)庫的情況
if(StringUtils.isNotBlank(schema)){
List<String> schemaTables = Lists.newArrayList();
//只監(jiān)聽指定庫 例如:demo.user
Arrays.stream(tables.split(",")).forEach(table-> schemaTables.add(Joiner.on('.').join(schema,table)));
return new CustomerBinlogListener(schemaTables);
}
else{
//監(jiān)聽多個庫
return new CustomerBinlogListener(Arrays.asList(tables.split(",")));
}
}
@Bean
public BinaryLogClient binaryLogClient() {
//不指定schema監(jiān)聽所有的數(shù)據(jù)庫
BinaryLogClient client = new BinaryLogClient(host,port,userName,password);
//BinaryLogClient client = new BinaryLogClient(host,port,schema,userName,password);
client.registerEventListener(customerBinlogListener());
client.setServerId(1);
client.setKeepAlive(true); // 保持連接
client.setKeepAliveInterval(10 * 1000); // 心跳包發(fā)送頻率
client.setKeepAliveConnectTimeout(5 * 1000); // 心跳發(fā)送超時設(shè)置
try {
client.connect();
} catch (IOException e) {
throw new RuntimeException(e);
}
return client;
}
}
自己實現(xiàn)監(jiān)聽事件:
import com.alibaba.fastjson.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.google.common.collect.Maps;
import com.stephen.model.BinlogDto;
import lombok.extern.slf4j.Slf4j;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @Description 自定義監(jiān)聽數(shù)據(jù)
* @Author jack
* @Date 2024/8/22 15:05
*/
@Slf4j
public class CustomerBinlogListener implements BinaryLogClient.EventListener {
private HashMap<Long, String> tableMap = Maps.newHashMap();
private List<String> databaseTables;
public CustomerBinlogListener(List<String> databaseTables){
this.databaseTables = databaseTables;
}
@Override
public void onEvent(Event event) {
// binlog事件
EventData data = event.getData();
if (data != null) {
if (data instanceof TableMapEventData) {
TableMapEventData tableMapEventData = (TableMapEventData) data;
tableMap.put(tableMapEventData.getTableId(), tableMapEventData.getDatabase() + "." + tableMapEventData.getTable());
}
// update數(shù)據(jù)
if (data instanceof UpdateRowsEventData) {
UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
String tableName = tableMap.get(updateRowsEventData.getTableId());
if (tableName != null && databaseTables.contains(tableName)) {
String eventKey = tableName + ".update";
System.out.println("212121212121");
for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
String msg = JSON.toJSONString(new BinlogDto(eventKey, row.getValue()));
log.info("binlog修改日志:{}",msg);
}
}
}
// insert數(shù)據(jù)
else if (data instanceof WriteRowsEventData) {
WriteRowsEventData writeRowsEventData = (WriteRowsEventData) data;
String tableName = tableMap.get(writeRowsEventData.getTableId());
if (tableName != null && databaseTables.contains(tableName)) {
String eventKey = tableName + ".insert";
for (Serializable[] row : writeRowsEventData.getRows()) {
String msg = JSON.toJSONString(new BinlogDto(eventKey, row));
log.info("binlog插入日志:{}",msg);
}
}
}
// delete數(shù)據(jù)
else if (data instanceof DeleteRowsEventData) {
DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) data;
String tableName = tableMap.get(deleteRowsEventData.getTableId());
if (tableName != null && databaseTables.contains(tableName)) {
String eventKey = tableName + ".delete";
for (Serializable[] row : deleteRowsEventData.getRows()) {
String msg = JSON.toJSONString(new BinlogDto(eventKey, row));
log.info("binlog刪除日志:{}",msg);
}
}
}
}
}
}
可以結(jié)合kafka等消息隊列骄酗,推送數(shù)據(jù)到es等數(shù)據(jù)倉庫