1.首先mysql要開啟binlog，修改mysql的配置文件：

server_id=2023  #主節(jié)點(diǎn)的編號，保證唯一即可
log_bin=mysql-bin
binlog_format=row

2.使用mysql-binlog-connector-java來監(jiān)聽MySQL的二進(jìn)制日志（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;
    }
}

自己實(shí)現(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等消息隊(duì)列，推送數(shù)據(jù)到es等數(shù)據(jù)倉庫