!147 merge

Merge pull request !147 from AE86/V_1.0.0_RC

dbsyncer-biz/src/main/java/org/dbsyncer/biz/MappingService.java

@@ -19,6 +19,13 @@ public interface MappingService {
      */
     String add(Map<String, String> params);
 
+    /**
+     * 复制驱动
+     *
+     * @param id
+     */
+    String copy(String id);
+
     /**
      * 修改驱动
      *

dbsyncer-biz/src/main/java/org/dbsyncer/biz/MonitorService.java

@@ -1,7 +1,6 @@
 package org.dbsyncer.biz;
 
 import org.dbsyncer.biz.vo.AppReportMetricVo;
-import org.dbsyncer.biz.vo.MessageVo;
 import org.dbsyncer.biz.vo.MetaVo;
 import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.monitor.enums.MetricEnum;
@@ -71,6 +70,11 @@ public interface MonitorService {
      */
     String clearLog();
 
+    /**
+     * 删除过期的数据和日志
+     */
+    void deleteExpiredDataAndLog();
+
     /**
      * 获取所有同步数据状态类型
      *

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/mapping/MappingChecker.java

@@ -113,6 +113,17 @@ public class MappingChecker extends AbstractChecker {
         return mapping;
     }
 
+    public void addMeta(Mapping mapping) {
+        Meta meta = new Meta();
+        meta.setMappingId(mapping.getId());
+
+        // 修改基本配置
+        this.modifyConfigModel(meta, new HashMap<>());
+
+        String id = manager.addConfigModel(meta);
+        mapping.setMetaId(id);
+    }
+
     /**
      * 更新元信息
      *
@@ -199,17 +210,6 @@ public class MappingChecker extends AbstractChecker {
         manager.editConfigModel(meta);
     }
 
-    private void addMeta(Mapping mapping) {
-        Meta meta = new Meta();
-        meta.setMappingId(mapping.getId());
-
-        // 修改基本配置
-        this.modifyConfigModel(meta, new HashMap<>());
-
-        String id = manager.addConfigModel(meta);
-        mapping.setMetaId(id);
-    }
-
     private void getMetaTotal(Meta meta, String model) {
         // 全量同步
         if (ModelEnum.isFull(model)) {

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/system/SystemConfigChecker.java

@@ -53,12 +53,13 @@ public class SystemConfigChecker extends AbstractChecker {
         SystemConfig systemConfig = manager.getSystemConfig();
         Assert.notNull(systemConfig, "配置文件为空.");
 
+        // 同步数据过期时间（天）
+        systemConfig.setExpireDataDays(NumberUtil.toInt(params.get("expireDataDays"), systemConfig.getExpireDataDays()));
+        // 系统日志过期时间（天）
+        systemConfig.setExpireLogDays(NumberUtil.toInt(params.get("expireLogDays"), systemConfig.getExpireLogDays()));
         // 刷新监控间隔（秒）
-        String refreshInterval = params.get("refreshInterval");
-        if (StringUtil.isNotBlank(refreshInterval)) {
-            int time = NumberUtil.toInt(refreshInterval, 10);
-            systemConfig.setRefreshInterval(time);
-        }
+        systemConfig.setRefreshIntervalSeconds(NumberUtil.toInt(params.get("refreshIntervalSeconds"), systemConfig.getRefreshIntervalSeconds()));
+
         logService.log(LogType.SystemLog.INFO, "修改系统配置");
 
         // 修改基本配置

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/MappingServiceImpl.java

@@ -8,17 +8,24 @@ import org.dbsyncer.biz.checker.impl.mapping.MappingChecker;
 import org.dbsyncer.biz.vo.ConnectorVo;
 import org.dbsyncer.biz.vo.MappingVo;
 import org.dbsyncer.biz.vo.MetaVo;
+import org.dbsyncer.common.snowflake.SnowflakeIdWorker;
 import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.parser.enums.ModelEnum;
 import org.dbsyncer.parser.logger.LogType;
-import org.dbsyncer.parser.model.*;
+import org.dbsyncer.parser.model.ConfigModel;
+import org.dbsyncer.parser.model.Connector;
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.Meta;
+import org.dbsyncer.parser.model.TableGroup;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 import org.springframework.util.Assert;
 
+import java.time.Instant;
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
@@ -42,6 +49,9 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
     @Autowired
     private ConnectorService connectorService;
 
+    @Autowired
+    private SnowflakeIdWorker snowflakeIdWorker;
+
     @Override
     public String add(Map<String, String> params) {
         ConfigModel model = mappingChecker.checkAddConfigModel(params);
@@ -58,6 +68,36 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
         return id;
     }
 
+    @Override
+    public String copy(String id) {
+        Mapping mapping = manager.getMapping(id);
+        Assert.notNull(mapping, "The mapping id is invalid.");
+
+        String json = JsonUtil.objToJson(mapping);
+        Mapping newMapping = JsonUtil.jsonToObj(json, Mapping.class);
+        newMapping.setName(mapping.getName() + "(复制)");
+        newMapping.setId(String.valueOf(snowflakeIdWorker.nextId()));
+        newMapping.setUpdateTime(Instant.now().toEpochMilli());
+        mappingChecker.addMeta(newMapping);
+
+        manager.addConfigModel(newMapping);
+        log(LogType.MappingLog.COPY, newMapping);
+
+        // 复制映射表关系
+        List<TableGroup> groupList = manager.getTableGroupAll(mapping.getId());
+        if(!CollectionUtils.isEmpty(groupList)){
+            groupList.forEach(tableGroup -> {
+                String tableGroupJson = JsonUtil.objToJson(tableGroup);
+                TableGroup newTableGroup = JsonUtil.jsonToObj(tableGroupJson, TableGroup.class);
+                newTableGroup.setId(String.valueOf(snowflakeIdWorker.nextId()));
+                newTableGroup.setMappingId(newMapping.getId());
+                manager.addTableGroup(newTableGroup);
+                log(LogType.TableGroupLog.COPY, newTableGroup);
+            });
+        }
+        return String.format("复制成功[%s]", newMapping.getName());
+    }
+
     @Override
     public String edit(Map<String, String> params) {
         String id = params.get(ConfigConstant.CONFIG_MODEL_ID);
@@ -91,7 +131,7 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
             // 删除tableGroup
             List<TableGroup> groupList = manager.getTableGroupAll(id);
             if (!CollectionUtils.isEmpty(groupList)) {
-                groupList.forEach(t -> manager.removeConfigModel(t.getId()));
+                groupList.forEach(t -> manager.removeTableGroup(t.getId()));
             }
 
             // 删除驱动

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/MonitorServiceImpl.java

@@ -2,6 +2,7 @@ package org.dbsyncer.biz.impl;
 
 import org.dbsyncer.biz.DataSyncService;
 import org.dbsyncer.biz.MonitorService;
+import org.dbsyncer.biz.SystemConfigService;
 import org.dbsyncer.biz.metric.MetricDetailFormatter;
 import org.dbsyncer.biz.metric.impl.CpuMetricDetailFormatter;
 import org.dbsyncer.biz.metric.impl.DiskMetricDetailFormatter;
@@ -21,6 +22,7 @@ import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.monitor.Monitor;
 import org.dbsyncer.monitor.enums.DiskMetricEnum;
 import org.dbsyncer.monitor.enums.MetricEnum;
@@ -31,8 +33,13 @@ import org.dbsyncer.monitor.model.MetricResponse;
 import org.dbsyncer.parser.enums.ModelEnum;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.Meta;
+import org.dbsyncer.storage.StorageService;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.dbsyncer.storage.enums.StorageDataStatusEnum;
+import org.dbsyncer.storage.enums.StorageEnum;
+import org.dbsyncer.storage.query.BooleanFilter;
+import org.dbsyncer.storage.query.Query;
+import org.dbsyncer.storage.query.filter.LongFilter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeanUtils;
@@ -41,6 +48,8 @@ import org.springframework.util.Assert;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.Resource;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.LinkedHashMap;
@@ -67,6 +76,12 @@ public class MonitorServiceImpl extends BaseServiceImpl implements MonitorServic
     @Resource
     private ScheduledTaskService scheduledTaskService;
 
+    @Resource
+    private StorageService storageService;
+
+    @Resource
+    private SystemConfigService systemConfigService;
+
     private Map<String, MetricDetailFormatter> metricDetailFormatterMap = new LinkedHashMap<>();
 
     @PostConstruct
@@ -169,6 +184,12 @@ public class MonitorServiceImpl extends BaseServiceImpl implements MonitorServic
         return "清空日志成功";
     }
 
+    @Override
+    public void deleteExpiredDataAndLog() {
+        deleteExpiredData();
+        deleteExpiredLog();
+    }
+
     @Override
     public List<StorageDataStatusEnum> getStorageDataStatusEnumAll() {
         return monitor.getStorageDataStatusEnumAll();
@@ -219,6 +240,32 @@ public class MonitorServiceImpl extends BaseServiceImpl implements MonitorServic
         }
     }
 
+    private void deleteExpiredData() {
+        List<MetaVo> metaAll = getMetaAll();
+        if (!CollectionUtils.isEmpty(metaAll)) {
+            Query query = new Query();
+            query.setType(StorageEnum.DATA);
+            int expireDataDays = systemConfigService.getSystemConfigVo().getExpireDataDays();
+            long expiredTime = Timestamp.valueOf(LocalDateTime.now().minusDays(expireDataDays)).getTime();
+            LongFilter expiredFilter = new LongFilter(ConfigConstant.CONFIG_MODEL_CREATE_TIME, FilterEnum.LT, expiredTime);
+            query.setBooleanFilter(new BooleanFilter().add(expiredFilter));
+            metaAll.forEach(metaVo -> {
+                query.setMetaId(metaVo.getId());
+                storageService.delete(query);
+            });
+        }
+    }
+
+    private void deleteExpiredLog() {
+        Query query = new Query();
+        query.setType(StorageEnum.LOG);
+        int expireLogDays = systemConfigService.getSystemConfigVo().getExpireLogDays();
+        long expiredTime = Timestamp.valueOf(LocalDateTime.now().minusDays(expireLogDays)).getTime();
+        LongFilter expiredFilter = new LongFilter(ConfigConstant.CONFIG_MODEL_CREATE_TIME, FilterEnum.LT, expiredTime);
+        query.setBooleanFilter(new BooleanFilter().add(expiredFilter));
+        storageService.delete(query);
+    }
+
     private MetaVo convertMeta2Vo(Meta meta) {
         Mapping mapping = monitor.getMapping(meta.getMappingId());
         Assert.notNull(mapping, "驱动不存在.");

dbsyncer-biz/src/main/java/org/dbsyncer/biz/impl/TableGroupServiceImpl.java

@@ -96,7 +96,7 @@ public class TableGroupServiceImpl extends BaseServiceImpl implements TableGroup
         Stream.of(StringUtil.split(ids, ",")).parallel().forEach(id -> {
             TableGroup model = manager.getTableGroup(id);
             log(LogType.TableGroupLog.DELETE, model);
-            manager.removeConfigModel(id);
+            manager.removeTableGroup(id);
         });
 
         // 合并驱动公共字段

dbsyncer-connector/src/main/java/org/dbsyncer/connector/es/ESConnectorMapper.java

@@ -17,7 +17,7 @@ public final class ESConnectorMapper implements ConnectorMapper<ESConfig, EasyRe
         this.client = ESUtil.getConnection(config);
         try {
             MainResponse info = client.info(RequestOptions.DEFAULT);
-            client.setVersion(Version.fromString(info.getVersion().getNumber()));;
+            client.setVersion(Version.fromString(info.getVersion().getNumber()));
         } catch (Exception e) {
             throw new ConnectorException(String.format("获取ES版本信息异常 %s, %s", config.getUrl(), e.getMessage()));
         }
@@ -51,4 +51,4 @@ public final class ESConnectorMapper implements ConnectorMapper<ESConfig, EasyRe
     public Object clone() throws CloneNotSupportedException {
         return super.clone();
     }
-}
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/column/PgColumnValue.java

@@ -83,6 +83,8 @@ public final class PgColumnValue extends AbstractColumnValue<String> {
             return Timestamp.from(toInstantFromMicros(PGStatement.DATE_POSITIVE_INFINITY));
         } else if ("-infinity".equals(asString())) {
             return Timestamp.from(toInstantFromMicros(PGStatement.DATE_NEGATIVE_INFINITY));
+        } else if ("null".equals(asString()) || StringUtil.isBlank(asString())) {
+            return null;
         }
         return DateFormatUtil.timeWithoutTimeZoneToTimestamp(asString());
     }

dbsyncer-manager/src/main/java/org/dbsyncer/manager/Manager.java

@@ -89,6 +89,8 @@ public interface Manager extends Executor {
 
     String editTableGroup(TableGroup model);
 
+    void removeTableGroup(String id);
+
     TableGroup getTableGroup(String tableGroupId);
 
     List<TableGroup> getTableGroupAll(String mappingId);

dbsyncer-manager/src/main/java/org/dbsyncer/manager/ManagerFactory.java

@@ -173,6 +173,11 @@ public class ManagerFactory implements Manager, ApplicationListener<ClosedEvent>
         return operationTemplate.execute(new OperationConfig(model, CommandEnum.OPR_EDIT, GroupStrategyEnum.TABLE));
     }
 
+    @Override
+    public void removeTableGroup(String id) {
+        operationTemplate.remove(new OperationConfig(id, GroupStrategyEnum.TABLE));
+    }
+
     @Override
     public TableGroup getTableGroup(String tableGroupId) {
         return operationTemplate.queryObject(TableGroup.class, tableGroupId);

dbsyncer-parser/src/main/java/org/dbsyncer/parser/logger/LogType.java

@@ -115,7 +115,8 @@ public interface LogType {
         DELETE("32", "删除"),
         RUNNING("33", "启动"),
         STOP("34", "停止"),
-        CLEAR_DATA("35", "清空同步数据");
+        CLEAR_DATA("35", "清空同步数据"),
+        COPY("36", "复制");
 
         private String type;
         private String message;
@@ -149,7 +150,8 @@ public interface LogType {
         UPDATE("41", "修改"),
         DELETE("42", "删除"),
         INCREMENT_FAILED("43", "增量同步异常"),
-        FULL_FAILED("44", "全量同步异常");
+        FULL_FAILED("44", "全量同步异常"),
+        COPY("45", "复制");
 
         private String type;
         private String message;
@@ -180,8 +182,7 @@ public interface LogType {
      */
     enum MetaLog implements LogType {
         DELETE("50", "删除"),
-        CLEAR("51", "删除数据"),
-        TASK("52", "任务");
+        CLEAR("51", "删除数据");
 
         private String type;
         private String message;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/SystemConfig.java

@@ -15,14 +15,42 @@ public class SystemConfig extends ConfigModel {
         super.setType(ConfigConstant.SYSTEM);
     }
 
-    private int refreshInterval = 5;
+    /**
+     * 同步数据过期时间（天）
+     */
+    private int expireDataDays = 7;
 
-    public int getRefreshInterval() {
-        return refreshInterval;
+    /**
+     * 系统日志过期时间（天）
+     */
+    private int expireLogDays = 30;
+
+    /**
+     * 刷新页面间隔（秒）
+     */
+    private int refreshIntervalSeconds = 5;
+
+    public int getExpireDataDays() {
+        return expireDataDays;
     }
 
-    public void setRefreshInterval(int refreshInterval) {
-        this.refreshInterval = refreshInterval;
+    public void setExpireDataDays(int expireDataDays) {
+        this.expireDataDays = expireDataDays;
     }
 
+    public int getExpireLogDays() {
+        return expireLogDays;
+    }
+
+    public void setExpireLogDays(int expireLogDays) {
+        this.expireLogDays = expireLogDays;
+    }
+
+    public int getRefreshIntervalSeconds() {
+        return refreshIntervalSeconds;
+    }
+
+    public void setRefreshIntervalSeconds(int refreshIntervalSeconds) {
+        this.refreshIntervalSeconds = refreshIntervalSeconds;
+    }
 }

dbsyncer-parser/src/main/resources/Connector.json

@@ -1,128 +0,0 @@
-{
-  "config":{
-    "connectorType":"Mysql",
-    "password":"123",
-    "properties":{
-
-    },
-    "url":"jdbc:mysql://127.0.0.1:3305/test?rewriteBatchedStatements=true&useUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false&autoReconnect=true&failOverReadOnly=false",
-    "username":"root"
-  },
-  "createTime":1670169249093,
-  "id":"1049111399101501440",
-  "name":"Mysql",
-  "table":[
-    {
-      "count":0,
-      "name":"aqi_stations",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"dbs_test03",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"ds1",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"ds1_copy1",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"mdt_partnerrole",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_big_table",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_big_table_copy1",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_file",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_mark",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_org",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_user",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_visit",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"nc_customer",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"product",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"product_copy",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"shop_goods",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"shop_goods_class",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"shop_goods_spec",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"tb_jy_visit",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"tb_jy_visit_copy1",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"vote_records",
-      "type":"TABLE"
-    },
-    {
-      "count":0,
-      "name":"my_view",
-      "type":"VIEW"
-    }
-  ],
-  "type":"connector",
-  "updateTime":1672237644641
-}

dbsyncer-parser/src/main/resources/Mapping.json

@@ -1,360 +0,0 @@
-{
-  "batchNum":1000,
-  "convert":[
-
-  ],
-  "createTime":1671809807457,
-  "filter":[
-
-  ],
-  "id":"1055992399609860096",
-  "listener":{
-    "banDelete":false,
-    "banInsert":false,
-    "banUpdate":false,
-    "cron":"*/30 * * * * ?",
-    "delete":"D",
-    "eventFieldName":"",
-    "insert":"I",
-    "listenerType":"log",
-    "update":"U"
-  },
-  "metaId":"1055992399609860097",
-  "model":"increment",
-  "name":"同步测试",
-  "params":{
-
-  },
-  "readNum":10000,
-  "sourceColumn":[
-    {
-      "name":"id",
-      "pk":true,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_cust_id",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_type",
-      "pk":false,
-      "type":1,
-      "typeName":"CHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_user_id",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_day",
-      "pk":false,
-      "type":1,
-      "typeName":"CHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_lat",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_lng",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_distance",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_lat",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_lng",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_distance",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_status",
-      "pk":false,
-      "type":1,
-      "typeName":"CHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"pre_end_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"create_by",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"create_by_name",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"create_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"update_by",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"update_by_name",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"update_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"version",
-      "pk":false,
-      "type":4,
-      "typeName":"INT",
-      "unmodifiabled":false
-    },
-    {
-      "name":"is_del",
-      "pk":false,
-      "type":4,
-      "typeName":"INT",
-      "unmodifiabled":false
-    }
-  ],
-  "sourceConnectorId":"1049111399101501440",
-  "targetColumn":[
-    {
-      "name":"id",
-      "pk":true,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_cust_id",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_type",
-      "pk":false,
-      "type":1,
-      "typeName":"CHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_user_id",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_day",
-      "pk":false,
-      "type":1,
-      "typeName":"CHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_lat",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_lng",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"start_distance",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_lat",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_lng",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"end_distance",
-      "pk":false,
-      "type":8,
-      "typeName":"DOUBLE",
-      "unmodifiabled":false
-    },
-    {
-      "name":"visit_status",
-      "pk":false,
-      "type":1,
-      "typeName":"CHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"pre_end_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"create_by",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"create_by_name",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"create_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"update_by",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"update_by_name",
-      "pk":false,
-      "type":12,
-      "typeName":"VARCHAR",
-      "unmodifiabled":false
-    },
-    {
-      "name":"update_time",
-      "pk":false,
-      "type":93,
-      "typeName":"DATETIME",
-      "unmodifiabled":false
-    },
-    {
-      "name":"version",
-      "pk":false,
-      "type":4,
-      "typeName":"INT",
-      "unmodifiabled":false
-    },
-    {
-      "name":"is_del",
-      "pk":false,
-      "type":4,
-      "typeName":"INT",
-      "unmodifiabled":false
-    }
-  ],
-  "targetConnectorId":"1049111399101501440",
-  "threadNum":32,
-  "type":"mapping",
-  "updateTime":1673452598774
-}

dbsyncer-parser/src/main/resources/TableGroup.json

@@ -1,735 +0,0 @@
-{
-  "command":{
-    "QUERY_CURSOR":"SELECT `id`, `visit_cust_id`, `visit_type`, `visit_user_id`, `visit_day`, `start_time`, `start_lat`, `start_lng`, `start_distance`, `end_time`, `end_lat`, `end_lng`, `end_distance`, `visit_status`, `pre_end_time`, `create_by`, `create_by_name`, `create_time`, `update_by`, `update_by_name`, `update_time`, `version`, `is_del` FROM `tb_jy_visit` ORDER BY `id` LIMIT ?",
-    "DELETE":"DELETE FROM `tb_jy_visit_copy1` WHERE `id`=?",
-    "QUERY":"SELECT `id`, `visit_cust_id`, `visit_type`, `visit_user_id`, `visit_day`, `start_time`, `start_lat`, `start_lng`, `start_distance`, `end_time`, `end_lat`, `end_lng`, `end_distance`, `visit_status`, `pre_end_time`, `create_by`, `create_by_name`, `create_time`, `update_by`, `update_by_name`, `update_time`, `version`, `is_del` FROM `tb_jy_visit` WHERE `id` > ? ORDER BY `id` LIMIT ?",
-    "QUERY_COUNT":"SELECT COUNT(1) FROM (SELECT 1 FROM `tb_jy_visit` GROUP BY `id`) DBSYNCER_T",
-    "INSERT":"INSERT INTO `tb_jy_visit_copy1`(`id`, `visit_cust_id`, `visit_type`, `visit_user_id`, `visit_day`, `start_time`, `start_lat`, `start_lng`, `start_distance`, `end_time`, `end_lat`, `end_lng`, `end_distance`, `visit_status`, `pre_end_time`, `create_by`, `create_by_name`, `create_time`, `update_by`, `update_by_name`, `update_time`, `version`, `is_del`) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
-    "UPDATE":"UPDATE `tb_jy_visit_copy1` SET `visit_cust_id`=?,`visit_type`=?,`visit_user_id`=?,`visit_day`=?,`start_time`=?,`start_lat`=?,`start_lng`=?,`start_distance`=?,`end_time`=?,`end_lat`=?,`end_lng`=?,`end_distance`=?,`visit_status`=?,`pre_end_time`=?,`create_by`=?,`create_by_name`=?,`create_time`=?,`update_by`=?,`update_by_name`=?,`update_time`=?,`version`=?,`is_del`=? WHERE `id`=?",
-    "QUERY_COUNT_EXIST":"SELECT COUNT(1) FROM `tb_jy_visit_copy1` WHERE `id` = ?"
-  },
-  "convert":[
-
-  ],
-  "createTime":1672238741208,
-  "fieldMapping":[
-    {
-      "source":{
-        "name":"id",
-        "pk":true,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"id",
-        "pk":true,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"visit_cust_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"visit_cust_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"visit_type",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"visit_type",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"visit_user_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"visit_user_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"visit_day",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"visit_day",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"start_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"start_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"start_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"start_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"start_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"start_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"start_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"start_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"end_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"end_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"end_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"end_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"end_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"end_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"visit_status",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"visit_status",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"pre_end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"pre_end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"create_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"create_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"create_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"create_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"create_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"create_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"update_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"update_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"update_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"update_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"update_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"update_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"version",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"version",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      }
-    },
-    {
-      "source":{
-        "name":"is_del",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      },
-      "target":{
-        "name":"is_del",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      }
-    }
-  ],
-  "filter":[
-
-  ],
-  "id":"1057791478157414400",
-  "index":1,
-  "mappingId":"1055992399609860096",
-  "name":"tableGroup",
-  "params":{
-
-  },
-  "sourceTable":{
-    "column":[
-      {
-        "name":"id",
-        "pk":true,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_cust_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_type",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_user_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_day",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_status",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"pre_end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"create_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"create_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"create_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"update_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"update_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"update_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"version",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      },
-      {
-        "name":"is_del",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      }
-    ],
-    "count":0,
-    "name":"tb_jy_visit",
-    "primaryKey":"",
-    "type":"TABLE"
-  },
-  "targetTable":{
-    "column":[
-      {
-        "name":"id",
-        "pk":true,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_cust_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_type",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_user_id",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_day",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"start_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_lat",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_lng",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"end_distance",
-        "pk":false,
-        "type":8,
-        "typeName":"DOUBLE",
-        "unmodifiabled":false
-      },
-      {
-        "name":"visit_status",
-        "pk":false,
-        "type":1,
-        "typeName":"CHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"pre_end_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"create_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"create_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"create_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"update_by",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"update_by_name",
-        "pk":false,
-        "type":12,
-        "typeName":"VARCHAR",
-        "unmodifiabled":false
-      },
-      {
-        "name":"update_time",
-        "pk":false,
-        "type":93,
-        "typeName":"DATETIME",
-        "unmodifiabled":false
-      },
-      {
-        "name":"version",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      },
-      {
-        "name":"is_del",
-        "pk":false,
-        "type":4,
-        "typeName":"INT",
-        "unmodifiabled":false
-      }
-    ],
-    "count":0,
-    "name":"tb_jy_visit_copy1",
-    "primaryKey":"",
-    "type":"TABLE"
-  },
-  "type":"tableGroup",
-  "updateTime":1673452597645
-}

dbsyncer-parser/src/main/test/ConnectorParserTest.java

@@ -1,77 +0,0 @@
-import org.apache.commons.io.FileUtils;
-import org.dbsyncer.common.model.AbstractConnectorConfig;
-import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.connector.enums.ConnectorEnum;
-import org.dbsyncer.parser.model.Connector;
-import org.dbsyncer.parser.model.Mapping;
-import org.dbsyncer.parser.model.TableGroup;
-import org.junit.Test;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URL;
-import java.util.Map;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2019/10/9 23:46
- */
-public class ConnectorParserTest {
-
-    @Test
-    public void testConnector() throws IOException {
-        String json = readJson("Connector.json");
-        System.out.println(json);
-
-        // 解析基本信息
-        Map conn = JsonUtil.parseMap(json);
-        Map config = (Map) conn.remove("config");
-        Connector connector = JsonUtil.jsonToObj(conn.toString(), Connector.class);
-
-        // 解析配置
-        String connectorType = (String) config.get("connectorType");
-        Class<?> configClass = ConnectorEnum.getConfigClass(connectorType);
-        Object obj = JsonUtil.jsonToObj(config.toString(), configClass);
-        connector.setConfig((AbstractConnectorConfig) obj);
-        System.out.println(connector);
-    }
-
-    @Test
-    public void testMapping() throws IOException {
-        String json = readJson("Mapping.json");
-        System.out.println(json);
-
-        // 解析基本信息
-        Mapping mapping = JsonUtil.jsonToObj(json, Mapping.class);
-        System.out.println(mapping);
-    }
-
-    @Test
-    public void testTableGroup() throws IOException {
-        String json = readJson("TableGroup.json");
-        System.out.println(json);
-        // 解析基本信息
-        TableGroup tableGroup = JsonUtil.jsonToObj(json, TableGroup.class);
-        System.out.println(tableGroup);
-    }
-
-    /**
-     * 读取JSON文件
-     *
-     * @param fileName
-     * @return
-     * @throws IOException
-     */
-    private String readJson(String fileName) throws IOException {
-        ClassLoader loader = this.getClass().getClassLoader();
-        URL fileURL = loader.getResource(fileName);
-        if (null != fileURL) {
-            String filePath = fileURL.getFile();
-            File file = new File(filePath);
-            return FileUtils.readFileToString(file, "UTF-8");
-        }
-        return "";
-    }
-
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/AbstractStorageService.java

@@ -3,6 +3,7 @@ package org.dbsyncer.storage;
 import org.dbsyncer.common.model.Paging;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.storage.enums.StorageEnum;
+import org.dbsyncer.storage.query.BooleanFilter;
 import org.dbsyncer.storage.query.Query;
 import org.dbsyncer.storage.strategy.Strategy;
 import org.slf4j.Logger;
@@ -37,6 +38,8 @@ public abstract class AbstractStorageService implements StorageService, Disposab
 
     protected abstract Paging select(String sharding, Query query);
 
+    protected abstract void delete(String sharding, Query query);
+
     protected abstract void deleteAll(String sharding);
 
     protected abstract void batchInsert(StorageEnum type, String sharding, List<Map> list);
@@ -79,6 +82,29 @@ public abstract class AbstractStorageService implements StorageService, Disposab
         return new Paging(query.getPageNum(), query.getPageSize());
     }
 
+    @Override
+    public void delete(Query query) {
+        BooleanFilter q = query.getBooleanFilter();
+        if (CollectionUtils.isEmpty(q.getClauses()) && CollectionUtils.isEmpty(q.getFilters())) {
+            throw new StorageException("必须包含删除条件");
+        }
+
+        boolean locked = false;
+        try {
+            locked = lock.tryLock(3, TimeUnit.SECONDS);
+            if (locked) {
+                String sharding = getSharding(query.getType(), query.getMetaId());
+                delete(sharding, query);
+            }
+        } catch (InterruptedException e) {
+            logger.warn("tryLock error", e.getLocalizedMessage());
+        } finally {
+            if (locked) {
+                lock.unlock();
+            }
+        }
+    }
+
     @Override
     public void clear(StorageEnum type, String metaId) {
         boolean locked = false;

dbsyncer-storage/src/main/java/org/dbsyncer/storage/StorageService.java

@@ -22,6 +22,13 @@ public interface StorageService {
      */
     Paging query(Query query);
 
+    /**
+     * 根据条件删除
+     *
+     * @param query
+     */
+    void delete(Query query);
+
     /**
      * 清空数据/日志
      *

dbsyncer-storage/src/main/java/org/dbsyncer/storage/lucene/Shard.java

@@ -91,6 +91,18 @@ public class Shard {
         }
     }
 
+    public void delete(Query query) {
+        try {
+            indexWriter.deleteDocuments(query);
+            indexWriter.flush();
+            indexWriter.commit();
+            indexWriter.forceMergeDeletes();
+            indexWriter.deleteUnusedFiles();
+        } catch (IOException e) {
+            logger.error(e.getLocalizedMessage(), e);
+        }
+    }
+
     public void deleteAll() {
         // Fix Bug: this IndexReader is closed. 直接删除文件
         try {
@@ -247,6 +259,10 @@ public class Shard {
         indexReader = DirectoryReader.open(indexWriter);
     }
 
+    public Analyzer getAnalyzer() {
+        return analyzer;
+    }
+
     interface Callback {
 
         /**

dbsyncer-storage/src/main/java/org/dbsyncer/storage/support/DiskStorageServiceImpl.java

@@ -75,13 +75,7 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
             }
 
             Set<String> highLightKeys = new HashSet<>();
-            BooleanQuery build = null;
-            if (!CollectionUtils.isEmpty(filters)) {
-                build = buildQueryWithFilters(filters, highLightKeys);
-            } else {
-                build = buildQueryWithBooleanFilters(clauses, highLightKeys);
-            }
-
+            BooleanQuery build = buildQuery(filters, clauses, highLightKeys);
             option.setQuery(build);
 
             // 高亮查询
@@ -98,6 +92,13 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
         }
     }
 
+    @Override
+    protected void delete(String sharding, Query query) {
+        Shard shard = getShard(sharding);
+        BooleanFilter q = query.getBooleanFilter();
+        shard.delete(buildQuery(q.getFilters(), q.getClauses(), new HashSet<>()));
+    }
+
     @Override
     public void deleteAll(String sharding) {
         shards.computeIfPresent(sharding, (k, v) -> {
@@ -140,6 +141,13 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
         shards.clear();
     }
 
+    private BooleanQuery buildQuery(List<AbstractFilter> filters, List<BooleanFilter> clauses, Set<String> highLightKeys) {
+        if (!CollectionUtils.isEmpty(filters)) {
+            return buildQueryWithFilters(filters, highLightKeys);
+        }
+        return buildQueryWithBooleanFilters(clauses, highLightKeys);
+    }
+
     private BooleanQuery buildQueryWithFilters(List<AbstractFilter> filters, Set<String> highLightKeys) {
         BooleanQuery.Builder builder = new BooleanQuery.Builder();
         filters.forEach(p -> {

dbsyncer-storage/src/main/java/org/dbsyncer/storage/support/MysqlStorageServiceImpl.java

@@ -121,6 +121,16 @@ public class MysqlStorageServiceImpl extends AbstractStorageService {
         return paging;
     }
 
+    @Override
+    protected void delete(String sharding, Query query) {
+        Executor executor = getExecutor(query.getType(), sharding);
+        StringBuilder sql = new StringBuilder("DELETE FROM `").append(executor.getTable()).append("`");
+        List<Object> params = new ArrayList<>();
+        buildQuerySqlWithParams(query, params, sql, new ArrayList<>());
+        final List<Object[]> args = params.stream().map(val -> new Object[]{val}).collect(Collectors.toList());
+        connectorMapper.execute(databaseTemplate -> databaseTemplate.batchUpdate(sql.toString(), args));
+    }
+
     @Override
     protected void deleteAll(String sharding) {
         tables.computeIfPresent(sharding, (k, executor) -> {
@@ -265,6 +275,7 @@ public class MysqlStorageServiceImpl extends AbstractStorageService {
     private void buildQuerySqlWithFilters(List<AbstractFilter> filters, List<Object> args, StringBuilder sql, List<AbstractFilter> highLightKeys) {
         // 过滤值
         int size = filters.size();
+        String quotation = connector.buildSqlWithQuotation();
         for (int i = 0; i < size; i++) {
             AbstractFilter p = filters.get(i);
 
@@ -276,15 +287,15 @@ public class MysqlStorageServiceImpl extends AbstractStorageService {
             String name = UnderlineToCamelUtils.camelToUnderline(p.getName());
             switch (filterEnum) {
                 case EQUAL:
-                    sql.append(name).append(" = ?");
+                    sql.append(quotation).append(name).append(quotation).append(" = ?");
                     args.add(p.getValue());
                     break;
                 case LIKE:
-                    sql.append(name).append(" LIKE ?");
+                    sql.append(quotation).append(name).append(quotation).append(" LIKE ?");
                     args.add(new StringBuilder("%").append(p.getValue()).append("%"));
                     break;
                 case LT:
-                    sql.append(name).append(" < ?");
+                    sql.append(quotation).append(name).append(quotation).append(" < ?");
                     args.add(p.getValue());
                     break;
             }

dbsyncer-storage/src/main/test/LuceneFactoryTest.java

@@ -2,26 +2,47 @@ import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.analysis.TokenStream;
 import org.apache.lucene.analysis.cn.smart.SmartChineseAnalyzer;
 import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;
-import org.apache.lucene.document.*;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.Field;
+import org.apache.lucene.document.IntPoint;
+import org.apache.lucene.document.LongPoint;
+import org.apache.lucene.document.NumericDocValuesField;
+import org.apache.lucene.document.StoredField;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.document.TextField;
 import org.apache.lucene.index.IndexReader;
 import org.apache.lucene.index.Term;
 import org.apache.lucene.queryparser.classic.ParseException;
 import org.apache.lucene.queryparser.classic.QueryParser;
-import org.apache.lucene.search.*;
+import org.apache.lucene.search.BooleanClause;
+import org.apache.lucene.search.BooleanQuery;
+import org.apache.lucene.search.FuzzyQuery;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.PhraseQuery;
+import org.apache.lucene.search.PrefixQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.search.ScoreDoc;
+import org.apache.lucene.search.Sort;
+import org.apache.lucene.search.SortField;
+import org.apache.lucene.search.TermQuery;
+import org.apache.lucene.search.TopDocs;
+import org.apache.lucene.search.WildcardQuery;
 import org.apache.lucene.search.highlight.Highlighter;
 import org.apache.lucene.search.highlight.InvalidTokenOffsetsException;
 import org.apache.lucene.search.highlight.QueryScorer;
 import org.apache.lucene.search.highlight.SimpleHTMLFormatter;
 import org.apache.lucene.util.BytesRef;
 import org.dbsyncer.common.model.Paging;
+import org.dbsyncer.common.snowflake.SnowflakeIdWorker;
 import org.dbsyncer.common.util.CollectionUtils;
-import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.common.util.RandomUtil;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.dbsyncer.storage.enums.StorageDataStatusEnum;
+import org.dbsyncer.storage.lucene.Option;
 import org.dbsyncer.storage.lucene.Shard;
-import org.dbsyncer.storage.query.Option;
+import org.dbsyncer.storage.util.BinlogMessageUtil;
 import org.dbsyncer.storage.util.DocumentUtil;
 import org.junit.After;
 import org.junit.Before;
@@ -32,15 +53,24 @@ import org.slf4j.LoggerFactory;
 import java.io.IOException;
 import java.io.StringReader;
 import java.time.Instant;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.*;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.CyclicBarrier;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 public class LuceneFactoryTest {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
-    private       Shard  shard;
+
+    private Shard shard;
+
+    private SnowflakeIdWorker snowflakeIdWorker = new SnowflakeIdWorker();
 
     @Before
     public void setUp() throws IOException {
@@ -60,7 +90,7 @@ public class LuceneFactoryTest {
         final CyclicBarrier barrier = new CyclicBarrier(threadSize);
         final CountDownLatch latch = new CountDownLatch(threadSize);
         for (int i = 0; i < threadSize; i++) {
-            final int k = i + 3;
+            final int k = i + 1;
             pool.submit(() -> {
                 try {
                     barrier.await();
@@ -68,10 +98,9 @@ public class LuceneFactoryTest {
                     // 模拟操作
                     System.out.println(String.format("%s:%s", Thread.currentThread().getName(), k));
 
-                    Document data = DocumentUtil.convertData2Doc(createMap(k));
-                    //IndexableField field = data.getField(ConfigConstant.CONFIG_MODEL_ID);
-                    //shard.update(new Term(ConfigConstant.CONFIG_MODEL_ID, field.stringValue()), data);
-                    shard.insert(data);
+                    List<Document> docs = new ArrayList<>();
+                    docs.add(DocumentUtil.convertData2Doc(createMap(k)));
+                    shard.insertBatch(docs);
 
                 } catch (InterruptedException e) {
                     logger.error(e.getMessage());
@@ -92,7 +121,9 @@ public class LuceneFactoryTest {
             check();
             Sort sort = new Sort(new SortField(ConfigConstant.CONFIG_MODEL_UPDATE_TIME, SortField.Type.LONG, true),
                     new SortField(ConfigConstant.CONFIG_MODEL_CREATE_TIME, SortField.Type.LONG, true));
-            Paging paging = shard.query(new Option(new MatchAllDocsQuery()), 1, 20, sort);
+            Option option = new Option();
+            option.setQuery(new MatchAllDocsQuery());
+            Paging paging = shard.query(option, 1, 20, sort);
             if (!CollectionUtils.isEmpty(paging.getData())) {
                 List<Map> data = (List<Map>) paging.getData();
                 data.stream().forEach(r -> System.out.println(r));
@@ -109,17 +140,19 @@ public class LuceneFactoryTest {
 
     private Map<String, Object> createMap(int i) {
         Map<String, Object> params = new HashMap();
-        params.put(ConfigConstant.CONFIG_MODEL_ID, "953402886828589057");
+        params.put(ConfigConstant.CONFIG_MODEL_ID, String.valueOf(snowflakeIdWorker.nextId()));
         params.put(ConfigConstant.DATA_SUCCESS, StorageDataStatusEnum.SUCCESS.getValue());
         params.put(ConfigConstant.DATA_EVENT, ConnectorConstant.OPERTION_UPDATE);
+        params.put(ConfigConstant.DATA_TABLE_GROUP_ID, "" + i);
+        params.put(ConfigConstant.DATA_TARGET_TABLE_NAME, "MY_USER");
         params.put(ConfigConstant.DATA_ERROR, "");
         Map<String, Object> row = new HashMap<>();
         row.put("id", i);
         row.put("name", "中文");
         row.put("tel", "15800001234");
-        row.put("update_time", System.currentTimeMillis());
+        row.put("update_time", Instant.now().toEpochMilli());
         row.put("remark", "test" + i);
-        params.put(ConfigConstant.CONFIG_MODEL_JSON, JsonUtil.objToJson(row));
+        params.put(ConfigConstant.BINLOG_DATA, BinlogMessageUtil.toBinlogMap(row).toByteArray());
         params.put(ConfigConstant.CONFIG_MODEL_CREATE_TIME, Instant.now().toEpochMilli());
         return params;
     }
@@ -127,6 +160,7 @@ public class LuceneFactoryTest {
     @Test
     public void testQuery() throws IOException {
         int size = 3;
+        List<Document> docs = new ArrayList<>();
         for (int i = size; i > 0; i--) {
             Document doc = new Document();
             doc.add(new StringField("id", String.valueOf(i), Field.Store.YES));
@@ -147,16 +181,18 @@ public class LuceneFactoryTest {
             doc.add(new LongPoint("createTime", createTime));
             doc.add(new StoredField("createTime", createTime));
             doc.add(new NumericDocValuesField("createTime", createTime));
-
-            shard.insert(doc);
+            docs.add(doc);
         }
+        shard.insertBatch(docs);
         // 范围查询 IntPoint.newRangeQuery("id", 1, 100)
         // 集合查询 IntPoint.newSetQuery("id", 2, 3)
         // 单个查询 IntPoint.newExactQuery("id", 3)
         BooleanQuery query = new BooleanQuery.Builder()
                 .add(IntPoint.newRangeQuery("age", 1, 100), BooleanClause.Occur.MUST)
                 .build();
-        Paging paging = shard.query(new Option(query), 1, 20, new Sort(new SortField("createTime", SortField.Type.LONG, true)));
+        Option option = new Option();
+        option.setQuery(query);
+        Paging paging = shard.query(option, 1, 20, new Sort(new SortField("createTime", SortField.Type.LONG, true)));
         paging.getData().forEach(m -> System.out.println(m));
 
         // 清空
@@ -175,7 +211,9 @@ public class LuceneFactoryTest {
         String id = "100";
         doc.add(new StringField("id", id, Field.Store.YES));
         doc.add(new TextField("content", "这是一款大规模数据处理软件，名字叫做Apache Spark", Field.Store.YES));
-        shard.insert(doc);
+        List<Document> docs = new ArrayList<>();
+        docs.add(doc);
+        shard.insertBatch(docs);
         System.out.println("新增后：");
         maps = query(new MatchAllDocsQuery());
         maps.forEach(m -> System.out.println(m));
@@ -190,7 +228,7 @@ public class LuceneFactoryTest {
         check();
 
         // 删除
-        shard.delete(new Term("id", id));
+        shard.deleteBatch(new Term("id", id));
         System.out.println("删除后：");
         maps = query(new MatchAllDocsQuery());
         maps.forEach(m -> System.out.println(m));
@@ -213,7 +251,9 @@ public class LuceneFactoryTest {
         doc.add(new StringField("id", id, Field.Store.YES));
         BytesRef bytesRef = new BytesRef("中文".getBytes());
         doc.add(new StoredField("content", bytesRef));
-        shard.insert(doc);
+        List<Document> docs = new ArrayList<>();
+        docs.add(doc);
+        shard.insertBatch(docs);
         System.out.println("新增后：");
         maps = query(new MatchAllDocsQuery());
         maps.forEach(m -> {
@@ -444,7 +484,9 @@ public class LuceneFactoryTest {
     }
 
     private List<Map> query(Query query) throws IOException {
-        Paging paging = shard.query(new Option(query), 1, 20, null);
+        Option option = new Option();
+        option.setQuery(query);
+        Paging paging = shard.query(option, 1, 20, null);
         return (List<Map>) paging.getData();
     }
 

dbsyncer-storage/src/main/test/ShardBinlogTest.java

@@ -1,202 +0,0 @@
-import com.google.protobuf.ByteString;
-import org.apache.lucene.document.Document;
-import org.apache.lucene.document.IntPoint;
-import org.apache.lucene.document.LongPoint;
-import org.apache.lucene.index.IndexReader;
-import org.apache.lucene.index.Term;
-import org.apache.lucene.search.*;
-import org.apache.lucene.util.BytesRef;
-import org.dbsyncer.common.model.Paging;
-import org.dbsyncer.common.util.DateFormatUtil;
-import org.dbsyncer.storage.binlog.proto.BinlogMap;
-import org.dbsyncer.storage.binlog.proto.BinlogMessage;
-import org.dbsyncer.storage.binlog.proto.EventEnum;
-import org.dbsyncer.storage.constant.BinlogConstant;
-import org.dbsyncer.storage.enums.IndexFieldResolverEnum;
-import org.dbsyncer.storage.lucene.Shard;
-import org.dbsyncer.storage.query.Option;
-import org.dbsyncer.storage.util.BinlogMessageUtil;
-import org.dbsyncer.storage.util.DocumentUtil;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.nio.charset.Charset;
-import java.sql.Date;
-import java.sql.Timestamp;
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/18 23:46
- */
-public class ShardBinlogTest {
-
-    private final Logger logger = LoggerFactory.getLogger(getClass());
-
-    private Shard shard;
-
-    @Before
-    public void init() throws IOException {
-        shard = new Shard("target/indexDir/");
-    }
-
-    @After
-    public void close() throws IOException {
-        shard.deleteAll();
-    }
-
-    @Test
-    public void testBinlogMessage() throws IOException, InterruptedException {
-        mockData(1);
-
-        // 查询[待处理] 或 [处理中 且 处理超时]
-        List<Map> maps = queryReadyAndProcess();
-        logger.info("总条数：{}", maps.size());
-        TimeUnit.SECONDS.sleep(1);
-        markProcessing(maps);
-        logger.info("标记处理中");
-
-        // 模拟新记录
-        TimeUnit.SECONDS.sleep(1);
-        mockData(6);
-
-        maps = queryReadyAndProcess();
-        logger.info("【待处理】和【处理中 且 处理超时】总条数：{}", maps.size());
-
-        logger.info("模拟处理超时，等待10s");
-        TimeUnit.SECONDS.sleep(10);
-
-        maps = queryReadyAndProcess();
-        logger.info("【待处理】和【处理中 且 处理超时】总条数：{}", maps.size());
-    }
-
-    private void markProcessing(List<Map> maps) {
-        long updateTime = Instant.now().toEpochMilli();
-        maps.forEach(row -> {
-            String id = (String) row.get(BinlogConstant.BINLOG_ID);
-            BytesRef ref = (BytesRef) row.get(BinlogConstant.BINLOG_CONTENT);
-            try {
-                shard.update(new Term(BinlogConstant.BINLOG_ID, String.valueOf(id)), DocumentUtil.convertBinlog2Doc(id, BinlogConstant.PROCESSING, ref, updateTime));
-            } catch (IOException e) {
-                throw new RuntimeException(e);
-            }
-        });
-    }
-
-    private void mockData(int i) throws IOException {
-        List<Document> list = new ArrayList<>();
-        long now = Instant.now().toEpochMilli();
-        int size = i + 5;
-        while (i < size) {
-            BinlogMessage message = genMessage("123456", i + "");
-            BytesRef bytes = new BytesRef(message.toByteArray());
-            list.add(DocumentUtil.convertBinlog2Doc(String.valueOf(i), BinlogConstant.READY, bytes, now));
-
-            if (i % 1000 == 0) {
-                shard.insertBatch(list);
-                list.clear();
-            }
-            i++;
-        }
-
-        if (!list.isEmpty()) {
-            shard.insertBatch(list);
-        }
-        check();
-    }
-
-    private List<Map> queryReadyAndProcess() throws IOException {
-        long lastTime = Timestamp.valueOf(LocalDateTime.now().minusSeconds(5)).getTime();
-        BooleanQuery filter1 = new BooleanQuery.Builder()
-                .add(IntPoint.newExactQuery(BinlogConstant.BINLOG_STATUS, BinlogConstant.READY), BooleanClause.Occur.MUST)
-                .build();
-        BooleanQuery filter2 = new BooleanQuery.Builder()
-                .add(IntPoint.newExactQuery(BinlogConstant.BINLOG_STATUS, BinlogConstant.PROCESSING), BooleanClause.Occur.MUST)
-                .add(LongPoint.newRangeQuery(BinlogConstant.BINLOG_TIME, Long.MIN_VALUE, lastTime), BooleanClause.Occur.MUST)
-                .build();
-        BooleanQuery query = new BooleanQuery.Builder()
-                .add(filter1, BooleanClause.Occur.SHOULD)
-                .add(filter2, BooleanClause.Occur.SHOULD)
-                .build();
-        return query(new Option(query));
-    }
-
-    private List<Map> query(Option option) throws IOException {
-        option.addIndexFieldResolverEnum(BinlogConstant.BINLOG_ID, IndexFieldResolverEnum.STRING);
-        option.addIndexFieldResolverEnum(BinlogConstant.BINLOG_STATUS, IndexFieldResolverEnum.INT);
-        option.addIndexFieldResolverEnum(BinlogConstant.BINLOG_CONTENT, IndexFieldResolverEnum.BINARY);
-        option.addIndexFieldResolverEnum(BinlogConstant.BINLOG_TIME, IndexFieldResolverEnum.LONG);
-        Sort sort = new Sort(new SortField(BinlogConstant.BINLOG_TIME, SortField.Type.LONG));
-        Paging paging = shard.query(option, 1, 10001, sort);
-        List<Map> maps = (List<Map>) paging.getData();
-        for (Map m : maps) {
-            String id = (String) m.get(BinlogConstant.BINLOG_ID);
-            Integer s = (Integer) m.get(BinlogConstant.BINLOG_STATUS);
-            BytesRef ref = (BytesRef) m.get(BinlogConstant.BINLOG_CONTENT);
-            Long t = (Long) m.get(BinlogConstant.BINLOG_TIME);
-            BinlogMessage message = BinlogMessage.parseFrom(ref.bytes);
-            Map<String, ByteString> rowMap = message.getData().getRowMap();
-            String timestamp = DateFormatUtil.timestampToString(new Timestamp(t));
-            logger.info("t:{}, id:{}, s:{}, message:{}", timestamp, id, s, rowMap.get("name").toStringUtf8());
-        }
-        return maps;
-    }
-
-    private BinlogMessage genMessage(String tableGroupId, String key) {
-        Map<String, Object> data = new HashMap<>();
-        data.put("id", 1L);
-        data.put("name", key + "中文");
-        data.put("age", 88);
-        data.put("bd", new BigDecimal(88));
-        data.put("bigInt", new BigInteger("123456789876543210"));
-        data.put("localTime", LocalDateTime.now());
-        data.put("sex", 1);
-        data.put("f", 88.88f);
-        data.put("d", 999.99d);
-        data.put("b", true);
-        short ss = 32767;
-        data.put("ss", ss);
-        data.put("bytes", "中文666".getBytes(Charset.defaultCharset()));
-        data.put("create_date", new Date(Timestamp.valueOf(LocalDateTime.now()).getTime()));
-        data.put("update_time", Timestamp.valueOf(LocalDateTime.now()).getTime());
-
-        BinlogMap.Builder builder = BinlogMap.newBuilder();
-        data.forEach((k, v) -> {
-            if (null != v) {
-                ByteString bytes = BinlogMessageUtil.serializeValue(v);
-                if (null != bytes) {
-                    builder.putRow(k, bytes);
-                }
-            }
-        });
-
-        BinlogMessage build = BinlogMessage.newBuilder().setTableGroupId(tableGroupId).setEvent(EventEnum.UPDATE).setData(builder.build()).build();
-        return build;
-    }
-
-    private void check() throws IOException {
-        final IndexSearcher searcher = shard.getSearcher();
-        IndexReader reader = searcher.getIndexReader();
-        // 通过reader可以有效的获取到文档的数量
-        // 有效的索引文档
-        System.out.println("有效的索引文档:" + reader.numDocs());
-        // 总共的索引文档
-        System.out.println("总共的索引文档:" + reader.maxDoc());
-        // 删掉的索引文档，其实不恰当，应该是在回收站里的索引文档
-        System.out.println("删掉的索引文档:" + reader.numDeletedDocs());
-    }
-
-}

dbsyncer-web/src/main/java/org/dbsyncer/web/controller/index/MappingController.java

@@ -44,6 +44,17 @@ public class MappingController extends BaseController {
         return "mapping/" + page;
     }
 
+    @PostMapping("/copy")
+    @ResponseBody
+    public RestResult add(@RequestParam("id") String id) {
+        try {
+            return RestResult.restSuccess(mappingService.copy(id));
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e);
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
     @PostMapping(value = "/add")
     @ResponseBody
     public RestResult add(HttpServletRequest request) {
@@ -51,7 +62,7 @@ public class MappingController extends BaseController {
             Map<String, String> params = getParams(request);
             return RestResult.restSuccess(mappingService.add(params));
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }
@@ -63,7 +74,7 @@ public class MappingController extends BaseController {
             Map<String, String> params = getParams(request);
             return RestResult.restSuccess(mappingService.edit(params));
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }
@@ -74,7 +85,7 @@ public class MappingController extends BaseController {
         try {
             return RestResult.restSuccess(mappingService.remove(id));
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }
@@ -85,7 +96,7 @@ public class MappingController extends BaseController {
         try {
             return RestResult.restSuccess(mappingService.start(id));
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }
@@ -96,7 +107,7 @@ public class MappingController extends BaseController {
         try {
             return RestResult.restSuccess(mappingService.stop(id));
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }
@@ -107,7 +118,7 @@ public class MappingController extends BaseController {
         try {
             return RestResult.restSuccess(mappingService.getMapping(id));
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }
@@ -118,7 +129,7 @@ public class MappingController extends BaseController {
         try {
             return RestResult.restSuccess(mappingService.getMappingAll());
         } catch (Exception e) {
-            logger.error(e.getLocalizedMessage(), e.getClass());
+            logger.error(e.getLocalizedMessage(), e);
             return RestResult.restFail(e.getMessage());
         }
     }

dbsyncer-web/src/main/java/org/dbsyncer/web/controller/monitor/MonitorController.java

@@ -101,6 +101,11 @@ public class MonitorController extends BaseController {
         connectorService.refreshHealth();
     }
 
+    @Scheduled(fixedRate = 30000)
+    public void deleteExpiredDataAndLog() {
+        monitorService.deleteExpiredDataAndLog();
+    }
+
     @GetMapping("/queryData")
     @ResponseBody
     public RestResult queryData(HttpServletRequest request) {
@@ -180,11 +185,11 @@ public class MonitorController extends BaseController {
     }
 
     @ResponseBody
-    @GetMapping("/getRefreshInterval")
+    @GetMapping("/getRefreshIntervalSeconds")
     public RestResult getRefreshInterval() {
         try {
             SystemConfigVo config = systemConfigService.getSystemConfigVo();
-            return RestResult.restSuccess(config.getRefreshInterval());
+            return RestResult.restSuccess(config.getRefreshIntervalSeconds());
         } catch (Exception e) {
             logger.error(e.getLocalizedMessage(), e.getClass());
             return RestResult.restFail(e.getMessage());

dbsyncer-web/src/main/resources/public/index/index.html

@@ -196,6 +196,7 @@
                                                 <li th:if="${m?.meta?.state ne 1}" th:url="'/mapping/start?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-check-circle-o well-sign-green"></i>&nbsp;&nbsp;启动</a></li>
                                                 <!-- 运行中 -->
                                                 <li th:if="${m?.meta?.state eq 1}" th:url="'/mapping/stop?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-times-circle-o well-sign-red"></i>&nbsp;&nbsp;停止</a></li>
+                                                <li th:url="'/mapping/copy?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-copy"></i>&nbsp;复制</a></li>
                                                 <!-- 未运行 -->
                                                 <li th:if="${m?.meta?.state ne 1}" th:url="'/mapping/remove?id='+${m?.id}" confirm="true" confirmMessage="确认删除驱动?"><a href="javascript:;"><i class="fa fa-trash well-sign-red"></i>&nbsp;&nbsp;删除</a></li>
                                             </ul>
@@ -216,4 +217,4 @@
 </div>
 
 <script th:src="@{/js/index/index.js}"></script>
-</html>
+</html>

dbsyncer-web/src/main/resources/public/mapping/editFull.html

@@ -10,19 +10,19 @@
             <div class="col-md-4">
                 <label class="col-sm-3 control-label text-right">批量读取<strong class="driverVerifcateRequired">*</strong></label>
                 <div class="col-sm-9">
-                    <input type="number" name="readNum" class="form-control" min="1" dbsyncer-valid="require" th:value="${mapping?.readNum}">
+                    <input type="number" name="readNum" class="form-control" min="1" max="200000" dbsyncer-valid="require" th:value="${mapping?.readNum}">
                 </div>
             </div>
             <div class="col-md-4">
                 <label class="col-sm-3 control-label text-right">单次写入<strong class="driverVerifcateRequired">*</strong></label>
                 <div class="col-sm-9">
-                    <input type="number" name="batchNum" class="form-control" min="1" dbsyncer-valid="require" th:value="${mapping?.batchNum}">
+                    <input type="number" name="batchNum" class="form-control" min="1" max="20000" dbsyncer-valid="require" th:value="${mapping?.batchNum}">
                 </div>
             </div>
             <div class="col-md-4">
                 <label class="col-sm-3 control-label text-right">线程数<strong class="driverVerifcateRequired">*</strong></label>
                 <div class="col-sm-9">
-                    <input type="number" name="threadNum" class="form-control" min="1" dbsyncer-valid="require" th:value="${mapping?.threadNum}">
+                    <input type="number" name="threadNum" class="form-control" min="1" max="64" dbsyncer-valid="require" th:value="${mapping?.threadNum}">
                 </div>
             </div>
         </div>

dbsyncer-web/src/main/resources/public/system/system.html

@@ -15,10 +15,22 @@
             <!-- 系统参数配置 -->
             <div class="col-md-3"></div>
             <div class="col-md-6">
+                <div class="form-group">
+                    <label class="col-sm-4 control-label">同步数据过期时间(天) <strong class="driverVerifcateRequired">*</strong></label>
+                    <div class="col-sm-8">
+                        <input type="number" class="form-control" min="1" max="180" dbsyncer-valid="require" name="expireDataDays" th:value="${config?.expireDataDays}"/>
+                    </div>
+                </div>
+                <div class="form-group">
+                    <label class="col-sm-4 control-label">系统日志过期时间(天) <strong class="driverVerifcateRequired">*</strong></label>
+                    <div class="col-sm-8">
+                        <input type="number" class="form-control" min="1" max="180" dbsyncer-valid="require" name="expireLogDays" th:value="${config?.expireLogDays}"/>
+                    </div>
+                </div>
                 <div class="form-group">
                     <label class="col-sm-4 control-label">刷新监控频率(秒) <strong class="driverVerifcateRequired">*</strong></label>
                     <div class="col-sm-8">
-                        <input type="number" class="form-control" max="100" min="1" dbsyncer-valid="require" name="refreshInterval" th:value="${config?.refreshInterval}"/>
+                        <input type="number" class="form-control" min="1" max="60" dbsyncer-valid="require" name="refreshIntervalSeconds" th:value="${config?.refreshIntervalSeconds}"/>
                     </div>
                 </div>
                 <div class="form-group">

dbsyncer-web/src/main/resources/static/js/index/index.js

@@ -188,7 +188,7 @@ function doPost(url) {
 
 // 创建定时器
 function createTimer($projectGroupSelect){
-    doGetWithoutLoading("/monitor/getRefreshInterval",{}, function (data) {
+    doGetWithoutLoading("/monitor/getRefreshIntervalSeconds",{}, function (data) {
         if (data.success == true) {
             timer = setInterval(function(){
                 backIndexPage($projectGroupSelect.selectpicker('val'));
@@ -218,4 +218,4 @@ $(function () {
 
     bindConnectorDropdownMenu();
     bindMappingDropdownMenu();
-});
+});

dbsyncer-web/src/main/resources/static/js/monitor/index.js

@@ -477,7 +477,7 @@ function showChartTable() {
 // 创建定时器
 function createTimer() {
     showChartTable();
-    doGetWithoutLoading("/monitor/getRefreshInterval", {}, function (data) {
+    doGetWithoutLoading("/monitor/getRefreshIntervalSeconds", {}, function (data) {
         if (data.success == true) {
             timer = setInterval(function () {
                 showChartTable();

dbsyncer-web/src/main/resources/static/plugins/js/formValidate/formValidate.js

@@ -23,10 +23,20 @@ $.fn.formValidate = function(opt) {
 }
 
 var formValidateMethod = function($this){
+	let errorClassName = "dbsyncerVerifcateError";
 	if ($this.val() == "") {
-		$this.addClass("dbsyncerVerifcateError").attr("data-original-title", "必填").tooltip({trigger : 'manual'}).tooltip('show');
+		$this.addClass(errorClassName).attr("data-original-title", "必填").tooltip({trigger : 'manual'}).tooltip('show');
 		return false;
 	}
-	$this.tooltip('hide').removeClass("dbsyncerVerifcateError");
+	// 数字类型校验
+	if ($this.attr("type") == "number") {
+		let max = parseInt($this.attr("max"));
+		let min = parseInt($this.attr("min"));
+		if($this.val() > max || $this.val() < min){
+			$this.addClass(errorClassName).attr("data-original-title", "有效范围应在" + min + "-" + max).tooltip({trigger: 'manual'}).tooltip('show');
+			return false;
+		}
+	}
+	$this.tooltip('hide').removeClass(errorClassName);
 	return true;
-}
+}