Java 实现实时监听MySQL数据库变更MySQLBinListener

约定不等于承诺〃 2024-03-17 10:17 52阅读 0赞

**目录**

1、导出需要的类和接口

2、 定义 MySQLBinlogListener类

3、私有方法，启动重连定时器

4、完整代码

![b8a45eb3aec74b02abbcb3f6d2e12c07.gif][]

编写一个MySQL数据库实时变更的监听器。

为什么要编写这个一个监听器：为了实时监测和响应MySQL数据库中的变更事件

> 1.  实时数据同步：通过监听MySQL Binlog，可以捕获数据库的变更操作，例如插入、更新、删除等，从而能够实时地获取数据的变动情况。这对于需要及时同步数据的应用场景非常重要，例如实时分析、数据同步等
> 2.  数据库监控和审计：通过监听数据库的变更事件，可以实现对数据库的实时监控和审计功能。你可以捕获和记录数据库中的每个操作，了解数据库的变更情况，同时也方便进行故障排查和安全审计。
> 3.  数据变更触发业务逻辑：当数据库中的数据发生变化时，可以通过监听器触发相应的业务逻辑。例如，在某个表发生变更时，可以发送通知、调用其他服务或者进行数据处理等操作。
> 4.  数据缓存和更新：通过监听数据库变更事件，可以及时更新应用程序中的缓存数据，提高应用程序的性能和响应速度。当数据库数据变化时，可以通过监听器自动刷新缓存，确保应用程序使用的数据是最新的。

总之：写这样的监听器可以提供实时数据同步、数据库监控和审计、业务逻辑触发、数据缓存更新以及异构数据集成等多种好处。

那我们就开始来写一个这么的功能：

### 1、导出需要的类和接口 ###

import com.github.shyiko.mysql.binlog.BinaryLogClient;
    import com.github.shyiko.mysql.binlog.event.*;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.stereotype.Component;
    
    import java.io.IOException;
    import java.util.Timer;
    import java.util.TimerTask;

接下来，使用一个Spring组件标记

> @Component

### 2、 定义 `MySQLBinlogListener`类 ###

> public class MySQLBinlogListener \{

定义一个常量，表明定义的是重连间隔时间，单位可以设为毫秒。

private static final int RECONNECT_INTERVAL = 10000;

定义一个定时任务，用于定时尝试重新连接MySQL服务器

private static Timer reconnectTimer;

定义一个BinaryLogClient对象，用于连接到MySQL服务器

private static BinaryLogClient client;

定义一个用于依赖注入的ApplicationContext对象

@Autowired
    private static ApplicationContext applicationContext;

编辑程序的入口方法，它会调用`startMySQLBinlogListener`方法来开始监听MySQL Binlog事件

public void main(String[] args) {
        startMySQLBinlogListener();
    }

这是`startMySQLBinlogListener`方法，它用于启动MySQL Binlog监听器

public static void startMySQLBinlogListener() {

创建一个BinaryLogClient对象，指定MySQL服务器的主机名、端口号、数据库名、用户名和密码。

client = new BinaryLogClient("127.0.0.1",3306, "你的数据库表","root", "密码");

设置BinaryLogClient对象的一些属性，包括保持连接、心跳包发送间隔和心跳包连接超时时间。

client.setKeepAlive(true);
    client.setKeepAliveInterval(60 * 1000);
    client.setKeepAliveConnectTimeout(5 * 1000);

注册事件监听器，对不同类型的事件做出响应。在这里，我们对TableMapEventData类型的事件进行处理，根据表名发送WebSocket消息；对UpdateRowsEventData、WriteRowsEventData、DeleteRowsEventData类型的事件进行输出日志。

client.registerEventListener(event -> {
        try {
            EventData data = event.getData();
            if (data instanceof TableMapEventData) {
                TableMapEventData tableMapEventData = (TableMapEventData) data;
                String database = tableMapEventData.getDatabase();
                String table = tableMapEventData.getTable();
                WebsocketService websocketService = new WebsocketService();
                if ("你的数据库表".equalsIgnoreCase(table)) {
                    websocketService.groupSending("你的数据库表字段");
                }
                if ("你的数据库表".equalsIgnoreCase(table)) {
                    websocketService.groupSending("你的数据库表字段");
                }
            } else if (data instanceof UpdateRowsEventData) {
                System.out.println("更新:");
                System.out.println(data.toString());
            } else if (data instanceof WriteRowsEventData) {
                System.out.println("添加:");
                System.out.println(data.toString());
            } else if (data instanceof DeleteRowsEventData) {
                System.out.println("删除:");
                System.out.println(data.toString());
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    });

注册生命周期监听器，用于处理连接成功、通信异常、事件数据解析异常和连接断开等情况

client.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
        @Override
        public void onConnect(BinaryLogClient client) {
            System.out.println("MySQL数据库已连接！");
        }
    
        @Override
        public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
            System.out.println("已执行通信故障方法！");
            ex.printStackTrace();
        }
    
        @Override
        public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
            System.out.println("已执行反质化方法！");
            ex.printStackTrace();
        }
    
        @Override
        public void onDisconnect(BinaryLogClient client) {
            System.out.println("MySQL数据库已断开！");
            startReconnectTimer();
        }
    });

### 3、私有方法，启动重连定时器 ###

该功能如下：

private static void startReconnectTimer() {

如果已存在重连定时器，先取消之前的定时任务

if (reconnectTimer != null) {
        reconnectTimer.cancel();
    }

创建一个新的重连定时器，并执行定时任务。定时任务中会尝试重新连接MySQL服务器

reconnectTimer = new Timer();
    reconnectTimer.schedule(new TimerTask() {
        @Override
        public void run() {
            boolean isConnected = client != null && client.isConnected();
            try {
                if (isConnected) {
                    System.out.println("和数据库断开连接。重连。。。。。。");
                    client.disconnect();
                }
                client.connect();
            } catch (IOException e) {
                e.printStackTrace();
                startReconnectTimer();
            }
        }
    }, RECONNECT_INTERVAL);
    }

这就是整段代码的解释，它实现了一个MySQL Binlog监听器，能够监听MySQL数据库的变更事件并做出相应的处理。

### 4、完整代码 ###

需要私聊。。。。。。

[b8a45eb3aec74b02abbcb3f6d2e12c07.gif]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/17/ea7c2d892de64af0b57186789ba2826d.png