- 微信
- 微博
  
  分享文章到微博
- 复制链接
  
  复制链接到剪贴板

MySQL：基于Spring监听Binlog日志

不惑发表于 2024/10/31 08:50:49 2024/10/31

【摘要】 MySQL 的二进制日志（binlog）有三种不同的格式，通常被称为 binlog 模式。这三种模式分别是 Statement 模式、Row 模式和Mixed 模式。

binlog的三种模式

MySQL 的二进制日志（binlog）有三种不同的格式，通常被称为 binlog 模式。这三种模式分别是 Statement 模式、Row 模式和Mixed 模式。

Statement 模式：

在 Statement 模式下，MySQL 记录每个会更改数据的 SQL 语句。
binlog 记录的是执行的 SQL 语句本身，而不是具体的数据变化。
例如，如果执行了 UPDATE 语句，binlog 记录的是这个 UPDATE 语句的文本。

Row 模式：

在 Row 模式下，MySQL 记录每一行数据的变化。
binlog 记录的是行数据的变化，而不是 SQL 语句。
例如，如果执行了 UPDATE 语句，binlog 记录的是被修改的行的实际数据。

Mixed 模式：

Mixed 模式是 Statement 模式和 Row 模式的结合。
在 Mixed 模式下，MySQL 根据执行的 SQL 语句的类型来决定是记录语句还是记录行。
通常，对于简单的语句，使用 Statement 模式，对于涉及到行变化的复杂语句，使用 Row 模式。

这些模式可以通过 MySQL 配置文件中的 binlog_format 参数进行配置。例如：

[mysqld]
binlog_format=mixed

其中，statement、row 和 mixed 分别代表 Statement 模式、Row 模式和 Mixed 模式。选择适当的 binlog 模式取决于应用的特定需求和性能要求。不同的模式具有不同的优劣势，例如，Statement 模式可能会更轻量，而 Row 模式可能提供更详细的数据变化信息。

以Mixed 为例

查看binlog是否开启

show variables like '%log_bin%'

启动springboot程序

新建数据库

这个事件是一个 binlog 事件，其内容表示一个 SQL 查询事件。让我解释一下这个事件的各个部分：

事件类型 (eventType): 该事件的类型是 QUERY，表示这是一个 SQL 查询事件。
时间戳 (timestamp): 事件的时间戳为 1700045267000，表示事件发生的时间。
线程ID (threadId): 线程ID 是 189，表示执行这个查询的线程的标识符。
执行时间 (executionTime): 执行时间为 0，表示执行这个查询所花费的时间。
错误代码 (errorCode): 错误代码为 0，表示查询执行没有错误。
数据库 (database): 数据库为 test2023，表示这个查询发生在 test2023 数据库中。
SQL 查询 (sql): 实际的 SQL 查询为 CREATE DATABASE test2023 CHARACTER SET utf8 COLLATE utf8_general_ci，表示执行了创建数据库的操作。

这个事件的作用是在 test2023 数据库中执行了一个创建数据库的 SQL 查询。这是 binlog 中的一部分，用于记录数据库中的变化，以便进行数据备份、主从同步等操作。

新建表数据

CREATE TABLE `t_user` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `userName` varchar(100) NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8mb4;

这个事件也是一个 binlog 事件，表示一个 SQL 查询事件。让我解释一下这个事件的各个部分：

事件类型 (eventType): 该事件的类型是 QUERY，表示这是一个 SQL 查询事件。
时间戳 (timestamp): 事件的时间戳为 1700045422000，表示事件发生的时间。
线程ID (threadId): 线程ID 是 204，表示执行这个查询的线程的标识符。
执行时间 (executionTime): 执行时间为 0，表示执行这个查询所花费的时间。
错误代码 (errorCode): 错误代码为 0，表示查询执行没有错误。
数据库 (database): 数据库为 test2023，表示这个查询发生在 test2023 数据库中。
SQL 查询 (sql): 实际的 SQL 查询为 CREATE TABLE t_user(idbigint(20) NOT NULL AUTO_INCREMENT,userName varchar(100) NOT NULL, PRIMARY KEY (id) ) ENGINE=InnoDB AUTO_INCREMENT=10 DEFAULT CHARSET=utf8mb4，表示执行了在 test2023 数据库中创建名为 t_user 的表的操作。

这个事件的作用是在 test2023 数据库中创建了一个名为 t_user 的表，该表包含 id 和 userName 两个字段，其中 id 是自增的主键。这种类型的事件常常用于记录数据库结构的变化，以便进行数据备份、迁移和版本控制等操作。

插入表数据

INSERT INTO `test2023`.`t_user` (`id`, `userName`)
VALUES
    (
        "10086",
        "用心记录技术，走心分享，始于后端，不止于后端，励志成为一名优秀的全栈架构师，真正的实现码中致富。"
    );

这个事件也是一个 binlog 事件，表示一个 SQL 查询事件，具体如下：

