乐趣区

关于java:Java代码中如何监控Mysql的binlog

最近在工作中,遇到了这样一个业务场景,咱们须要关注一个业务零碎数据库中某几张表的数据,当数据产生新增或批改时,将它同步到另一个业务零碎数据库中的表中。

一提到数据库的同步,预计大家第一工夫想到的就是基于 binlog 的主从复制了,然而放在咱们的场景中,还有几个问题:

  • 第一,并不是须要复制所有表的数据,复制对象只有大量的几张表
  • 第二,也是比拟麻烦的,两个业务零碎数据库表构造可能不统一。例如,要同步数据库 1 的 A 表中的某些字段到数据库 2 的 B 表中,在这一过程中,A 表和 B 表的字段并不是完全相同

这样的话,咱们只能通过代码的形式,首先获取到数据库 1 表中数据的变动,再通过手动映射的形式,插入到数据库 2 的表中。然而,获取变动数据的这一过程,还是离不开 binlog,因而咱们就须要在代码中对binlog 进行一下监控。

先说论断,咱们最终应用了一个开源工具 mysql-binlog-connector-java,用来监控binlog 变动并获取数据,获取数据后再手动插入到另一个库的表中,基于它来实现了数据表的同步。我的项目的 git 地址如下:

https://github.com/shyiko/mysql-binlog-connector-java

在正式开始前,还是先简略介绍一下 mysqlbinlogbinlog是一个二进制文件,它保留在磁盘中,是用来记录数据库表构造变更、表数据批改的二进制日志。其实除了数据复制外,它还能够实现数据恢复、增量备份等性能。

启动项目前,首先须要确保 mysql 服务曾经启用了binlog

show variables like 'log_bin';

如果为值为OFF,示意没有启用,那么须要首先启用binlog,批改配置文件:

log_bin=mysql-bin
binlog-format=ROW
server-id=1

对参数做一个简要阐明:

  • 在配置文件中退出了 log_bin 配置项后,示意启用了binlog
  • binlog-formatbinlog 的日志格局,反对三种类型,别离是 STATEMENTROWMIXED,咱们在这里应用ROW 模式
  • server-id用于标识一个 sql 语句是从哪一个 server 写入的,这里肯定要进行设置,否则咱们在前面的代码中会无奈失常监听到事件

在更改完配置文件后,重启 mysql 服务。再次查看是否启用binlog,返回为ON,示意曾经开启胜利。

在 Java 我的项目中,首先引入 maven 坐标:

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

写一段简略的示例,看看它的具体应用形式:

public static void main(String[] args) {BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306, "hydra", "123456");
    client.setServerId(2);

    client.registerEventListener(event -> {EventData data = event.getData();
        if (data instanceof TableMapEventData) {System.out.println("Table:");
            TableMapEventData tableMapEventData = (TableMapEventData) data;
            System.out.println(tableMapEventData.getTableId()+": ["+tableMapEventData.getDatabase() + "-" + tableMapEventData.getTable()+"]");
        }
        if (data instanceof UpdateRowsEventData) {System.out.println("Update:");
            System.out.println(data.toString());
        } else if (data instanceof WriteRowsEventData) {System.out.println("Insert:");
            System.out.println(data.toString());
        } else if (data instanceof DeleteRowsEventData) {System.out.println("Delete:");
            System.out.println(data.toString());
        }
    });

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

首先,创立一个 BinaryLogClient 客户端对象,初始化时须要传入 mysql 的连贯信息,创立实现后,给客户端注册一个监听器,来实现它对 binlog 的监听和解析。在监听器中,咱们临时只对 4 种类型的事件数据进行了解决,除了 WriteRowsEventDataDeleteRowsEventDataUpdateRowsEventData 对应增删改操作类型的事件数据外,还有一个 TableMapEventData 类型的数据,蕴含了表的对应关系,在前面的例子中再具体阐明。

在这里,客户端监听到的是数据库级别的所有事件,并且能够监听到表的 DML 语句和 DDL 语句,所以咱们只须要解决咱们关怀的事件数据就行,否则会收到大量的冗余数据。

启动程序,控制台输入:

com.github.shyiko.mysql.binlog.BinaryLogClient openChannelToBinaryLogStream
信息: Connected to 127.0.0.1:3306 at mysql-bin.000002/1046 (sid:2, cid:10)

