canal集成

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/common/AbstractFileWork.java

@@ -22,6 +22,8 @@ public abstract class AbstractFileWork {
 
     protected FileDecodeConfig fileConfig;
 
+    protected FieldMappingVo fieldMapping;
+
     protected DbEngine getDbEngine() {
         return DbEngine.getInstance();
     }
@@ -29,6 +31,7 @@ public abstract class AbstractFileWork {
 
     public AbstractFileWork() {
         fileConfig = SpringUtil.getBean(FileDecodeConfig.class);
+        fieldMapping = FileDecodeFactory.getInstance().getFieldMapping();
     }
 
     /**

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/common/FieldMappingVo.java

@@ -0,0 +1,21 @@
+package cc.smtweb.system.canal.file.common;
+
+import lombok.Data;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @Author yaoq
+ * @Date 2022年07月20日 10:38
+ * @Description
+ */
+@Data
+public class FieldMappingVo {
+
+    private String type;
+
+    private List<Map<String, String>> fields;
+}
+
+

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/common/FileDecodeFactory.java

@@ -1,13 +1,18 @@
 package cc.smtweb.system.canal.file.common;
 
 import cc.smtweb.framework.core.common.SwEnum;
+import cc.smtweb.framework.core.db.DbEngine;
 import cc.smtweb.framework.core.exception.SwException;
+import cc.smtweb.framework.core.util.CommUtil;
+import cc.smtweb.framework.core.util.FileUtil;
+import cc.smtweb.framework.core.util.JsonUtil;
 import cc.smtweb.system.canal.file.impl.DmSqlFileWork;
 import cc.smtweb.system.canal.file.impl.KbSqlFileWork;
 import cc.smtweb.system.canal.file.impl.MySqlFileWork;
 import cc.smtweb.system.canal.file.impl.OracleFileWork;
 
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -29,6 +34,7 @@ public class FileDecodeFactory {
     }
 
     private Map<String, Class<? extends AbstractFileWork>> mapWorks;
+    private Map<String, FieldMappingVo> mapMapping;
 
 
     private FileDecodeFactory() {
@@ -37,6 +43,9 @@ public class FileDecodeFactory {
         mapWorks.put(SwEnum.DbType.KBSQL.value.toLowerCase(), KbSqlFileWork.class);
         mapWorks.put(SwEnum.DbType.MYSQL.value.toLowerCase(), MySqlFileWork.class);
         mapWorks.put(SwEnum.DbType.ORACLE.value.toLowerCase(), OracleFileWork.class);
+
+        mapMapping = new HashMap<>();
+        buildFieldMapping();
     }
 
     public AbstractFileWork getFileWork(String key) throws Exception {
@@ -44,4 +53,17 @@ public class FileDecodeFactory {
         if (cls == null) throw new SwException("暂不支持[" + key + "]类型数据库文件解析！");
         return cls.newInstance();
     }
+
+    private void buildFieldMapping() {
+        String path = Thread.currentThread().getContextClassLoader().getResource("").getPath() + "config/field-mapping.json";
+        List<FieldMappingVo> list = JsonUtil.parseList(FileUtil.readFileStr(path), FieldMappingVo.class);
+        if (CommUtil.isEmpty(list)) return;
+        for (FieldMappingVo bean : list) {
+            mapMapping.put(bean.getType(), bean);
+        }
+    }
+
+    public FieldMappingVo getFieldMapping() {
+        return mapMapping.get(DbEngine.getInstance().getDbType());
+    }
 }

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/common/FileDecodeService.java

@@ -17,7 +17,7 @@ public class FileDecodeService extends BaseSysService {
 
     @Override
     public int getInterval() {
-        return 10;
+        return 5;
     }
 
     @Override

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/impl/DmSqlFileWork.java

@@ -2,6 +2,7 @@ package cc.smtweb.system.canal.file.impl;
 
 import cc.smtweb.framework.core.common.CanalVO;
 import cc.smtweb.framework.core.common.SwEnum;
+import cc.smtweb.framework.core.exception.SwException;
 import cc.smtweb.system.canal.file.common.AbstractFileWork;
 
 /**
@@ -19,11 +20,11 @@ public class DmSqlFileWork extends AbstractFileWork {
 
     @Override
     protected void doCreateSql(CanalVO canalVO) {
-
+        throw new SwException("暂不支持新增表");
     }
 
     @Override
     protected void doAlterSql(CanalVO canalVO) {
-
+        throw new SwException("暂不支持修改表");
     }
 }

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/impl/KbSqlFileWork.java

@@ -42,13 +42,58 @@ public class KbSqlFileWork extends AbstractFileWork {
 
     @Override
     protected void doAlterSql(CanalVO canalVO) {
+        String sql = canalVO.getSql().replaceAll("`", "").replace("\r\n", "");
+        StringBuilder sb = new StringBuilder();
+        //添加sql
+        if (sql.contains("ADD")) {
+            sb.append("ALTER TABLE public.").append(canalVO.getTableName());
+            sql = sql.substring(sql.indexOf("ADD COLUMN"));
+            String[] sqls = sql.split(",");
+            for (String str : sqls) {
+                String[] ss = str.split(" ");
+                sb.append(" ADD COLUMN ").append(ss[2]).append(" ").append(convertFieldType(ss[3])).append(" NULL ").append(",");
+            }
+            sb.deleteCharAt(sb.length() - 1);
+        } else if (sql.contains("DROP")) {
+            sb.append("ALTER TABLE public.").append(canalVO.getTableName());
+            sql = sql.substring(sql.indexOf("DROP COLUMN"));
+            String[] sqls = sql.split(",");
+            for (String str : sqls) {
+                String[] ss = str.split(" ");
+                sb.append(" DROP COLUMN ").append(ss[2]).append(",");
+            }
+            sb.deleteCharAt(sb.length() - 1);
 
+        } else if (sql.contains("CHANGE")) {
+            sql = sql.substring(sql.indexOf("CHANGE COLUMN"));
+            String[] sqls = sql.split(",");
+            for (String str : sqls) {
+                String[] ss = str.split(" ");
+                sb.append("ALTER TABLE public.").append(canalVO.getTableName()).append(" RENAME COLUMN ").append(ss[2]).append(" TO ").append(ss[3]).append(";");
+            }
+            sb.deleteCharAt(sb.length() - 1);
+        } else if (sql.contains("MODIFY")) {
+            sb.append("ALTER TABLE public.").append(canalVO.getTableName());
+            sql = sql.substring(sql.indexOf("MODIFY COLUMN"));
+            String[] sqls = sql.split(",");
+            for (String str : sqls) {
+                String[] ss = str.split(" ");
+                sb.append(" ALTER COLUMN ").append(ss[2]).append(" TYPE ").append(convertFieldType(ss[3])).append(",");
+            }
+            sb.deleteCharAt(sb.length() - 1);
+        }
+        log.debug("sql:::" + sb.toString());
+        getDbEngine().update(sb.toString());
     }
 
 
     private String convertFieldType(String type) {
-        if (type.contains("bigint")) return "bigint";
-        if (type.contains("tinyint")) return "tinyint";
+        if (fieldMapping == null) return type;
+        String[] types = type.replace("(", "-").replace(")", "").split("-");
+        for (Map<String, String> field : fieldMapping.getFields()) {
+            String key = types[0].toUpperCase();
+            if (field.containsKey(key)) return field.get(key);
+        }
         return type;
     }
 }

smtweb-framework/canal/file/src/main/java/cc/smtweb/system/canal/file/impl/OracleFileWork.java

@@ -2,6 +2,7 @@ package cc.smtweb.system.canal.file.impl;
 
 import cc.smtweb.framework.core.common.CanalVO;
 import cc.smtweb.framework.core.common.SwEnum;
+import cc.smtweb.framework.core.exception.SwException;
 import cc.smtweb.system.canal.file.common.AbstractFileWork;
 
 /**
@@ -19,11 +20,11 @@ public class OracleFileWork extends AbstractFileWork {
 
     @Override
     protected void doCreateSql(CanalVO canalVO) {
-
+        throw new SwException("暂不支持新增表");
     }
 
     @Override
     protected void doAlterSql(CanalVO canalVO) {
-
+        throw new SwException("暂不支持修改表");
     }
 }

smtweb-framework/canal/file/src/main/resources/config/field-mapping.json

@@ -0,0 +1 @@
+[{"type":"kbsql","fields":[{"INT":"INT"},{"BIGINT":"BIGINT"},{"INTEGER":"INTEGER"},{"NUMERIC":"NUMERIC"},{"TINYINT":"TINYINT"},{"SMALLINT":"SMALLINT"},{"CHAR":"CHAR"},{"TEXT":"TEXT"},{"BLOB":"BLOB"},{"CLOB":"CLOB"},{"VARCHAR":"VARCHAR2"}]},{"type":"dmsql","fields":[{"INT":"INT"},{"BIGINT":"BIGINT"},{"INTEGER":"INTEGER"},{"NUMERIC":"NUMERIC"},{"TINYINT":"TINYINT"},{"SMALLINT":"SMALLINT"},{"CHAR":"CHAR"},{"TEXT":"TEXT"},{"BLOB":"BLOB"},{"CLOB":"CLOB"},{"VARCHAR":"VARCHAR2"}]},{"type":"oracle","fields":[{"INT":"INT"},{"BIGINT":"BIGINT"},{"INTEGER":"INTEGER"},{"NUMERIC":"NUMERIC"},{"TINYINT":"TINYINT"},{"SMALLINT":"SMALLINT"},{"CHAR":"CHAR"},{"TEXT":"VARCHAR2"},{"BLOB":"BLOB"},{"CLOB":"CLOB"},{"VARCHAR":"VARCHAR2"}]}]