最近在工作中，遇到了这样一个业务场景，咱们须要关注一个业务零碎数据库中某几张表的数据，当数据产生新增或批改时，将它同步到另一个业务零碎数据库中的表中。
一提到数据库的同步，预计大家第一工夫想到的就是基于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在正式开始前，还是先简略介绍一下mysql的binlog，binlog是一个二进制文件，它保留在磁盘中，是用来记录数据库表构造变更、表数据批改的二进制日志。其实除了数据复制外，它还能够实现数据恢复、增量备份等性能。
启动项目前，首先须要确保mysql服务曾经启用了binlog：
show variables like 'log_bin';如果为值为OFF，示意没有启用，那么须要首先启用binlog，批改配置文件：
log_bin=mysql-binbinlog-format=ROWserver-id=1对参数做一个简要阐明：
- 在配置文件中退出了
log_bin配置项后，示意启用了binlog binlog-format是binlog的日志格局，反对三种类型，别离是STATEMENT、ROW、MIXED，咱们在这里应用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种类型的事件数据进行了解决，除了WriteRowsEventData、DeleteRowsEventData、UpdateRowsEventData对应增删改操作类型的事件数据外，还有一个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}能够看到，将批改后的这一条记录中的属性填装到了它对应的列中，之后咱们再依据具体的业务逻辑中，就能够依据字段名取出数据，将数据同步到其余的表了。
如果文章对您有所帮忙，欢送关注公众号 码农参上