连贯 mysql 的 binlog 胜利,接下来,咱们在数据库中插入一条数据,这里操作的数据库名字是tenant,表是dept

insert into dept VALUES(8,"人力","","1");

这时,控制台就会打印监听到事件的数据:

Table:
108: [tenant-dept]
Insert:
WriteRowsEventData{tableId=108, includedColumns={0, 1, 2, 3}, rows=[[8, 人力, , 1]
]}

咱们监听到的事件类型数据有两类,第一类是 TableMapEventData,通过它能够获取操作的数据库名称、表名称以及表的id。之所以咱们要监听这个事件,是因为之后监听的实际操作中返回数据中蕴含了表的id,而没有表名等信息,所以如果咱们想晓得具体的操作是在哪一张表的话,就要先保护一个id 与表的对应关系。

第二个打印进去的监听事件数据是 WriteRowsEventData,其中记录了insert 语句作用的表,插入波及到的列,以及理论插入的数据。另外,如果咱们只须要对特定的一张或几张表进行解决的话,也能够提前设置表的名单,在这里依据表 id 到表名的映射关系,实现数据的过滤,

接下来,咱们再执行一条 update 语句:

update dept set tenant_id=3 where id=8 or id=9

控制台输入:

Table:
108: [tenant-dept]
Update:
UpdateRowsEventData{tableId=108, includedColumnsBeforeUpdate={0, 1, 2, 3}, includedColumns={0, 1, 2, 3}, rows=[{before=[8, 人力, , 1], after=[8, 人力, , 3]},
    {before=[9, 人力, , 1], after=[9, 人力, , 3]}
]}

在执行 update 语句时,可能会作用于多条数据,因而在理论批改的数据中,可能蕴含多行记录,这一点体现在下面的 rows 中,蕴含了 id 为 8 和 9 的两条数据。

最初,再执行一条 delete 语句:

delete from dept where tenant_id=3

控制台打印如下,rows中同样返回了失效的两条数据:

Table:
108: [tenant-dept]
Delete:
DeleteRowsEventData{tableId=108, includedColumns={0, 1, 2, 3}, rows=[[8, 人力, , 3],
    [9, 人力, , 3]
]}

简略的应用原理介绍实现后,再回到咱们原先的需要上,须要将一张表中新增或批改的数据同步到另一张表中,问题还有一个,就是如何将返回的数据对应到所在的列上。这时应该怎么实现呢?以 update 操作为例,咱们要对提取的数据后进行一下解决,更改下面例子中的办法:

if (data instanceof UpdateRowsEventData) {System.out.println("Update:");
    UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
    for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {List<Serializable> entries = Arrays.asList(row.getValue());
        System.out.println(entries);
        JSONObject dataObject = getDataObject(entries);
        System.out.println(dataObject);
    }
}

在将 data 类型强制转换为 UpdateRowsEventData 后,能够应用 getRows 办法获取到更新的行数据,并且可能取到每一列的值。

之后,调用了一个本人实现的 getDataObject 办法,用它来实现数据到列的绑定过程:

private static JSONObject getDataObject(List message) {JSONObject resultObject = new JSONObject();
    String format = "{\"id\":\"0\",\"dept_name\":\"1\",\"comment\":\"2\",\"tenant_id\":\"3\"}";
    JSONObject json = JSON.parseObject(format);
    for (String key : json.keySet()) {resultObject.put(key, message.get(json.getInteger(key)));
    }
    return resultObject;
}

format 字符串中,提前保护了一个数据库表的字段程序的字符串,标识了每个字段位于程序中的第几个地位。通过下面这个函数,可能实现数据到列的填装过程,咱们再执行一条 update 语句来查看一下后果:

update dept set tenant_id=3,comment="1" where id=8

控制台打印后果如下:

Table:
108: [tenant-dept]
Update:
[8, 人力, 1, 3]
{"tenant_id":3,"dept_name":"人力","comment":"1","id":8}

能够看到,将批改后的这一条记录中的属性填装到了它对应的列中,之后咱们再依据具体的业务逻辑中,就能够依据字段名取出数据,将数据同步到其余的表了。


如果文章对您有所帮忙,欢送关注公众号 码农参上

退出移动版