Java代码中，如何监控Mysql的binlog？

小伙伴们对数据库编程感兴趣吗？是否正在学习相关知识点？如果是，那么本文《Java代码中，如何监控Mysql的binlog？》，就很适合你，本篇文章讲解的知识点主要包括MySQL、Java。在之后的文章中也会多多分享相关知识点，希望对大家的知识积累有所帮助！

最近在工作中，遇到了这样一个业务场景，我们需要关注一个业务系统数据库中某几张表的数据，当数据发生新增或修改时，将它同步到另一个业务系统数据库中的表中。

一提到数据库的同步，估计大家第一时间想到的就是基于

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

在正式开始前，还是先简单介绍一下

show variables like 'log_bin';

如果为值为

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

对参数做一个简要说明：

• 在配置文件中加入了

  com.github.shyikomysql-binlog-connector-java0.21.0

  写一段简单的示例，看看它的具体使用方式：

  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();
      }
  }

  首先，创建一个

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

  连接mysql的

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

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

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

  我们监听到的事件类型数据有两类，第一类是

  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]}
  ]}

  在执行

  delete from dept where tenant_id=3

  控制台打印如下，

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

  简单的使用原理介绍完成后，再回到我们原先的需求上，需要将一张表中新增或修改的数据同步到另一张表中，问题还有一个，就是如何将返回的数据对应到所在的列上。这时应该怎么实现呢？以

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

  在将

  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;
  }

  在

  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}

  可以看到，将修改后的这一条记录中的属性填装到了它对应的列中，之后我们再根据具体的业务逻辑中，就可以根据字段名取出数据，将数据同步到其他的表了。

  ---

  如果文章对您有所帮助，欢迎关注公众号

  码农参上

  

  以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于数据库的相关知识，也可关注golang学习网公众号。