Java?binLog日志監(jiān)聽方式

更新時(shí)間：2024年11月15日 10:40:51 作者：Liu.jie

文章介紹了如何在Windows下開啟MySQL的binLog日志,并提供了一個(gè)Java代碼示例,演示如何監(jiān)聽指定的表并進(jìn)行處理邏輯

Java binLog日志監(jiān)聽

監(jiān)聽指定的表去做一些處理邏輯，首先是要開啟M有SQL的配置，然后再擼代碼。

一、Windows下開啟MySQL binLog日志

首先要開啟MySQL的BinLog 管理

show variables like '%log_bin%';

如果發(fā)現(xiàn)是OFF，打開mysql文件夾下面的my.ini，修改一下

如果不知道m(xù)y.ini 在哪里，打開【服務(wù)】-> 右擊屬性

拉到最后就可以看見my.ini，然后找到文件后

在 [mysqld] 下面加

# 開啟bin-log
log-bin=mysql-bin # 開啟binlog功能
binlog-format=ROW # 設(shè)置binlog格式
server_id=1 # 設(shè)置服務(wù)ID號

然后 重啟服務(wù)，就會發(fā)現(xiàn)已經(jīng)起好了

二、Java代碼示例演示

首先引入Maven包

 <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 日志監(jiān)聽
 */
@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"); //數(shù)據(jù)庫表，需要監(jiān)聽的表


    {
        System.out.println("啟動監(jiān)聽：啟動中");
        getThread().start();
        System.out.println("啟動監(jiān)聽：成功");
    }

    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("數(shù)據(jù)表:{},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();
            }
        });

    }
}