Java binLog日志监听方式

show variables like '%log_bin%';

# 开启bin-log
log-bin=mysql-bin # 开启binlog功能
binlog-format=ROW # 设置binlog格式
server_id=1 # 设置服务ID号

 <dependency>
     <groupId>com.github.shyiko</groupId>
     <artifactId>mysql-binlog-connector-java</artifactId>
     <version>0.21.0</version>
 </dependency>

import cn.hutool.core.collection.ListUtil;
import com.alibaba.fastjson2.JSON;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.web.controller.websocket.AlarmWebSocket;
import com.ruoyi.web.service.IWidfireDataService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.List;
import java.util.Map;


/**
 * mysql bin log 日志监听
 */
@Component
@Slf4j
public class MySQLBinaryLogConfig {


    public static IWidfireDataService widfireDataService;

    @Autowired
    public void setSenderService(IWidfireDataService widfireDataService){
        MySQLBinaryLogConfig.widfireDataService= widfireDataService;
    }


    private static final List<String> TABLE_NAME = ListUtil.of("alart_ai"); //数据库表，需要监听的表


    {
        System.out.println("启动监听：启动中");
        getThread().start();
        System.out.println("启动监听：成功");
    }

    public Thread getThread() {
        BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306, "root", "123456");
        client.setServerId(1);
        return new Thread(() -> {
            client.registerEventListener(event -> {
                String table =null;
                final EventData data = event.getData();
                if (data instanceof TableMapEventData) {
                    TableMapEventData tableMapEventData = (TableMapEventData) data;
                    String database = tableMapEventData.getDatabase();
                    table = tableMapEventData.getTable();

                    log.info("数据表:{},data:{},database:{}",table,data.toString(),database);
                }else if (data instanceof UpdateRowsEventData) {
                    UpdateRowsEventData tableMapEventData = (UpdateRowsEventData) data;

                    System.out.println("更新:");
                } else if (data instanceof WriteRowsEventData) {
                    System.out.println("添加:");
                } else if (data instanceof DeleteRowsEventData) {
                    System.out.println("删除:");
                }

                if(StringUtils.isNotEmpty(table) && TABLE_NAME.contains(table)){
                    log.info("<<<<<< 收到MySQL binLog 日志推送 >>>>>>>");
                    //开始编写具体的逻辑
                    
                }

            });
            try {
                client.connect();
            } catch (IOException e) {
                e.printStackTrace();
            }
        });

    }
}