事件类型 (eventType): 该事件的类型是 QUERY，表示这是一个 SQL 查询事件。
时间戳 (timestamp): 事件的时间戳为 1700045547000，表示事件发生的时间。
线程ID (threadId): 线程ID 是 204，表示执行这个查询的线程的标识符。
执行时间 (executionTime): 执行时间为 0，表示执行这个查询所花费的时间。
错误代码 (errorCode): 错误代码为 0，表示查询执行没有错误。
数据库 (database): 数据库为 test2023，表示这个查询发生在 test2023 数据库中。
SQL 查询 (sql): 实际的 SQL 查询为 INSERT INTO test2023.t_user (id, userName) VALUES ( "10086", "用心记录技术，走心分享，始于后端，不止于后端，励志成为一名优秀的全栈架构师，真正的实现码中致富。"，表示执行了向 test2023 数据库的 t_user 表中插入一行数据的操作。

这个事件的作用是向 t_user 表中插入了一行数据，包含了 id 和 userName 两个字段的值。这种类型的事件通常用于记录数据的变化，以便进行数据备份、同步和迁移等操作。

修改表数据修改表数据

修改表数据

UPDATE `test2023`.`t_user`
SET `id` = '10086',
 `userName` = '我的修改数据！！！'
WHERE
	(`id` = '10086');

这个事件同样是一个 binlog 事件，表示一个 SQL 查询事件，具体如下：

事件类型 (eventType): 该事件的类型是 QUERY，表示这是一个 SQL 查询事件。
时间戳 (timestamp): 事件的时间戳为 1700045675000，表示事件发生的时间。
线程ID (threadId): 线程ID 是 204，表示执行这个查询的线程的标识符。
执行时间 (executionTime): 执行时间为 0，表示执行这个查询所花费的时间。
错误代码 (errorCode): 错误代码为 0，表示查询执行没有错误。
数据库 (database): 数据库为 test2023，表示这个查询发生在 test2023 数据库中。
SQL 查询 (sql): 实际的 SQL 查询为 UPDATE test2023.t_userSETid= '10086', userName = '我的修改数据！！！' WHERE (id = '10086')，表示执行了更新 test2023 数据库中的 t_user 表中一行数据的操作。

这个事件的作用是将 t_user 表中 id 为 10086 的行的数据进行更新，将 id 修改为 10086，userName 修改为 ‘我的修改数据！！！’。这种类型的事件通常用于记录数据的变化，以便进行数据备份、同步和迁移等操作。

删除表数据

DELETE
FROM
	t_user
WHERE
	id = '10086';

这个事件同样是一个 binlog 事件，表示一个 SQL 查询事件，具体如下：

事件类型 (eventType): 该事件的类型是 QUERY，表示这是一个 SQL 查询事件。
时间戳 (timestamp): 事件的时间戳为 1700045755000，表示事件发生的时间。
线程ID (threadId): 线程ID 是 204，表示执行这个查询的线程的标识符。
执行时间 (executionTime): 执行时间为 0，表示执行这个查询所花费的时间。
错误代码 (errorCode): 错误代码为 0，表示查询执行没有错误。
数据库 (database): 数据库为 test2023，表示这个查询发生在 test2023 数据库中。
SQL 查询 (sql): 实际的 SQL 查询为 DELETE FROM t_user WHERE id = '10086'，表示执行了删除 test2023 数据库中的 t_user 表中一行数据的操作。

这个事件的作用是删除 t_user 表中 id 为 10086 的行。这种类型的事件通常用于记录数据的删除操作，以便进行数据备份、同步和迁移等操作。

总结：
binlog_format 设置为 mixed 时，对于 INSERT、UPDATE 和 DELETE 操作，它们在 binlog 中的事件类型都会被表示为 QUERY 事件。这是因为在 mixed 模式下，MySQL 使用了不同的方式来记录不同类型的操作，但在 binlog 中，它们都被包装成了 QUERY 事件。

在 mixed 模式下：

对于某些语句级别的操作（例如非确定性的语句或不支持事务的存储引擎），会使用 STATEMENT 事件。
对于其他一些情况，会使用 ROW 事件，将变更的行作为事件的一部分进行记录。

这就是为什么看到的 INSERT、UPDATE 和 DELETE 操作的事件类型都是 QUERY。在处理这些事件时，需要根据具体的 SQL 查询语句或其他信息来确定操作的类型。

源码示例

pom.xml

		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-data-jpa</artifactId>
		</dependency>
		
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-web</artifactId>
		</dependency>
		
		<dependency>
			<groupId>org.springframework.boot</groupId>
			<artifactId>spring-boot-starter-test</artifactId>
			<scope>test</scope>
		</dependency>
		
		<dependency>
			<groupId>mysql</groupId>
			<artifactId>mysql-connector-java</artifactId>
			<version>5.1.48</version> <!-- 查看最新版本 -->
		</dependency>
			
		<dependency>
			<groupId>com.github.shyiko</groupId>
			<artifactId>mysql-binlog-connector-java</artifactId>
			<version>0.21.0</version>
		</dependency>

Java示例

package com.example.demo.listener;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.Event;
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.QueryEventData;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.naming.AuthenticationException;
import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import java.util.concurrent.TimeoutException;

public class BinlogListenerMixed {

    private static final Logger logger = LoggerFactory.getLogger(BinlogListenerMixed.class);

    private static final String MYSQL_HOST = "8.130.74.105";
    private static final int MYSQL_PORT = 3306;
    private static final String MYSQL_USERNAME = "root";
    private static final String MYSQL_PASSWORD = "zhang.ting.123";

    public static void main(String[] args) {
        try {
            BinaryLogClient client = new BinaryLogClient(MYSQL_HOST, MYSQL_PORT, MYSQL_USERNAME, MYSQL_PASSWORD);
//            client.setBinlogFilename(null);
//            client.setBinlogPosition(-1); // 或者设置为其他适当的初始位置
//            client.setServerId(1);
//            client.setBinlogFilename("mysql-bin.000005");
//            client.setBinlogPosition(154);
            EventDeserializer eventDeserializer = new EventDeserializer();
            eventDeserializer.setCompatibilityMode(
                    EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                    EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
            );
            logger.info("使用主机={}, 端口={}, 用户名={}, 密码={} 连接到 MySQL", MYSQL_HOST, MYSQL_PORT, MYSQL_USERNAME, MYSQL_PASSWORD);
            client.setEventDeserializer(eventDeserializer);
            client.registerEventListener(BinlogListenerMixed::handleEvent);
            client.registerLifecycleListener(new BinaryLogClient.LifecycleListener() {
                @Override
                public void onConnect(BinaryLogClient client) {
                    logger.info("Connected to MySQL server");
                }

                @Override
                public void onCommunicationFailure(BinaryLogClient client, Exception ex) {
                    logger.error("Communication failure with MySQL server", ex);
                }

                @Override
                public void onEventDeserializationFailure(BinaryLogClient client, Exception ex) {
                    logger.error("Event deserialization failure", ex);
                }

                @Override
                public void onDisconnect(BinaryLogClient client) {
                    logger.warn("Disconnected from MySQL server");
                    // 在这里添加重新连接或其他处理逻辑
                }
            });

            client.connect();
        } catch (IOException e) {
            logger.error("@@ 连接到 MySQL 时发生错误", e);
            logger.error("@@ Error connecting to MySQL", e);
        }
    }

    private static void handleEvent(Event event) {
        logger.info("@@ 打印 event: {}", event);
        logger.info("@@ Received event type: {}", event.getHeader().getEventType());

        switch (event.getHeader().getEventType()) {
            case WRITE_ROWS:
            case EXT_WRITE_ROWS:
                handleWriteRowsEvent((WriteRowsEventData) event.getData());
                break;
            case QUERY:
                handleQueryEvent((QueryEventData) event.getData());
                break;
            case TABLE_MAP:
                handleTableMapEvent((TableMapEventData) event.getData());
                break;
            // 其他事件处理...
        }
    }

    private static void handleWriteRowsEvent(WriteRowsEventData eventData) {
        List<Serializable[]> rows = eventData.getRows();

        // 获取表名
        String tableName = getTableName(eventData);

        // 处理每一行数据
        for (Serializable[] row : rows) {
            // 根据需要调整以下代码以获取具体的列值
            String column1Value = row[0].toString();
            String column2Value = row[1].toString();

            // 将数据备份到另一个数据库
            backupToAnotherDatabase(tableName, column1Value, column2Value);
        }
    }

    private static void handleQueryEvent(QueryEventData eventData) {
        String sql = eventData.getSql();
        logger.info("@@ handleQueryEvent函数执行Query event SQL: {}", sql);

        // 解析SQL语句，根据需要处理
        // 例如，检查是否包含写入操作，然后执行相应的逻辑
    }

    private static void handleTableMapEvent(TableMapEventData eventData) {
        // 获取表映射信息，根据需要处理
        logger.info("@@ handleTableMapEvent函数执行TableMap event: {}", eventData);
    }

    private static String getTableName(EventData eventData) {
        // 获取表名的逻辑，可以使用TableMapEventData等信息
        // 根据实际情况实现
        return "example_table";
    }

    private static void backupToAnotherDatabase(String tableName, String column1Value, String column2Value) {
        // 将数据备份到另一个数据库的逻辑
        logger.info("Backup to another database: Table={}, Column1={}, Column2={}", tableName, column1Value, column2Value);
    }
}

点赞
收藏
关注作者

0/1000

抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认取消

加入云驻计划，成为创作者

华为云周边好礼
免费体验产品
特殊身份标识
线下官方门票
内部专家零距离
与10000+优质创作者共同成长

立即加入

MySQL：基于Spring监听Binlog日志

binlog的三种模式

以Mixed 为例

查看binlog是否开启

启动springboot程序

新建数据库

新建表数据

插入表数据

修改表数据

删除表数据

源码示例

pom.xml

Java示例

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

MySQL：基于Spring监听Binlog日志

binlog的三种模式

以Mixed 为例

查看binlog是否开启

启动springboot程序

新建数据库

新建表数据

插入表数据

修改表数据

删除表数据

源码示例

pom.xml

Java示例

全部回复

设置昵称

关于作者

目录

加入云驻计划，成为创作者

推荐阅读

相关产品