我要评论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 databasetest2023character 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 tablet_user(idbigint(20) not null auto_increment,usernamevarchar(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 intotest2023.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 查询为updatetest2023.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);
}
}
以上就是基于spring监听binlog日志的方法详解的详细内容，更多关于spring监听binlog日志的资料请关注代码网其它相关文章！
发表评论