binlog 数据同步完善（新增，修改，删除同步）

zhongrj

2024-02-23 fc47a705ab33a976f7a116c0ce6e236bb5057fa5

 src/main/java/org/springblade/binlog/listener/TourBinLogListener.java

@@ -1,8 +1,8 @@
package org.springblade.binlog.listener;


import cn.hutool.core.collection.CollectionUtil;
import lombok.extern.slf4j.Slf4j;
import org.springblade.binlog.client.MysqlClient;
import org.springblade.binlog.config.DataSourceConfig;
import org.springblade.binlog.constant.BinLogConstants;
import org.springblade.binlog.util.BinLogUtils;
@@ -16,6 +16,7 @@
import javax.annotation.Resource;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

@@ -90,28 +91,30 @@
   private void saveHandle(BinLogItem item) {
      String tableName = item.getDbTable().split("-")[1];
      Map<String, Serializable> data = item.getAfter();
      log.info(tableName);
      Map<String, DataProperty> dataProperty = item.getDataProperty();

      // 创建更新对象
      List<String> keyList = new ArrayList<>();
      List<String> valueList = new ArrayList<>();
      // 遍历匹配数据
      dataProperty.forEach((key,value)->{
         log.info(key);
         log.info(value.dataType);
//         log.info("数据类型 " + value.dataType);
         if(null!=data.get(key)) {
            log.info(data.get(key).toString());
            valueList.add(data.get(key).toString());
         }else {
            if (value.dataType.equals("int") ||
               value.dataType.equals("bigint") ||
               value.dataType.equals("datetime") ||
               value.dataType.equals("date")){
               valueList.add(null);
            }else{
               valueList.add("");
            keyList.add(key);
            if (value.dataType.equals("varchar") ||
               value.getDataType().equals("char")) {
               valueList.add("'" + data.get(key).toString() + "'");
            }else if(value.dataType.equals("text") ||
               value.dataType.equals("mediumtext")) {
               byte[] bytes = (byte[])data.get(key);
               valueList.add("'" + new String(bytes) + "'");
            }else if(value.dataType.equals("date") ||
               value.getDataType().equals("datetime")) {
               Long time = Long.parseLong(data.get(key).toString());
               valueList.add("'" + BinLogUtils.getDateFormatStr(new Date(time)) + "'");
            }else {
               valueList.add(data.get(key).toString());
            }
         }
         keyList.add(key);
      });
      // 拼接sql
      StringBuilder sqlBuilder = new StringBuilder();
@@ -124,6 +127,8 @@
         .append(String.join(",",valueList))
         .append(");");
      log.info("sql： " + sqlBuilder.toString());
      // 同步到其他数据库
      MysqlClient.insert(sqlBuilder.toString());
   }

   /**
@@ -131,6 +136,43 @@
    * @param item
    */
   private void updateHandle(BinLogItem item) {
      String tableName = item.getDbTable().split("-")[1];
      Map<String, Serializable> data = item.getAfter();
      Map<String, DataProperty> dataProperty = item.getDataProperty();
      // 创建更新对象
      List<String> updateList = new ArrayList<>();
      // 遍历匹配数据
      dataProperty.forEach((key,value)->{
//         log.info("数据类型 " + value.dataType);
         if(null!=data.get(key) && !key.equals("id")) {
            if (value.dataType.equals("varchar") ||
               value.getDataType().equals("char")) {
               updateList.add(key + " = '" + data.get(key).toString() + "'");
            }else if(value.dataType.equals("text") ||
               value.dataType.equals("mediumtext")) {
               byte[] bytes = (byte[])data.get(key);
               updateList.add(key + " = '" + new String(bytes) + "'");
            }else if(value.dataType.equals("date") ||
               value.getDataType().equals("datetime")) {
               Long time = Long.parseLong(data.get(key).toString());
               updateList.add(key + " = '" + BinLogUtils.getDateFormatStr(new Date(time)) + "'");
            }else {
               updateList.add(key + " = " + data.get(key).toString());
            }
         }
      });
      // 拼接sql
      StringBuilder sqlBuilder = new StringBuilder();
      sqlBuilder.append("update ")
         .append(tableName)
         .append(" set ")
         .append(String.join(",",updateList))
         .append(" where id = ")
         .append(data.get("id"))
         .append(";");
      log.info("sql： " + sqlBuilder.toString());
      // 同步到其他数据库
      MysqlClient.update(sqlBuilder.toString());
   }

   /**
@@ -138,6 +180,18 @@
    * @param item 数据
    */
   private void deletedHandle(BinLogItem item) {
      String tableName = item.getDbTable().split("-")[1];
      Map<String, Serializable> data = item.getBefore();
      // 拼接sql
      StringBuilder sqlBuilder = new StringBuilder();
      sqlBuilder.append("delete from ")
         .append(tableName)
         .append(" where id = ")
         .append(data.get("id"))
         .append(";");
      log.info("sql： " + sqlBuilder.toString());
      // 同步到其他数据库
      MysqlClient.delete(sqlBuilder.toString());
   }
}