!64 merge

Merge pull request !64 from AE86/V_1.0.0_Beta

README.md

@@ -1,5 +1,5 @@
 ## 介绍
-DBSyncer是一款开源的数据同步中间件，提供Mysql、Oracle、SqlServer、PostgreSQL、Elasticsearch(ES)、Kafka、File、SQL(Mysql/Oracle/SqlServer/PostgreSQL)等同步场景。支持上传插件自定义同步转换业务，提供监控全量和增量数据统计图、应用性能预警等。
+DBSyncer是一款开源的数据同步中间件，提供Mysql、Oracle、SqlServer、PostgreSQL、Elasticsearch(ES)、Kafka、File、SQL等同步场景。支持上传插件自定义同步转换业务，提供监控全量和增量数据统计图、应用性能预警等。
 
 > 特点
 * 组合驱动，自定义库同步到库组合，关系型数据库与非关系型之间组合，任意搭配表同步映射关系

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

@@ -1,6 +1,7 @@
 package org.dbsyncer.biz;
 
 import org.dbsyncer.biz.vo.ConfigVo;
+import org.dbsyncer.parser.model.ConfigModel;
 
 import java.util.List;
 import java.util.Map;
@@ -40,4 +41,11 @@ public interface ConfigService {
      */
     List<ConfigVo> queryConfig();
 
+    /**
+     * 获取所有配置
+     *
+     * @return
+     */
+    List<ConfigModel> getConfigModelAll();
+
 }

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/AbstractChecker.java

@@ -10,7 +10,7 @@ import org.dbsyncer.parser.model.AbstractConfigModel;
 import org.dbsyncer.parser.model.ConfigModel;
 import org.dbsyncer.parser.model.Convert;
 import org.dbsyncer.plugin.config.Plugin;
-import org.dbsyncer.storage.SnowflakeIdWorker;
+import org.dbsyncer.common.snowflake.SnowflakeIdWorker;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.json.JSONArray;
 import org.json.JSONException;

dbsyncer-biz/src/main/java/org/dbsyncer/biz/checker/impl/connector/AbstractDataBaseConfigChecker.java

@@ -19,6 +19,7 @@ public abstract class AbstractDataBaseConfigChecker implements ConnectorConfigCh
         String password = params.get("password");
         String url = params.get("url");
         String driverClassName = params.get("driverClassName");
+        String primaryKey = params.get("primaryKey");
         Assert.hasText(username, "Username is empty.");
         Assert.hasText(password, "Password is empty.");
         Assert.hasText(url, "Url is empty.");
@@ -28,15 +29,19 @@ public abstract class AbstractDataBaseConfigChecker implements ConnectorConfigCh
         connectorConfig.setPassword(password);
         connectorConfig.setUrl(url);
         connectorConfig.setDriverClassName(driverClassName);
+        connectorConfig.setPrimaryKey(primaryKey);
     }
 
     protected void modifyDql(DatabaseConfig connectorConfig, Map<String, String> params) {
         String sql = params.get("sql");
         String table = params.get("table");
+        String primaryKey = params.get("primaryKey");
         Assert.hasText(sql, "Sql is empty.");
         Assert.hasText(table, "Table is empty.");
+        Assert.hasText(primaryKey, "PrimaryKey is empty.");
         connectorConfig.setSql(sql);
         connectorConfig.setTable(table);
+        connectorConfig.setPrimaryKey(primaryKey);
     }
 
     protected void modifySchema(DatabaseConfig connectorConfig, Map<String, String> params) {

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

@@ -22,10 +22,6 @@ public class DqlMysqlLogConfigChecker implements MappingLogConfigChecker {
     public void modify(Mapping mapping, Map<String, String> params) {
         ListenerConfig config = mapping.getListener();
         Assert.notNull(config, "ListenerConfig can not be null.");
-
-        String label = params.get("incrementStrategyLogTableLabel");
-        Assert.hasText(label, "MysqlLogConfigChecker check params incrementStrategyLogTableLabel is empty");
-        config.setTableLabel(label);
         mapping.setListener(config);
     }
 

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

@@ -22,10 +22,6 @@ public class DqlOracleLogConfigChecker implements MappingLogConfigChecker {
     public void modify(Mapping mapping, Map<String, String> params) {
         ListenerConfig config = mapping.getListener();
         Assert.notNull(config, "ListenerConfig can not be null.");
-
-        String label = params.get("incrementStrategyLogTableLabel");
-        Assert.hasText(label, "DqlOracleLogConfigChecker check params incrementStrategyLogTableLabel is empty");
-        config.setTableLabel(label);
         mapping.setListener(config);
     }
 

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

@@ -12,6 +12,7 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -66,6 +67,16 @@ public class ConfigServiceImpl implements ConfigService {
         return list;
     }
 
+    @Override
+    public List<ConfigModel> getConfigModelAll() {
+        List<ConfigModel> list = new ArrayList<>();
+        manager.getConfigAll().forEach(config -> list.add(config));
+        manager.getConnectorAll().forEach(config -> list.add(config));
+        manager.getMappingAll().forEach(config -> list.add(config));
+        manager.getMetaAll().forEach(config -> list.add(config));
+        return list;
+    }
+
     private ConfigVo convertConfig2Vo(Config config) {
         ConfigVo configVo = new ConfigVo();
         BeanUtils.copyProperties(config, configVo);

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

@@ -16,6 +16,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.util.Assert;
 
 import java.util.*;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 /**
@@ -98,7 +99,11 @@ public class TableGroupServiceImpl extends BaseServiceImpl implements TableGroup
 
     @Override
     public List<TableGroup> getTableGroupAll(String mappingId) {
-        return manager.getTableGroupAll(mappingId);
+        List<TableGroup> list = manager.getTableGroupAll(mappingId)
+                .stream()
+                .sorted(Comparator.comparing(TableGroup::getUpdateTime).reversed())
+                .collect(Collectors.toList());
+        return list;
     }
 
     private void mergeMappingColumn(String mappingId) {

dbsyncer-cache/src/main/java/org/dbsyncer/cache/CacheService.java

@@ -1,5 +1,7 @@
 package org.dbsyncer.cache;
 
+import java.util.Map;
+
 /**
  * @author AE86
  * @version 1.0.0
@@ -50,4 +52,11 @@ public interface CacheService {
      */
     <T> T get(String key, Class<T> valueType);
 
+    /**
+     * 获取缓存
+     *
+     * @return
+     */
+    Map<String, Object> getAll();
+
 }

dbsyncer-cache/src/main/java/org/dbsyncer/cache/CacheServiceImpl.java

@@ -2,6 +2,7 @@ package org.dbsyncer.cache;
 
 import org.springframework.stereotype.Component;
 
+import java.util.Collections;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -40,4 +41,9 @@ public class CacheServiceImpl implements CacheService {
         return (T) cache.get(key);
     }
 
+    @Override
+    public Map<String, Object> getAll() {
+        return Collections.unmodifiableMap(cache);
+    }
+
 }

dbsyncer-common/src/main/java/org/dbsyncer/common/config/AppConfig.java

@@ -0,0 +1,44 @@
+package org.dbsyncer.common.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/18 0:04
+ */
+@Configuration
+@ConfigurationProperties(prefix = "info.app")
+public class AppConfig {
+
+    private String name;
+
+    private String version;
+
+    private String copyright;
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public String getVersion() {
+        return version;
+    }
+
+    public void setVersion(String version) {
+        this.version = version;
+    }
+
+    public String getCopyright() {
+        return copyright;
+    }
+
+    public void setCopyright(String copyright) {
+        this.copyright = copyright;
+    }
+}

dbsyncer-common/src/main/java/org/dbsyncer/common/config/ScheduleConfig.java

@@ -0,0 +1,50 @@
+package org.dbsyncer.common.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.SchedulingConfigurer;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+
+import java.util.concurrent.RejectedExecutionHandler;
+
+/**
+ * @author yjwang
+ * @date 2022/4/29 10:27
+ */
+@Configuration
+public class ScheduleConfig implements SchedulingConfigurer {
+
+    @Override
+    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
+        scheduledTaskRegistrar.setTaskScheduler(taskScheduler());
+    }
+
+    @Bean(name = "taskScheduler", destroyMethod = "shutdown")
+    public ThreadPoolTaskScheduler taskScheduler() {
+        int poolSize = Runtime.getRuntime().availableProcessors() * 2;
+        ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
+        //核心线程池大小
+        scheduler.setPoolSize(poolSize);
+        //线程名字前缀
+        scheduler.setThreadNamePrefix("taskScheduler-");
+        //设置线程池关闭的时候等待所有任务都完成再继续销毁其他的Bean
+        scheduler.setWaitForTasksToCompleteOnShutdown(true);
+        //设置线程池中任务的等待时间，如果超过这个时候还没有销毁就强制销毁，以确保应用最后能够被关闭，而不是阻塞住
+        scheduler.setAwaitTerminationSeconds(60);
+        // 线程池满，拒绝策略
+        scheduler.setRejectedExecutionHandler(rejectedExecutionHandler());
+
+        return scheduler;
+    }
+
+    public RejectedExecutionHandler rejectedExecutionHandler() {
+        return (r, executor) -> {
+            try {
+                executor.getQueue().put(r);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        };
+    }
+}

dbsyncer-common/src/main/java/org/dbsyncer/common/config/ThreadPoolConfig.java

@@ -25,12 +25,12 @@ public class ThreadPoolConfig {
     /**
      * 最大工作线程数
      */
-    private int maxSize = 128;
+    private int maxSize = 64;
 
     /**
      * 工作线任务队列
      */
-    private int queueCapacity = 2000;
+    private int queueCapacity = 1000;
 
     @Bean("taskExecutor")
     public Executor taskExecutor() {

dbsyncer-common/src/main/java/org/dbsyncer/common/event/RowChangedEvent.java

@@ -68,10 +68,18 @@ public class RowChangedEvent {
         return beforeData;
     }
 
+    public void setBeforeData(List<Object> beforeData) {
+        this.beforeData = beforeData;
+    }
+
     public List<Object> getAfterData() {
         return afterData;
     }
 
+    public void setAfterData(List<Object> afterData) {
+        this.afterData = afterData;
+    }
+
     public Map<String, Object> getBefore() {
         return before;
     }

dbsyncer-storage/src/main/java/org/dbsyncer/storage/SnowflakeIdWorker.java → dbsyncer-common/src/main/java/org/dbsyncer/common/snowflake/SnowflakeIdWorker.java

@@ -1,36 +1,49 @@
-package org.dbsyncer.storage;
+package org.dbsyncer.common.snowflake;
 
-import org.springframework.beans.factory.annotation.Value;
+import org.dbsyncer.common.CommonException;
+import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.stereotype.Component;
 
 import java.time.Instant;
 
 @Component
+@ConfigurationProperties(prefix = "dbsyncer.common.worker")
 public class SnowflakeIdWorker {
+
     /**
-     * 开始时间截 (2015-01-01)
+     * 工作机器ID(0~31)
      */
-    private final long twepoch = 1420041600000L;
+    private long id = 1L;
 
     /**
-     * 机器id所占的位数
+     * 数据中心ID(0~31)
      */
-    private final long workerIdBits = 5L;
+    private long dataCenterId = 1L;
 
     /**
-     * 数据标识id所占的位数
+     * 毫秒内序列(0~4095)
+     */
+    private long sequence = 0L;
+
+    /**
+     * 上次生成ID的时间截
+     */
+    private long lastTimestamp = -1L;
+
+    /**
+     * 开始时间截 (2015-01-01)
      */
-    private final long datacenterIdBits = 5L;
+    private final long twepoch = 1420041600000L;
 
     /**
-     * 支持的最大机器id，结果是31 (这个移位算法可以很快的计算出几位二进制数所能表示的最大十进制数)
+     * 机器id所占的位数
      */
-    private final long maxWorkerId = -1L ^ (-1L << workerIdBits);
+    private final long workerIdBits = 5L;
 
     /**
-     * 支持的最大数据标识id，结果是31
+     * 数据标识id所占的位数
      */
-    private final long maxDatacenterId = -1L ^ (-1L << datacenterIdBits);
+    private final long dataCenterIdBits = 5L;
 
     /**
      * 序列在id中占的位数
@@ -45,44 +58,23 @@ public class SnowflakeIdWorker {
     /**
      * 数据标识id向左移17位(12+5)
      */
-    private final long datacenterIdShift = sequenceBits + workerIdBits;
+    private final long dataCenterIdShift = sequenceBits + workerIdBits;
 
     /**
      * 时间截向左移22位(5+5+12)
      */
-    private final long timestampLeftShift = sequenceBits + workerIdBits + datacenterIdBits;
+    private final long timestampLeftShift = sequenceBits + workerIdBits + dataCenterIdBits;
 
     /**
      * 生成序列的掩码，这里为4095 (0b111111111111=0xfff=4095)
      */
     private final long sequenceMask = -1L ^ (-1L << sequenceBits);
 
-    /**
-     * 工作机器ID(0~31)
-     */
-    @Value(value = "${dbsyncer.storage.id}")
-    private long workerId;
-
-    /**
-     * 数据中心ID(0~31)
-     */
-    private long datacenterId;
-
-    /**
-     * 毫秒内序列(0~4095)
-     */
-    private long sequence = 0L;
-
-    /**
-     * 上次生成ID的时间截
-     */
-    private long lastTimestamp = -1L;
-
     public SnowflakeId revert(Long id) {
         long workerId = id >> workerIdShift & ~(-1L << workerIdBits);
-        long datacenterId = id >> datacenterIdShift & ~(-1L << datacenterIdBits);
+        long dataCenterId = id >> dataCenterIdShift & ~(-1L << dataCenterIdBits);
         long timestamp = new Long(id >> timestampLeftShift) + twepoch;
-        return new SnowflakeId(workerId, datacenterId, timestamp);
+        return new SnowflakeId(workerId, dataCenterId, timestamp);
     }
 
     /**
@@ -95,8 +87,7 @@ public class SnowflakeIdWorker {
 
         //如果当前时间小于上一次ID生成的时间戳，说明系统时钟回退过这个时候应当抛出异常
         if (timestamp < lastTimestamp) {
-            throw new StorageException(
-                    String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
+            throw new CommonException(String.format("Clock moved backwards.  Refusing to generate id for %d milliseconds", lastTimestamp - timestamp));
         }
 
         //如果是同一时间生成的，则进行毫秒内序列
@@ -118,8 +109,8 @@ public class SnowflakeIdWorker {
 
         //移位并通过或运算拼到一起组成64位的ID
         return ((timestamp - twepoch) << timestampLeftShift) //
-                | (datacenterId << datacenterIdShift) //
-                | (workerId << workerIdShift) //
+                | (dataCenterId << dataCenterIdShift) //
+                | (id << workerIdShift) //
                 | sequence;
     }
 
@@ -146,23 +137,30 @@ public class SnowflakeIdWorker {
         return Instant.now().toEpochMilli();
     }
 
-//    public static void main(String[] args) {
-//        SnowflakeIdWorker idWorker = new SnowflakeIdWorker();
-//        for (int i = 0; i < 1000; i++) {
-//            long id = idWorker.nextId();
-//            System.out.println(Long.toBinaryString(id));
-//            System.out.println(id);
-//        }
-//    }
+    public long getId() {
+        return id;
+    }
+
+    public void setId(long id) {
+        this.id = id;
+    }
+
+    public long getDataCenterId() {
+        return dataCenterId;
+    }
+
+    public void setDataCenterId(long dataCenterId) {
+        this.dataCenterId = dataCenterId;
+    }
 
     private class SnowflakeId {
         private long workerId;
-        private long datacenterId;
+        private long dataCenterId;
         private long timestamp;
 
-        public SnowflakeId(long workerId, long datacenterId, long timestamp) {
+        public SnowflakeId(long workerId, long dataCenterId, long timestamp) {
             this.workerId = workerId;
-            this.datacenterId = datacenterId;
+            this.dataCenterId = dataCenterId;
             this.timestamp = timestamp;
         }
 
@@ -170,8 +168,8 @@ public class SnowflakeIdWorker {
             return workerId;
         }
 
-        public long getDatacenterId() {
-            return datacenterId;
+        public long getDataCenterId() {
+            return dataCenterId;
         }
 
         public long getTimestamp() {

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/CommandConfig.java

@@ -24,13 +24,11 @@ public class CommandConfig {
 
     private ConnectorConfig connectorConfig;
 
-    public CommandConfig(String type, Table table, Table originalTable) {
-        this.type = type;
-        this.table = table;
-        this.originalTable = originalTable;
+    public CommandConfig(String type, Table table, Table originalTable, ConnectorConfig connectorConfig) {
+        this(type, table, originalTable, connectorConfig, null);
     }
 
-    public CommandConfig(String type, Table table, Table originalTable, List<Filter> filter, ConnectorConfig connectorConfig) {
+    public CommandConfig(String type, Table table, Table originalTable, ConnectorConfig connectorConfig, List<Filter> filter) {
         this.type = type;
         this.table = table;
         this.originalTable = originalTable;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/DatabaseConfig.java

@@ -11,35 +11,56 @@ import java.util.Map;
  */
 public class DatabaseConfig extends ConnectorConfig {
 
-    // 驱动com.mysql.jdbc.Driver
+    /**
+     * 驱动
+     */
     private String driverClassName;
 
-    // 连接地址
+    /**
+     * 连接地址
+     */
     private String url;
 
-    // 帐号
+    /**
+     * 帐号
+     */
     private String username;
 
-    // 密码
+    /**
+     * 密码
+     */
     private String password;
 
-    // 通过SQL获取表
+    /**
+     * 主表
+     */
     private String table;
 
-    // 通过SQL获取表信息
+    /**
+     * 主键
+     */
+    private String primaryKey;
+
+    /**
+     * SQL
+     */
     private String sql;
 
-    // 构架名
+    /**
+     * 构架名
+     */
     private String schema;
 
-    // 参数配置
+    /**
+     * 参数配置
+     */
     private Map<String, String> properties = new LinkedHashMap<>();
 
-    public String getProperty(String key){
+    public String getProperty(String key) {
         return properties.get(key);
     }
 
-    public String getProperty(String key, String defaultValue){
+    public String getProperty(String key, String defaultValue) {
         return properties.containsKey(key) ? properties.get(key) : defaultValue;
     }
 
@@ -83,6 +104,14 @@ public class DatabaseConfig extends ConnectorConfig {
         this.table = table;
     }
 
+    public String getPrimaryKey() {
+        return primaryKey;
+    }
+
+    public void setPrimaryKey(String primaryKey) {
+        this.primaryKey = primaryKey;
+    }
+
     public String getSql() {
         return sql;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/config/SqlBuilderConfig.java

@@ -19,13 +19,7 @@ public class SqlBuilderConfig {
     // 引号
     private String quotation;
 
-    public SqlBuilderConfig(String name, String pk) {
-        this.tableName = name;
-        this.pk = pk;
-    }
-
-    public SqlBuilderConfig(Database database, String tableName, String pk, List<Field> fields, String queryFilter,
-                            String quotation) {
+    public SqlBuilderConfig(Database database, String tableName, String pk, List<Field> fields, String queryFilter, String quotation) {
         this.database = database;
         this.tableName = tableName;
         this.pk = pk;

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/AbstractDatabaseConnector.java

@@ -12,7 +12,6 @@ import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.enums.OperationEnum;
 import org.dbsyncer.connector.enums.SetterEnum;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
-import org.dbsyncer.connector.enums.TableTypeEnum;
 import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.connector.model.MetaInfo;
@@ -175,20 +174,17 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
         Map<String, String> map = new HashMap<>();
 
         String query = ConnectorConstant.OPERTION_QUERY;
-        map.put(query, buildSql(query, table, commandConfig.getOriginalTable(), queryFilterSql));
+        map.put(query, buildSql(query, commandConfig, queryFilterSql));
 
         // 获取查询总数SQL
         String quotation = buildSqlWithQuotation();
-        String pk = findTablePrimaryKey(commandConfig.getOriginalTable(), quotation);
+        String pk = findOriginalTablePrimaryKey(commandConfig, quotation);
         StringBuilder queryCount = new StringBuilder();
         queryCount.append("SELECT COUNT(1) FROM (SELECT 1 FROM ").append(quotation).append(table.getName()).append(quotation);
         if (StringUtil.isNotBlank(queryFilterSql)) {
             queryCount.append(queryFilterSql);
         }
-        if (!StringUtil.isBlank(pk)) {
-            queryCount.append(" GROUP BY ").append(pk);
-        }
-        queryCount.append(") DBSYNCER_T");
+        queryCount.append(" GROUP BY ").append(pk).append(") DBSYNCER_T");
         map.put(ConnectorConstant.OPERTION_QUERY_COUNT, queryCount.toString());
         return map;
     }
@@ -197,22 +193,19 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
     public Map<String, String> getTargetCommand(CommandConfig commandConfig) {
         // 获取增删改SQL
         Map<String, String> map = new HashMap<>();
-        Table table = commandConfig.getTable();
-        Table originalTable = commandConfig.getOriginalTable();
-
         String insert = SqlBuilderEnum.INSERT.getName();
-        map.put(insert, buildSql(insert, table, originalTable, null));
+        map.put(insert, buildSql(insert, commandConfig, null));
 
         String update = SqlBuilderEnum.UPDATE.getName();
-        map.put(update, buildSql(update, table, originalTable, null));
+        map.put(update, buildSql(update, commandConfig, null));
 
         String delete = SqlBuilderEnum.DELETE.getName();
-        map.put(delete, buildSql(delete, table, originalTable, null));
+        map.put(delete, buildSql(delete, commandConfig, null));
 
         // 获取查询数据行是否存在
         String quotation = buildSqlWithQuotation();
-        String pk = findTablePrimaryKey(commandConfig.getOriginalTable(), quotation);
-        StringBuilder queryCount = new StringBuilder().append("SELECT COUNT(1) FROM ").append(quotation).append(table.getName()).append(
+        String pk = findOriginalTablePrimaryKey(commandConfig, quotation);
+        StringBuilder queryCount = new StringBuilder().append("SELECT COUNT(1) FROM ").append(quotation).append(commandConfig.getTable().getName()).append(
                 quotation).append(" WHERE ").append(pk).append(" = ?");
         String queryCountExist = ConnectorConstant.OPERTION_QUERY_COUNT_EXIST;
         map.put(queryCountExist, queryCount.toString());
@@ -323,19 +316,18 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
      * 获取查询SQL
      *
      * @param type           {@link SqlBuilderEnum}
-     * @param table
-     * @param originalTable
+     * @param commandConfig
      * @param queryFilterSQL
      * @return
      */
-    protected String buildSql(String type, Table table, Table originalTable, String queryFilterSQL) {
+    protected String buildSql(String type, CommandConfig commandConfig, String queryFilterSQL) {
+        Table table = commandConfig.getTable();
         if (null == table) {
             logger.error("Table can not be null.");
             throw new ConnectorException("Table can not be null.");
         }
         List<Field> column = table.getColumn();
         if (CollectionUtils.isEmpty(column)) {
-            logger.warn("Table column is null.");
             return null;
         }
         String pk = null;
@@ -364,7 +356,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
             throw new ConnectorException("Table name can not be empty.");
         }
         if (StringUtil.isBlank(pk)) {
-            pk = findTablePrimaryKey(originalTable, "");
+            pk = findOriginalTablePrimaryKey(commandConfig, "");
         }
 
         SqlBuilderConfig config = new SqlBuilderConfig(this, tableName, pk, fields, queryFilterSQL, buildSqlWithQuotation());
@@ -420,11 +412,12 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
     /**
      * 返回主键名称
      *
-     * @param table
+     * @param commandConfig
      * @param quotation
      * @return
      */
-    protected String findTablePrimaryKey(Table table, String quotation) {
+    protected String findOriginalTablePrimaryKey(CommandConfig commandConfig, String quotation) {
+        Table table = commandConfig.getOriginalTable();
         if (null != table) {
             List<Field> column = table.getColumn();
             if (!CollectionUtils.isEmpty(column)) {
@@ -435,10 +428,13 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
                 }
             }
         }
-        if (!TableTypeEnum.isView(table.getType())) {
+
+        DatabaseConfig cfg = (DatabaseConfig) commandConfig.getConnectorConfig();
+        if (StringUtil.isBlank(cfg.getPrimaryKey())) {
             throw new ConnectorException("Table primary key can not be empty.");
         }
-        return "";
+
+        return new StringBuilder(quotation).append(cfg.getPrimaryKey()).append(quotation).toString();
     }
 
     /**

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/AbstractDQLConnector.java

@@ -6,7 +6,6 @@ import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
-import org.dbsyncer.connector.model.Filter;
 import org.dbsyncer.connector.model.MetaInfo;
 import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.model.Table;
@@ -27,7 +26,7 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
     public List<Table> getTable(DatabaseConnectorMapper config) {
         DatabaseConfig cfg = config.getConfig();
         List<Table> tables = new ArrayList<>();
-        tables.add(new Table(cfg.getSql()));
+        tables.add(new Table(cfg.getTable()));
         return tables;
     }
 
@@ -56,20 +55,19 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
      */
     protected Map<String, String> getDqlSourceCommand(CommandConfig commandConfig, boolean groupByPK) {
         // 获取过滤SQL
-        List<Filter> filter = commandConfig.getFilter();
-        String queryFilterSql = getQueryFilterSql(filter);
+        String queryFilterSql = getQueryFilterSql(commandConfig.getFilter());
+        DatabaseConfig cfg = (DatabaseConfig) commandConfig.getConnectorConfig();
 
         // 获取查询SQL
-        Table table = commandConfig.getTable();
         Map<String, String> map = new HashMap<>();
-        String querySql = table.getName();
+        String querySql = cfg.getSql();
 
         // 存在条件
         if (StringUtil.isNotBlank(queryFilterSql)) {
             querySql += queryFilterSql;
         }
         String quotation = buildSqlWithQuotation();
-        String pk = findTablePrimaryKey(commandConfig.getOriginalTable(), quotation);
+        String pk = new StringBuilder(quotation).append(cfg.getPrimaryKey()).append(quotation).toString();
         map.put(SqlBuilderEnum.QUERY.getName(), getPageSql(new PageSql(querySql, pk)));
 
         // 获取查询总数SQL

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sql/DQLSqlServerConnector.java

@@ -1,9 +1,7 @@
 package org.dbsyncer.connector.sql;
 
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorException;
-import org.dbsyncer.connector.model.PageSql;
 import org.dbsyncer.connector.constant.DatabaseConstant;
+import org.dbsyncer.connector.model.PageSql;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -13,10 +11,6 @@ public final class DQLSqlServerConnector extends AbstractDQLConnector {
 
     @Override
     public String getPageSql(PageSql config) {
-        if (StringUtil.isBlank(config.getPk())) {
-            logger.error("Table primary key can not be empty.");
-            throw new ConnectorException("Table primary key can not be empty.");
-        }
         return String.format(DatabaseConstant.SQLSERVER_PAGE_SQL, config.getPk(), config.getQuerySql());
     }
 

dbsyncer-connector/src/main/java/org/dbsyncer/connector/sqlserver/SqlServerConnector.java

@@ -1,15 +1,14 @@
 package org.dbsyncer.connector.sqlserver;
 
 import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.CommandConfig;
 import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.model.PageSql;
-import org.dbsyncer.connector.model.Table;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.constant.DatabaseConstant;
 import org.dbsyncer.connector.database.AbstractDatabaseConnector;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
+import org.dbsyncer.connector.model.PageSql;
+import org.dbsyncer.connector.model.Table;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,10 +28,6 @@ public final class SqlServerConnector extends AbstractDatabaseConnector {
 
     @Override
     public String getPageSql(PageSql config) {
-        if (StringUtil.isBlank(config.getPk())) {
-            logger.error("Table primary key can not be empty.");
-            throw new ConnectorException("Table primary key can not be empty.");
-        }
         return String.format(DatabaseConstant.SQLSERVER_PAGE_SQL, config.getPk(), config.getQuerySql());
     }
 
@@ -51,7 +46,7 @@ public final class SqlServerConnector extends AbstractDatabaseConnector {
         Map<String, String> map = new HashMap<>();
 
         String query = ConnectorConstant.OPERTION_QUERY;
-        map.put(query, this.buildSql(query, table, commandConfig.getOriginalTable(), queryFilterSql));
+        map.put(query, this.buildSql(query, commandConfig, queryFilterSql));
 
         // 获取查询总数SQL
         StringBuilder queryCount = new StringBuilder();

dbsyncer-connector/src/main/resources/META-INF/spring.factories

@@ -1,3 +1,3 @@
-# 在这里开启自动状态，遵循spring 扫描规则
+# 鍦ㄨ繖閲屽紑鍚�嚜鍔ㄧ姸鎬侊紝閬靛惊spring 鎵�弿瑙勫垯
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
 org.dbsyncer.connector.AutoConfiguration

dbsyncer-listener/src/main/java/org/dbsyncer/listener/AbstractDatabaseExtractor.java

@@ -0,0 +1,106 @@
+package org.dbsyncer.listener;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.connector.database.DatabaseConnectorMapper;
+import org.dbsyncer.connector.model.Field;
+import org.dbsyncer.connector.model.MetaInfo;
+import org.springframework.util.Assert;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/29 21:46
+ */
+public abstract class AbstractDatabaseExtractor extends AbstractExtractor {
+
+    private DqlMapper dqlMapper;
+
+    /**
+     * 发送增量事件
+     *
+     * @param event
+     */
+    protected abstract void sendChangedEvent(RowChangedEvent event);
+
+    /**
+     * 发送DQL增量事件
+     *
+     * @param event
+     */
+    protected void sendDqlChangedEvent(RowChangedEvent event) {
+        if (null != event && event.getSourceTableName().equals(dqlMapper.tableName)) {
+            switch (event.getEvent()){
+                case ConnectorConstant.OPERTION_UPDATE:
+                case ConnectorConstant.OPERTION_INSERT:
+                    event.setAfterData(queryData(event.getAfterData()));
+                    break;
+                default:
+                    break;
+            }
+            changedEvent(event);
+        }
+    }
+
+    /**
+     * 初始化Dql连接配置
+     */
+    protected void postProcessDqlBeforeInitialization() {
+        DatabaseConnectorMapper mapper = (DatabaseConnectorMapper) connectorFactory.connect(connectorConfig);
+        DatabaseConfig cfg = mapper.getConfig();
+        final String tableName = cfg.getTable();
+        final String primaryKey = cfg.getPrimaryKey();
+        Assert.hasText(tableName, String.format("The table name '%s' is null.", tableName));
+        MetaInfo metaInfo = connectorFactory.getMetaInfo(mapper, tableName);
+        final List<Field> column = metaInfo.getColumn();
+        Assert.notEmpty(column, String.format("The column of table name '%s' is empty.", tableName));
+
+        int pkIndex = 0;
+        boolean findPkIndex = false;
+        for (Field f : column) {
+            if (f.isPk() && f.getName().equals(primaryKey)) {
+                pkIndex = column.indexOf(f);
+                findPkIndex = true;
+                break;
+            }
+        }
+        Assert.isTrue(findPkIndex, "The primaryKey is invalid.");
+        String sql = new StringBuilder(cfg.getSql()).append(" AND ").append(cfg.getPrimaryKey()).append("=?").toString();
+
+        dqlMapper = new DqlMapper(mapper, tableName, column, pkIndex, sql);
+    }
+
+    private List<Object> queryData(List<Object> data) {
+        if (data.size() >= dqlMapper.pkIndex) {
+            Map<String, Object> row = dqlMapper.mapper.execute(databaseTemplate -> databaseTemplate.queryForMap(dqlMapper.sql, data.get(dqlMapper.pkIndex)));
+            if (!CollectionUtils.isEmpty(row)) {
+                data.clear();
+                dqlMapper.column.forEach(field -> data.add(row.get(field.getName())));
+            }
+        }
+        return data;
+    }
+
+    final class DqlMapper {
+        DatabaseConnectorMapper mapper;
+        String tableName;
+        List<Field> column;
+        int pkIndex;
+        String sql;
+
+        public DqlMapper(DatabaseConnectorMapper mapper, String tableName, List<Field> column, int pkIndex, String sql) {
+            this.mapper = mapper;
+            this.tableName = tableName;
+            this.column = column;
+            this.pkIndex = pkIndex;
+            this.sql = sql;
+        }
+    }
+
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/Listener.java

@@ -1,7 +1,9 @@
 package org.dbsyncer.listener;
 
+import org.dbsyncer.listener.enums.ListenerTypeEnum;
+
 public interface Listener {
 
-    <T> T getExtractor(String groupType, String listenerType, Class<T> valueType) throws IllegalAccessException, InstantiationException;
+    <T> T getExtractor(ListenerTypeEnum listenerTypeEnum, String connectorType, Class<T> valueType) throws IllegalAccessException, InstantiationException;
 
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/ListenerFactory.java

@@ -1,15 +1,38 @@
 package org.dbsyncer.listener;
 
-import org.dbsyncer.listener.enums.ListenerEnum;
+import org.dbsyncer.listener.enums.ListenerTypeEnum;
+import org.dbsyncer.listener.enums.LogExtractorEnum;
+import org.dbsyncer.listener.enums.TimingExtractorEnum;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.PostConstruct;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
 @Component
 public class ListenerFactory implements Listener {
 
+    private Map<ListenerTypeEnum, ExtractorMapper> map = new LinkedHashMap<>();
+
+    @PostConstruct
+    private void init() {
+        map.putIfAbsent(ListenerTypeEnum.LOG, (connectorType) -> LogExtractorEnum.getExtractor(connectorType));
+        map.putIfAbsent(ListenerTypeEnum.TIMING, (connectorType) -> TimingExtractorEnum.getExtractor(connectorType));
+    }
+
     @Override
-    public <T> T getExtractor(String groupType, String listenerType, Class<T> valueType) throws IllegalAccessException, InstantiationException {
-        Class<T> clazz = (Class<T>) ListenerEnum.getExtractor(groupType + listenerType);
+    public <T> T getExtractor(ListenerTypeEnum listenerTypeEnum, String connectorType, Class<T> valueType) throws IllegalAccessException, InstantiationException {
+        ExtractorMapper mapper = map.get(listenerTypeEnum);
+        if (null == mapper) {
+            throw new ListenerException(String.format("Unsupported type \"%s\" for extractor \"%s\".", listenerTypeEnum, connectorType));
+        }
+
+        Class<T> clazz = (Class<T>) mapper.getExtractor(connectorType);
         return clazz.newInstance();
     }
 
+    interface ExtractorMapper {
+        Class getExtractor(String connectorType);
+    }
+
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/config/ListenerConfig.java

@@ -35,9 +35,6 @@ public class ListenerConfig {
     // 删除事件
     private String delete = "D";
 
-    // 表别名
-    private String tableLabel = "T1";
-
     public ListenerConfig() {
     }
 
@@ -93,14 +90,6 @@ public class ListenerConfig {
         this.delete = delete;
     }
 
-    public String getTableLabel() {
-        return tableLabel;
-    }
-
-    public void setTableLabel(String tableLabel) {
-        this.tableLabel = tableLabel;
-    }
-
     public String getCron() {
         return cron;
     }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/enums/ListenerEnum.java

@@ -1,100 +0,0 @@
-package org.dbsyncer.listener.enums;
-
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.connector.enums.ConnectorEnum;
-import org.dbsyncer.listener.ListenerException;
-import org.dbsyncer.listener.file.FileExtractor;
-import org.dbsyncer.listener.kafka.KafkaExtractor;
-import org.dbsyncer.listener.mysql.MysqlExtractor;
-import org.dbsyncer.listener.oracle.OracleExtractor;
-import org.dbsyncer.listener.postgresql.PostgreSQLExtractor;
-import org.dbsyncer.listener.quartz.DatabaseQuartzExtractor;
-import org.dbsyncer.listener.quartz.ESQuartzExtractor;
-import org.dbsyncer.listener.sqlserver.SqlServerExtractor;
-
-/**
- * 监听器Extractor支持日志和定时模式
- *
- * @author AE86
- * @version 1.0.0
- * @date 2020/04/24 14:19
- */
-public enum ListenerEnum {
-
-    /**
-     * log_Mysql
-     */
-    LOG_MYSQL(ListenerTypeEnum.LOG.getType() + ConnectorEnum.MYSQL.getType(), MysqlExtractor.class),
-    /**
-     * log_Oracle
-     */
-    LOG_ORACLE(ListenerTypeEnum.LOG.getType() + ConnectorEnum.ORACLE.getType(), OracleExtractor.class),
-    /**
-     * log_SqlServer
-     */
-    LOG_SQL_SERVER(ListenerTypeEnum.LOG.getType() + ConnectorEnum.SQL_SERVER.getType(), SqlServerExtractor.class),
-    /**
-     * log_PostgreSQL
-     */
-    LOG_POSTGRE_SQL(ListenerTypeEnum.LOG.getType() + ConnectorEnum.POSTGRE_SQL.getType(), PostgreSQLExtractor.class),
-    /**
-     * log_Kafka
-     */
-    LOG_KAFKA(ListenerTypeEnum.LOG.getType() + ConnectorEnum.KAFKA.getType(), KafkaExtractor.class),
-    /**
-     * log_File
-     */
-    LOG_FILE(ListenerTypeEnum.LOG.getType() + ConnectorEnum.FILE.getType(), FileExtractor.class),
-    /**
-     * timing_Mysql
-     */
-    TIMING_MYSQL(ListenerTypeEnum.TIMING.getType() + ConnectorEnum.MYSQL.getType(), DatabaseQuartzExtractor.class),
-    /**
-     * timing_Mysql
-     */
-    TIMING_ORACLE(ListenerTypeEnum.TIMING.getType() + ConnectorEnum.ORACLE.getType(), DatabaseQuartzExtractor.class),
-    /**
-     * timing_SqlServer
-     */
-    TIMING_SQL_SERVER(ListenerTypeEnum.TIMING.getType() + ConnectorEnum.SQL_SERVER.getType(), DatabaseQuartzExtractor.class),
-    /**
-     * timing_PostgreSQL
-     */
-    TIMING_POSTGRE_SQL(ListenerTypeEnum.TIMING.getType() + ConnectorEnum.POSTGRE_SQL.getType(), DatabaseQuartzExtractor.class),
-    /**
-     * timing_Elasticsearch
-     */
-    TIMING_ELASTIC_SEARCH(ListenerTypeEnum.TIMING.getType() + ConnectorEnum.ELASTIC_SEARCH.getType(), ESQuartzExtractor.class);
-
-    private String type;
-    private Class<?> clazz;
-
-    ListenerEnum(String type, Class<?> clazz) {
-        this.type = type;
-        this.clazz = clazz;
-    }
-
-    /**
-     * 获取抽取器
-     *
-     * @param type
-     * @return
-     * @throws ListenerException
-     */
-    public static Class<?> getExtractor(String type) throws ListenerException {
-        for (ListenerEnum e : ListenerEnum.values()) {
-            if (StringUtil.equals(type, e.getType())) {
-                return e.getClazz();
-            }
-        }
-        throw new ListenerException(String.format("Extractor type \"%s\" does not exist.", type));
-    }
-
-    public String getType() {
-        return type;
-    }
-
-    public Class<?> getClazz() {
-        return clazz;
-    }
-}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/enums/LogExtractorEnum.java

@@ -0,0 +1,98 @@
+package org.dbsyncer.listener.enums;
+
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.enums.ConnectorEnum;
+import org.dbsyncer.listener.ListenerException;
+import org.dbsyncer.listener.file.FileExtractor;
+import org.dbsyncer.listener.kafka.KafkaExtractor;
+import org.dbsyncer.listener.mysql.DqlMysqlExtractor;
+import org.dbsyncer.listener.mysql.MysqlExtractor;
+import org.dbsyncer.listener.oracle.DqlOracleExtractor;
+import org.dbsyncer.listener.oracle.OracleExtractor;
+import org.dbsyncer.listener.postgresql.DqlPostgreSQLExtractor;
+import org.dbsyncer.listener.postgresql.PostgreSQLExtractor;
+import org.dbsyncer.listener.sqlserver.DqlSqlServerExtractor;
+import org.dbsyncer.listener.sqlserver.SqlServerExtractor;
+
+/**
+ * 日志模式支持类型
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2020/04/24 14:19
+ */
+public enum LogExtractorEnum {
+
+    /**
+     * Mysql
+     */
+    MYSQL(ConnectorEnum.MYSQL.getType(), MysqlExtractor.class),
+    /**
+     * Oracle
+     */
+    ORACLE(ConnectorEnum.ORACLE.getType(), OracleExtractor.class),
+    /**
+     * SqlServer
+     */
+    SQL_SERVER(ConnectorEnum.SQL_SERVER.getType(), SqlServerExtractor.class),
+    /**
+     * PostgreSQL
+     */
+    POSTGRE_SQL(ConnectorEnum.POSTGRE_SQL.getType(), PostgreSQLExtractor.class),
+    /**
+     * Kafka
+     */
+    KAFKA(ConnectorEnum.KAFKA.getType(), KafkaExtractor.class),
+    /**
+     * File
+     */
+    FILE(ConnectorEnum.FILE.getType(), FileExtractor.class),
+    /**
+     * DqlMysql
+     */
+    DQL_MYSQL(ConnectorEnum.DQL_MYSQL.getType(), DqlMysqlExtractor.class),
+    /**
+     * DqlOracle
+     */
+    DQL_ORACLE(ConnectorEnum.DQL_ORACLE.getType(), DqlOracleExtractor.class),
+    /**
+     * DqlSqlServer
+     */
+    DQL_SQL_SERVER(ConnectorEnum.DQL_SQL_SERVER.getType(), DqlSqlServerExtractor.class),
+    /**
+     * DqlPostgreSQL
+     */
+    DQL_POSTGRE_SQL(ConnectorEnum.DQL_POSTGRE_SQL.getType(), DqlPostgreSQLExtractor.class);
+
+    private String type;
+    private Class clazz;
+
+    LogExtractorEnum(String type, Class clazz) {
+        this.type = type;
+        this.clazz = clazz;
+    }
+
+    /**
+     * 获取抽取器
+     *
+     * @param type
+     * @return
+     * @throws ListenerException
+     */
+    public static Class getExtractor(String type) throws ListenerException {
+        for (LogExtractorEnum e : LogExtractorEnum.values()) {
+            if (StringUtil.equals(type, e.getType())) {
+                return e.getClazz();
+            }
+        }
+        throw new ListenerException(String.format("LogExtractorEnum type \"%s\" does not exist.", type));
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public Class getClazz() {
+        return clazz;
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/enums/TimingExtractorEnum.java

@@ -0,0 +1,70 @@
+package org.dbsyncer.listener.enums;
+
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.enums.ConnectorEnum;
+import org.dbsyncer.listener.ListenerException;
+import org.dbsyncer.listener.quartz.DatabaseQuartzExtractor;
+import org.dbsyncer.listener.quartz.ESQuartzExtractor;
+
+/**
+ * 定时模式支持类型
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2020/04/24 14:19
+ */
+public enum TimingExtractorEnum {
+
+    /**
+     * Mysql
+     */
+    MYSQL(ConnectorEnum.MYSQL.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * Mysql
+     */
+    ORACLE(ConnectorEnum.ORACLE.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * SqlServer
+     */
+    SQL_SERVER(ConnectorEnum.SQL_SERVER.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * PostgreSQL
+     */
+    POSTGRE_SQL(ConnectorEnum.POSTGRE_SQL.getType(), DatabaseQuartzExtractor.class),
+    /**
+     * Elasticsearch
+     */
+    ELASTIC_SEARCH(ConnectorEnum.ELASTIC_SEARCH.getType(), ESQuartzExtractor.class);
+
+    private String type;
+    private Class clazz;
+
+    TimingExtractorEnum(String type, Class clazz) {
+        this.type = type;
+        this.clazz = clazz;
+    }
+
+    /**
+     * 获取抽取器
+     *
+     * @param type
+     * @return
+     * @throws ListenerException
+     */
+    public static Class getExtractor(String type) throws ListenerException {
+        for (TimingExtractorEnum e : TimingExtractorEnum.values()) {
+            if (StringUtil.equals(type, e.getType())) {
+                return e.getClazz();
+            }
+        }
+        throw new ListenerException(String.format("TimingListenerEnum type \"%s\" does not exist.", type));
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public Class getClazz() {
+        return clazz;
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/mysql/DqlMysqlExtractor.java

@@ -0,0 +1,22 @@
+package org.dbsyncer.listener.mysql;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/28 22:02
+ */
+public class DqlMysqlExtractor extends MysqlExtractor {
+
+    @Override
+    public void start() {
+        super.postProcessDqlBeforeInitialization();
+        super.start();
+    }
+
+    @Override
+    public void sendChangedEvent(RowChangedEvent event) {
+        super.sendDqlChangedEvent(event);
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/mysql/MysqlExtractor.java

@@ -6,7 +6,7 @@ import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
-import org.dbsyncer.listener.AbstractExtractor;
+import org.dbsyncer.listener.AbstractDatabaseExtractor;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.config.Host;
 import org.slf4j.Logger;
@@ -28,7 +28,7 @@ import static java.util.regex.Pattern.compile;
  * @Author AE86
  * @Date 2020-05-12 21:14
  */
-public class MysqlExtractor extends AbstractExtractor {
+public class MysqlExtractor extends AbstractDatabaseExtractor {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -76,6 +76,11 @@ public class MysqlExtractor extends AbstractExtractor {
         }
     }
 
+    @Override
+    protected void sendChangedEvent(RowChangedEvent event) {
+        changedEvent(event);
+    }
+
     private void run() throws Exception {
         final DatabaseConfig config = (DatabaseConfig) connectorConfig;
         if (StringUtil.isBlank(config.getUrl())) {
@@ -239,7 +244,7 @@ public class MysqlExtractor extends AbstractExtractor {
                     data.getRows().forEach(m -> {
                         List<Object> before = Stream.of(m.getKey()).collect(Collectors.toList());
                         List<Object> after = Stream.of(m.getValue()).collect(Collectors.toList());
-                        changedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_UPDATE, before, after));
+                        sendChangedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_UPDATE, before, after));
                     });
                 }
                 return;
@@ -250,7 +255,7 @@ public class MysqlExtractor extends AbstractExtractor {
                 if (isFilterTable(data.getTableId())) {
                     data.getRows().forEach(m -> {
                         List<Object> after = Stream.of(m).collect(Collectors.toList());
-                        changedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, after));
+                        sendChangedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, after));
                     });
                 }
                 return;
@@ -261,7 +266,7 @@ public class MysqlExtractor extends AbstractExtractor {
                 if (isFilterTable(data.getTableId())) {
                     data.getRows().forEach(m -> {
                         List<Object> before = Stream.of(m).collect(Collectors.toList());
-                        changedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_DELETE, before, Collections.EMPTY_LIST));
+                        sendChangedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_DELETE, before, Collections.EMPTY_LIST));
                     });
                 }
                 return;

dbsyncer-listener/src/main/java/org/dbsyncer/listener/oracle/DqlOracleExtractor.java

@@ -0,0 +1,22 @@
+package org.dbsyncer.listener.oracle;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/29 22:44
+ */
+public class DqlOracleExtractor extends OracleExtractor {
+
+    @Override
+    public void start() {
+        super.postProcessDqlBeforeInitialization();
+        super.start();
+    }
+
+    @Override
+    public void changedEvent(RowChangedEvent event) {
+        super.sendDqlChangedEvent(event);
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/oracle/OracleExtractor.java

@@ -1,7 +1,8 @@
 package org.dbsyncer.listener.oracle;
 
+import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.listener.AbstractExtractor;
+import org.dbsyncer.listener.AbstractDatabaseExtractor;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.oracle.dcn.DBChangeNotification;
 import org.slf4j.Logger;
@@ -12,7 +13,7 @@ import org.slf4j.LoggerFactory;
  * @Author AE86
  * @Date 2020-05-12 21:14
  */
-public class OracleExtractor extends AbstractExtractor {
+public class OracleExtractor extends AbstractDatabaseExtractor {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -27,7 +28,7 @@ public class OracleExtractor extends AbstractExtractor {
             String url = config.getUrl();
             client = new DBChangeNotification(username, password, url);
             client.setFilterTable(filterTable);
-            client.addRowEventListener((e) -> changedEvent(e));
+            client.addRowEventListener((e) -> sendChangedEvent(e));
             client.start();
         } catch (Exception e) {
             logger.error("启动失败:{}", e.getMessage());
@@ -42,4 +43,9 @@ public class OracleExtractor extends AbstractExtractor {
         }
     }
 
+    @Override
+    protected void sendChangedEvent(RowChangedEvent event) {
+        changedEvent(event);
+    }
+
 }

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/DqlPostgreSQLExtractor.java

@@ -0,0 +1,22 @@
+package org.dbsyncer.listener.postgresql;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/29 22:44
+ */
+public class DqlPostgreSQLExtractor extends PostgreSQLExtractor {
+
+    @Override
+    public void start() {
+        super.postProcessDqlBeforeInitialization();
+        super.start();
+    }
+
+    @Override
+    public void changedEvent(RowChangedEvent event) {
+        super.sendDqlChangedEvent(event);
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/PostgreSQLExtractor.java

@@ -1,11 +1,12 @@
 package org.dbsyncer.listener.postgresql;
 
+import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.util.BooleanUtil;
 import org.dbsyncer.common.util.RandomUtil;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.connector.util.DatabaseUtil;
-import org.dbsyncer.listener.AbstractExtractor;
+import org.dbsyncer.listener.AbstractDatabaseExtractor;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.postgresql.enums.MessageDecoderEnum;
 import org.postgresql.PGConnection;
@@ -35,7 +36,7 @@ import java.util.concurrent.locks.ReentrantLock;
  * @version 1.0.0
  * @date 2022/4/10 22:36
  */
-public class PostgreSQLExtractor extends AbstractExtractor {
+public class PostgreSQLExtractor extends AbstractDatabaseExtractor {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -129,6 +130,11 @@ public class PostgreSQLExtractor extends AbstractExtractor {
         }
     }
 
+    @Override
+    protected void sendChangedEvent(RowChangedEvent event) {
+        changedEvent(event);
+    }
+
     private void connect() throws SQLException {
         Properties props = new Properties();
         PGProperty.USER.set(props, config.getUsername());
@@ -278,7 +284,7 @@ public class PostgreSQLExtractor extends AbstractExtractor {
 
                     flushLsn(lsn);
                     // process decoder
-                    changedEvent(messageDecoder.processMessage(msg));
+                    sendChangedEvent(messageDecoder.processMessage(msg));
 
                     // feedback
                     stream.setAppliedLSN(lsn);

dbsyncer-listener/src/main/java/org/dbsyncer/listener/sqlserver/DqlSqlServerExtractor.java

@@ -0,0 +1,22 @@
+package org.dbsyncer.listener.sqlserver;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/5/22 22:56
+ */
+public class DqlSqlServerExtractor extends SqlServerExtractor {
+
+    @Override
+    public void start() {
+        super.postProcessDqlBeforeInitialization();
+        super.start();
+    }
+
+    @Override
+    public void changedEvent(RowChangedEvent event) {
+        super.sendDqlChangedEvent(event);
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/sqlserver/SqlServerExtractor.java

@@ -7,7 +7,7 @@ import org.dbsyncer.common.util.RandomUtil;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.database.DatabaseConnectorMapper;
-import org.dbsyncer.listener.AbstractExtractor;
+import org.dbsyncer.listener.AbstractDatabaseExtractor;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.enums.TableOperationEnum;
 import org.slf4j.Logger;
@@ -27,7 +27,7 @@ import java.util.concurrent.locks.ReentrantLock;
  * @Author AE86
  * @Date 2021-06-18 01:20
  */
-public class SqlServerExtractor extends AbstractExtractor {
+public class SqlServerExtractor extends AbstractDatabaseExtractor {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
@@ -49,8 +49,8 @@ public class SqlServerExtractor extends AbstractExtractor {
     private static final int OFFSET_COLUMNS = 4;
     private final Lock connectLock = new ReentrantLock();
     private volatile boolean connected;
-    private static Set<String> tables;
-    private static Set<SqlServerChangeTable> changeTables;
+    private Set<String> tables;
+    private Set<SqlServerChangeTable> changeTables;
     private DatabaseConnectorMapper connectorMapper;
     private Worker worker;
     private Lsn lastLsn;
@@ -102,6 +102,11 @@ public class SqlServerExtractor extends AbstractExtractor {
         }
     }
 
+    @Override
+    protected void sendChangedEvent(RowChangedEvent event) {
+        changedEvent(event);
+    }
+
     private void close(AutoCloseable closeable) {
         if (null != closeable) {
             try {
@@ -245,17 +250,17 @@ public class SqlServerExtractor extends AbstractExtractor {
         for (CDCEvent event : list) {
             int code = event.getCode();
             if (TableOperationEnum.isUpdateAfter(code)) {
-                changedEvent(new RowChangedEvent(event.getTableName(), ConnectorConstant.OPERTION_UPDATE, Collections.EMPTY_LIST, event.getRow()));
+                sendChangedEvent(new RowChangedEvent(event.getTableName(), ConnectorConstant.OPERTION_UPDATE, Collections.EMPTY_LIST, event.getRow()));
                 continue;
             }
 
             if (TableOperationEnum.isInsert(code)) {
-                changedEvent(new RowChangedEvent(event.getTableName(), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, event.getRow()));
+                sendChangedEvent(new RowChangedEvent(event.getTableName(), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, event.getRow()));
                 continue;
             }
 
             if (TableOperationEnum.isDelete(code)) {
-                changedEvent(new RowChangedEvent(event.getTableName(), ConnectorConstant.OPERTION_DELETE, event.getRow(), Collections.EMPTY_LIST));
+                sendChangedEvent(new RowChangedEvent(event.getTableName(), ConnectorConstant.OPERTION_DELETE, event.getRow(), Collections.EMPTY_LIST));
             }
         }
     }

dbsyncer-manager/src/main/java/org/dbsyncer/manager/puller/IncrementPuller.java

@@ -142,7 +142,7 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
 
         // 默认定时抽取
         if (ListenerTypeEnum.isTiming(listenerType)) {
-            AbstractQuartzExtractor extractor = listener.getExtractor(ListenerTypeEnum.TIMING.getType(), connectorConfig.getConnectorType(), AbstractQuartzExtractor.class);
+            AbstractQuartzExtractor extractor = listener.getExtractor(ListenerTypeEnum.TIMING, connectorConfig.getConnectorType(), AbstractQuartzExtractor.class);
             List<Map<String, String>> commands = list.stream().map(t -> t.getCommand()).collect(Collectors.toList());
             extractor.setCommands(commands);
             setExtractorConfig(extractor, connectorConfig, listenerConfig, meta.getMap(), new QuartzListener(mapping, list));
@@ -151,9 +151,9 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
 
         // 基于日志抽取
         if (ListenerTypeEnum.isLog(listenerType)) {
-            AbstractExtractor extractor = listener.getExtractor(ListenerTypeEnum.LOG.getType(), connectorConfig.getConnectorType(), AbstractExtractor.class);
-            LogListener logListener = new LogListener(mapping, list, extractor);
+            AbstractExtractor extractor = listener.getExtractor(ListenerTypeEnum.LOG, connectorConfig.getConnectorType(), AbstractExtractor.class);
             Set<String> filterTable = new HashSet<>();
+            LogListener logListener = new LogListener(mapping, list, extractor);
             logListener.getTablePicker().forEach((k, fieldPickers) -> filterTable.add(k));
             extractor.setFilterTable(filterTable);
             setExtractorConfig(extractor, connectorConfig, listenerConfig, meta.getMap(), logListener);
@@ -302,7 +302,7 @@ public class IncrementPuller extends AbstractPuller implements ScheduledTaskJob
             // 处理过程有异常向上抛
             List<FieldPicker> pickers = tablePicker.get(rowChangedEvent.getSourceTableName());
             if (!CollectionUtils.isEmpty(pickers)) {
-                pickers.parallelStream().forEach(picker -> {
+                pickers.forEach(picker -> {
                     final Map<String, Object> before = picker.getColumns(rowChangedEvent.getBeforeData());
                     final Map<String, Object> after = picker.getColumns(rowChangedEvent.getAfterData());
                     if (picker.filter(StringUtil.equals(ConnectorConstant.OPERTION_DELETE, rowChangedEvent.getEvent()) ? before : after)) {

dbsyncer-manager/src/main/java/org/dbsyncer/manager/template/AbstractTemplate.java

@@ -23,4 +23,4 @@ public abstract class AbstractTemplate {
         return null != strategy ? strategy : GroupStrategyEnum.DEFAULT;
     }
 
-}
+}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/template/impl/OperationTemplate.java

@@ -164,10 +164,6 @@ public final class OperationTemplate extends AbstractTemplate {
             index.remove(e);
         }
 
-        public List<String> subList(int fromIndex, int toIndex) {
-            return index.subList(fromIndex, toIndex);
-        }
-
         public int size() {
             return index.size();
         }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/ParserFactory.java

@@ -139,9 +139,8 @@ public class ParserFactory implements Parser {
 
     @Override
     public Map<String, String> getCommand(Mapping mapping, TableGroup tableGroup) {
-        ConnectorConfig connectorConfig = getConnectorConfig(mapping.getSourceConnectorId());
-        String sType = connectorConfig.getConnectorType();
-        String tType = getConnectorConfig(mapping.getTargetConnectorId()).getConnectorType();
+        ConnectorConfig sConnConfig = getConnectorConfig(mapping.getSourceConnectorId());
+        ConnectorConfig tConnConfig = getConnectorConfig(mapping.getTargetConnectorId());
         Table sourceTable = tableGroup.getSourceTable();
         Table targetTable = tableGroup.getTargetTable();
         Table sTable = new Table(sourceTable.getName(), sourceTable.getType(), new ArrayList<>());
@@ -157,8 +156,8 @@ public class ParserFactory implements Parser {
                 }
             });
         }
-        final CommandConfig sourceConfig = new CommandConfig(sType, sTable, sourceTable, tableGroup.getFilter(), connectorConfig);
-        final CommandConfig targetConfig = new CommandConfig(tType, tTable, targetTable);
+        final CommandConfig sourceConfig = new CommandConfig(sConnConfig.getConnectorType(), sTable, sourceTable, sConnConfig, tableGroup.getFilter());
+        final CommandConfig targetConfig = new CommandConfig(tConnConfig.getConnectorType(), tTable, targetTable, tConnConfig);
         // 获取连接器同步参数
         Map<String, String> command = connectorFactory.getCommand(sourceConfig, targetConfig);
         return command;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/AbstractBufferActuator.java

@@ -85,16 +85,14 @@ public abstract class AbstractBufferActuator<Request, Response> implements Buffe
 
     @Override
     public void offer(BufferRequest request) {
-        int size = 0;
         if (running) {
             temp.offer((Request) request);
-            size = temp.size();
         } else {
             buffer.offer((Request) request);
-            size = buffer.size();
         }
 
         // TODO 临时解决方案：生产大于消费问题，限制生产速度
+        int size = temp.size() + buffer.size();
         if (size >= CAPACITY) {
             try {
                 TimeUnit.SECONDS.sleep(30);

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/AbstractFlushStrategy.java

@@ -3,6 +3,7 @@ package org.dbsyncer.parser.flush;
 import org.dbsyncer.cache.CacheService;
 import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.parser.model.Meta;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.util.Assert;
@@ -14,6 +15,8 @@ import org.springframework.util.Assert;
  */
 public abstract class AbstractFlushStrategy implements FlushStrategy {
 
+    private static final int MAX_ERROR_LENGTH = 1000;
+
     @Autowired
     private FlushService flushService;
 
@@ -34,10 +37,11 @@ public abstract class AbstractFlushStrategy implements FlushStrategy {
         refreshTotal(metaId, result);
 
         if (!CollectionUtils.isEmpty(result.getFailData())) {
-            flushService.asyncWrite(metaId, event, false, result.getFailData(), result.getError().toString());
+            final String error = StringUtil.substring(result.getError().toString(), 0, MAX_ERROR_LENGTH);
+            flushService.write(metaId, event, false, result.getFailData(), error);
         }
         if (!CollectionUtils.isEmpty(result.getSuccessData())) {
-            flushService.asyncWrite(metaId, event, true, result.getSuccessData(), "");
+            flushService.write(metaId, event, true, result.getSuccessData(), "");
         }
     }
 

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/FlushService.java

@@ -24,6 +24,5 @@ public interface FlushService {
      * @param success
      * @param data
      */
-    @Async("taskExecutor")
-    void asyncWrite(String metaId, String event, boolean success, List<Map> data, String error);
+    void write(String metaId, String event, boolean success, List<Map> data, String error);
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/FlushServiceImpl.java

@@ -2,11 +2,10 @@ package org.dbsyncer.parser.flush.impl;
 
 import com.alibaba.fastjson.JSONException;
 import org.dbsyncer.common.util.JsonUtil;
-import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.parser.flush.BufferActuator;
 import org.dbsyncer.parser.flush.FlushService;
 import org.dbsyncer.parser.flush.model.StorageRequest;
-import org.dbsyncer.storage.SnowflakeIdWorker;
+import org.dbsyncer.common.snowflake.SnowflakeIdWorker;
 import org.dbsyncer.storage.StorageService;
 import org.dbsyncer.storage.constant.ConfigConstant;
 import org.dbsyncer.storage.enums.StorageDataStatusEnum;
@@ -20,7 +19,6 @@ import java.time.Instant;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.stream.Collectors;
 
 /**
  * 持久化
@@ -36,8 +34,6 @@ public class FlushServiceImpl implements FlushService {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
-    private static final int MAX_ERROR_LENGTH = 1000;
-
     @Autowired
     private StorageService storageService;
 
@@ -58,25 +54,23 @@ public class FlushServiceImpl implements FlushService {
     }
 
     @Override
-    public void asyncWrite(String metaId, String event, boolean success, List<Map> data, String error) {
+    public void write(String metaId, String event, boolean success, List<Map> data, String error) {
         long now = Instant.now().toEpochMilli();
-        List<Map> list = data.parallelStream().map(r -> {
-            Map<String, Object> params = new HashMap();
-            params.put(ConfigConstant.CONFIG_MODEL_ID, String.valueOf(snowflakeIdWorker.nextId()));
-            params.put(ConfigConstant.DATA_SUCCESS, success ? StorageDataStatusEnum.SUCCESS.getValue() : StorageDataStatusEnum.FAIL.getValue());
-            params.put(ConfigConstant.DATA_EVENT, event);
-            params.put(ConfigConstant.DATA_ERROR, StringUtil.substring(error, 0, MAX_ERROR_LENGTH));
+        data.forEach(r -> {
+            Map<String, Object> row = new HashMap();
+            row.put(ConfigConstant.CONFIG_MODEL_ID, String.valueOf(snowflakeIdWorker.nextId()));
+            row.put(ConfigConstant.DATA_SUCCESS, success ? StorageDataStatusEnum.SUCCESS.getValue() : StorageDataStatusEnum.FAIL.getValue());
+            row.put(ConfigConstant.DATA_EVENT, event);
+            row.put(ConfigConstant.DATA_ERROR, error);
             try {
-                params.put(ConfigConstant.CONFIG_MODEL_JSON, JsonUtil.objToJson(r));
+                row.put(ConfigConstant.CONFIG_MODEL_JSON, JsonUtil.objToJson(r));
             } catch (JSONException e) {
                 logger.warn("可能存在Blob或inputStream大文件类型, 无法序列化:{}", r);
-                params.put(ConfigConstant.CONFIG_MODEL_JSON, r.toString());
+                row.put(ConfigConstant.CONFIG_MODEL_JSON, r.toString());
             }
-            params.put(ConfigConstant.CONFIG_MODEL_CREATE_TIME, now);
-            return params;
-        }).collect(Collectors.toList());
-
-        storageBufferActuator.offer(new StorageRequest(metaId, list));
+            row.put(ConfigConstant.CONFIG_MODEL_CREATE_TIME, now);
+            storageBufferActuator.offer(new StorageRequest(metaId, row));
+        });
     }
 
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/impl/StorageBufferActuator.java

@@ -38,7 +38,7 @@ public class StorageBufferActuator extends AbstractBufferActuator<StorageRequest
     @Override
     protected void partition(StorageRequest request, StorageResponse response) {
         response.setMetaId(request.getMetaId());
-        response.getDataList().addAll(request.getList());
+        response.getDataList().add(request.getRow());
     }
 
     @Override

dbsyncer-parser/src/main/java/org/dbsyncer/parser/flush/model/StorageRequest.java

@@ -2,7 +2,6 @@ package org.dbsyncer.parser.flush.model;
 
 import org.dbsyncer.parser.flush.BufferRequest;
 
-import java.util.List;
 import java.util.Map;
 
 /**
@@ -14,18 +13,18 @@ public class StorageRequest implements BufferRequest {
 
     private String metaId;
 
-    private List<Map> list;
+    private Map row;
 
-    public StorageRequest(String metaId, List<Map> list) {
+    public StorageRequest(String metaId, Map row) {
         this.metaId = metaId;
-        this.list = list;
+        this.row = row;
     }
 
     public String getMetaId() {
         return metaId;
     }
 
-    public List<Map> getList() {
-        return list;
+    public Map getRow() {
+        return row;
     }
 }

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

@@ -237,4 +237,35 @@ public interface LogType {
         }
     }
 
+    /**
+     * 缓存配置信息7
+     */
+    enum CacheLog implements LogType {
+        IMPORT("70", "导入配置"),
+        EXPORT("71", "导出配置");
+
+        private String type;
+        private String message;
+
+        CacheLog(String type, String message) {
+            this.type = type;
+            this.message = message;
+        }
+
+        @Override
+        public String getName() {
+            return "插件";
+        }
+
+        @Override
+        public String getType() {
+            return type;
+        }
+
+        @Override
+        public String getMessage() {
+            return message;
+        }
+    }
+
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/util/ConvertUtil.java

@@ -20,10 +20,7 @@ public abstract class ConvertUtil {
      */
     public static void convert(List<Convert> convert, List<Map> data) {
         if (!CollectionUtils.isEmpty(convert) && !CollectionUtils.isEmpty(data)) {
-            // 并行流计算
-            data.parallelStream().forEach(row -> {
-                convert(convert, row);
-            });
+            data.forEach(row -> convert(convert, row));
         }
     }
 

dbsyncer-plugin/src/main/java/org/dbsyncer/plugin/service/DemoConvertServiceImpl.java

@@ -1,9 +1,10 @@
 package org.dbsyncer.plugin.service;
 
+import org.dbsyncer.common.config.AppConfig;
 import org.dbsyncer.common.spi.ConvertService;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Value;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
@@ -14,11 +15,8 @@ public class DemoConvertServiceImpl implements ConvertService {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
-    /**
-     * 版本号
-     */
-    @Value(value = "${info.app.version}")
-    private String version;
+    @Autowired
+    private AppConfig appConfig;
 
     @Override
     public void convert(List<Map> source, List<Map> target) {
@@ -31,7 +29,7 @@ public class DemoConvertServiceImpl implements ConvertService {
 
     @Override
     public String getVersion() {
-        return version;
+        return appConfig.getVersion();
     }
 
     public String getName() {

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

@@ -135,7 +135,7 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
     @Override
     public void insertData(StorageEnum type, String collection, List<Map> list) throws IOException {
         createShardIfNotExist(collection);
-        List<Document> docs = list.parallelStream().map(r -> ParamsUtil.convertData2Doc(r)).collect(Collectors.toList());
+        List<Document> docs = list.stream().map(r -> ParamsUtil.convertData2Doc(r)).collect(Collectors.toList());
         map.get(collection).insertBatch(docs);
     }
 

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

@@ -374,13 +374,13 @@ public class MysqlStorageServiceImpl extends AbstractStorageService {
         // 开启高亮
         if (!CollectionUtils.isEmpty(list) && query.isEnableHighLightSearch()) {
             List<Param> highLight = query.getParams().stream().filter(p -> p.isHighlighter()).collect(Collectors.toList());
-            list.parallelStream().forEach(row -> {
+            list.forEach(row ->
                 highLight.forEach(p -> {
                     String text = String.valueOf(row.get(p.getKey()));
                     String replacement = new StringBuilder("<span style='color:red'>").append(p.getValue()).append("</span>").toString();
                     row.put(p.getKey(), StringUtil.replace(text, p.getValue(), replacement));
-                });
-            });
+                })
+            );
         }
     }
 

dbsyncer-web/src/main/java/org/dbsyncer/web/config/AppConfig.java → dbsyncer-web/src/main/java/org/dbsyncer/web/config/OpenApiConfig.java

@@ -12,13 +12,12 @@ import org.springframework.util.unit.DataSize;
 import javax.servlet.MultipartConfigElement;
 
 /**
- *
  * @author AE86
  * @version 1.0.0
  * @date 2019/9/10 23:07
  */
 @Configuration
-public class AppConfig {
+public class OpenApiConfig {
 
     /**
      * 发布服务接口

dbsyncer-web/src/main/java/org/dbsyncer/web/controller/config/ConfigController.java

@@ -0,0 +1,116 @@
+package org.dbsyncer.web.controller.config;
+
+import org.apache.commons.io.IOUtils;
+import org.dbsyncer.biz.ConfigService;
+import org.dbsyncer.biz.vo.RestResult;
+import org.dbsyncer.cache.CacheService;
+import org.dbsyncer.common.config.AppConfig;
+import org.dbsyncer.common.snowflake.SnowflakeIdWorker;
+import org.dbsyncer.common.util.JsonUtil;
+import org.dbsyncer.parser.logger.LogService;
+import org.dbsyncer.parser.logger.LogType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Controller;
+import org.springframework.ui.ModelMap;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.multipart.MultipartFile;
+
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.charset.Charset;
+
+@Controller
+@RequestMapping("/config")
+public class ConfigController {
+
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    private ConfigService configService;
+
+    @Autowired
+    private CacheService cacheService;
+
+    @Autowired
+    private LogService logService;
+
+    @Autowired
+    private AppConfig appConfig;
+
+    @Autowired
+    private SnowflakeIdWorker snowflakeIdWorker;
+
+    @RequestMapping("")
+    public String index(ModelMap model) {
+        model.put("config", configService.getConfigModelAll());
+        model.put("fileSize", JsonUtil.objToJson(cacheService.getAll()).getBytes(Charset.defaultCharset()).length);
+        return "config/config";
+    }
+
+    @PostMapping(value = "/getAll")
+    @ResponseBody
+    public RestResult getAll() {
+        try {
+            return RestResult.restSuccess("ok");
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e.getClass());
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
+    @PostMapping(value = "/upload")
+    @ResponseBody
+    public RestResult upload(MultipartFile[] files) {
+        try {
+            if (files != null && files.length > 0) {
+                MultipartFile file = null;
+                for (int i = 0; i < files.length; i++) {
+                    file = files[i];
+                    if (file != null) {
+                        String filename = file.getOriginalFilename();
+                        // TODO checkFileSuffix(filename);
+                        logger.info(filename);
+                        String msg = String.format("导入配置文件%s。", filename);
+                        logger.info(msg);
+                        logService.log(LogType.CacheLog.IMPORT, msg);
+                    }
+                }
+            }
+            return RestResult.restSuccess("ok");
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e.getClass());
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
+    @GetMapping("/download")
+    public void download(HttpServletResponse response) {
+        String fileName = String.format("%s-%s-%s.json", appConfig.getName(), appConfig.getVersion(), snowflakeIdWorker.nextId());
+        response.setHeader("content-type", "application/octet-stream");
+        response.setHeader("Content-Disposition", String.format("attachment; filename=%s", fileName));
+        response.setContentType("application/octet-stream");
+        OutputStream outputStream = null;
+        try {
+            outputStream = response.getOutputStream();
+            String cache = JsonUtil.objToJson(cacheService.getAll());
+            byte[] bytes = cache.getBytes(Charset.defaultCharset());
+            int length = bytes.length;
+            String msg = String.format("导出配置文件%s，大小%dKB。", fileName, (length / 1024));
+            logger.info(msg);
+            logService.log(LogType.CacheLog.EXPORT, msg);
+            outputStream.write(bytes, 0, length);
+            outputStream.flush();
+        } catch (IOException e) {
+            logger.error(e.getMessage());
+        } finally {
+            IOUtils.closeQuietly(outputStream);
+        }
+    }
+
+}

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

@@ -4,8 +4,8 @@ import org.dbsyncer.biz.ConnectorService;
 import org.dbsyncer.biz.MappingService;
 import org.dbsyncer.biz.vo.RestResult;
 import org.dbsyncer.biz.vo.VersionVo;
+import org.dbsyncer.common.config.AppConfig;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Controller;
 import org.springframework.ui.ModelMap;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -13,6 +13,7 @@ import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.ResponseBody;
 
 import javax.servlet.http.HttpServletRequest;
+import java.io.UnsupportedEncodingException;
 
 @Controller
 @RequestMapping("/index")
@@ -24,11 +25,8 @@ public class IndexController {
     @Autowired
     private MappingService mappingService;
 
-    @Value(value = "${info.app.name}")
-    private String appName;
-
-    @Value(value = "${info.app.copyright}")
-    private String appCopyRight;
+    @Autowired
+    private AppConfig appConfig;
 
     @GetMapping("")
     public String index(HttpServletRequest request, ModelMap model) {
@@ -39,8 +37,8 @@ public class IndexController {
 
     @GetMapping("/version.json")
     @ResponseBody
-    public RestResult version() {
-        return RestResult.restSuccess(new VersionVo(appName, appCopyRight));
+    public RestResult version() throws UnsupportedEncodingException {
+        return RestResult.restSuccess(new VersionVo(appConfig.getName(), appConfig.getCopyright()));
     }
 
 }

dbsyncer-web/src/main/java/org/dbsyncer/web/controller/upload/UploadController.java → dbsyncer-web/src/main/java/org/dbsyncer/web/controller/plugin/PluginController.java

@@ -1,14 +1,14 @@
-package org.dbsyncer.web.controller.upload;
+package org.dbsyncer.web.controller.plugin;
 
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.io.IOUtils;
 import org.dbsyncer.biz.PluginService;
 import org.dbsyncer.biz.vo.RestResult;
+import org.dbsyncer.common.config.AppConfig;
 import org.dbsyncer.common.util.JsonUtil;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Controller;
 import org.springframework.ui.ModelMap;
 import org.springframework.web.bind.annotation.GetMapping;
@@ -21,27 +21,24 @@ import javax.servlet.http.HttpServletResponse;
 import java.io.*;
 
 @Controller
-@RequestMapping("/upload")
-public class UploadController {
+@RequestMapping("/plugin")
+public class PluginController {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
-    /**
-     * 版本号
-     */
-    @Value(value = "${info.app.version}")
-    private String version;
+    @Autowired
+    private PluginService pluginService;
+
+    @Autowired
+    private AppConfig appConfig;
 
     @RequestMapping("")
     public String index(ModelMap model) {
         model.put("plugins", pluginService.getPluginAll());
-        model.put("version", version);
-        return "upload/upload";
+        model.put("version", appConfig.getVersion());
+        return "plugin/plugin";
     }
 
-    @Autowired
-    private PluginService pluginService;
-
     @PostMapping(value = "/upload")
     @ResponseBody
     public RestResult upload(MultipartFile[] files) {
@@ -71,9 +68,9 @@ public class UploadController {
 
     @GetMapping("/download")
     public void download(HttpServletResponse response) {
-        String fileName = String.format("dbsyncer-common-%s.jar", version);
+        String fileName = String.format("dbsyncer-common-%s.jar", appConfig.getVersion());
         File file = new File(pluginService.getLibraryPath() + fileName);
-        if(!file.exists()){
+        if (!file.exists()) {
             write(response, RestResult.restFail("Could not find file", 404));
             return;
         }

dbsyncer-web/src/main/resources/application.properties

@@ -5,18 +5,15 @@ dbsyncer.web.login.username=admin
 dbsyncer.web.login.password=0DPiKuNIrrVmD8IUCuw1hQxNqZc=
 server.servlet.session.timeout=1800
 server.servlet.context-path=/
+#dbsyncer.common.worker.id=1
 #dbsyncer.web.thread.pool.coreSize=8
-#dbsyncer.web.thread.pool.maxSize=128
-#dbsyncer.web.thread.pool.queueCapacity=2000
-#storage
-dbsyncer.storage.id=1
+#dbsyncer.web.thread.pool.maxSize=64
+#dbsyncer.web.thread.pool.queueCapacity=1000
 #dbsyncer.storage.support.mysql.enabled=true
 #dbsyncer.storage.support.mysql.config.url=jdbc:mysql://127.0.0.1:3306/dbsyncer?rewriteBatchedStatements=true&seUnicode=true&characterEncoding=UTF8&serverTimezone=Asia/Shanghai&useSSL=false&verifyServerCertificate=false
 #dbsyncer.storage.support.mysql.config.username=root
 #dbsyncer.storage.support.mysql.config.password=123
-#parser
 #dbsyncer.parser.flush.full.enabled=true
-
 #monitor
 management.endpoints.web.base-path=/app
 management.endpoints.web.exposure.include=*
@@ -24,7 +21,7 @@ management.endpoint.health.show-details=always
 management.health.elasticsearch.enabled=false
 info.app.name=DBSyncer
 info.app.version=1.1.8-Beta
-info.app.copyright=&copy;2021 ${info.app.name}(${info.app.version})<footer>Designed By <a href='https://gitee.com/ghi/dbsyncer' target='_blank' >AE86</a></footer>
+info.app.copyright=&copy;2022 ${info.app.name}(${info.app.version})<footer>Designed By <a href='https://gitee.com/ghi/dbsyncer' target='_blank' >AE86</a></footer>
 
 #All < Trace < Debug < Info < Warn < Error < Fatal < OFF
 logging.level.root=info

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

@@ -0,0 +1,84 @@
+<!DOCTYPE html>
+<html xmlns="http://www.w3.org/1999/xhtml"
+      xmlns:th="http://www.thymeleaf.org" lang="zh-CN">
+
+<div class="container">
+    <form id="configEditForm" class="form-horizontal" role="form">
+        <div class="row text-center">
+            <div class="page-header">
+                <h3>配置管理</h3>
+            </div>
+        </div>
+
+        <!-- 操作 -->
+        <div class="row">
+            <div class="col-md-7">
+                <table class="table table-hover">
+                    <caption>配置列表([[${config?.size()} ?: 0]])，<a onClick="downLoad()" href="javascript:;"
+                                                                 th:title="'下载文件大小'+${#numbers.formatDecimal((fileSize / 1024),0 ,2)}+'KB'">下载</a>
+                    </caption>
+                    <thead>
+                    <tr>
+                        <th>ID</th>
+                        <th>类型</th>
+                        <th>名称</th>
+                        <th>创建时间</th>
+                        <th>修改时间</th>
+                    </tr>
+                    </thead>
+                    <tbody id="pluginList">
+                    <tr th:id="${c?.name}" th:each="c,state : ${config}">
+                        <td th:text="${c?.id}"/>
+                        <td th:text="${c?.type}"/>
+                        <td th:text="${c?.name}"/>
+                        <td th:text="${#dates.format(c?.createTime, 'yyyy-MM-dd HH:mm:ss')}"/>
+                        <td th:text="${#dates.format(c?.updateTime, 'yyyy-MM-dd HH:mm:ss')}"/>
+                    </tr>
+                    </tbody>
+                </table>
+            </div>
+
+            <!-- 配置列表 -->
+            <div class="col-md-5">
+                <form id="uploadForm" class="form-horizontal" role="form">
+                    <div class="page-header">
+                        <h3>上传配置 <small>只支持 "json" 的文件扩展名.</small></h3>
+                    </div>
+
+                    <div class="form-group">
+                        <div class="file-loading">
+                            <input id="fileConfig" type="file" name="files" multiple="multiple"/>
+                        </div>
+                    </div>
+                </form>
+            </div>
+
+        </div>
+    </form>
+</div>
+
+<script type="text/javascript">
+    $("#fileConfig").fileinput({
+        theme: 'fas',
+        language: 'zh',
+        uploadUrl: $basePath + '/config/upload',
+        enctype: 'multipart/form-data',
+        removeFromPreviewOnError: true, //当选择的文件不符合规则时，例如不是指定后缀文件、大小超出配置等，选择的文件不会出现在预览框中，只会显示错误信息
+        allowedFileExtensions: ['json'],
+        minFileCount: 0, //每次多次上载允许的最小文件数。如果设置为0，则表示文件数是可选的
+        maxFileCount: 5, //表示允许同时上传的最大文件个数 如果设置为0，则表示允许的文件数不受限制
+        showPreview: true,
+        showUpload: true,//不展示上传按钮
+        validateInitialCount: true,//是否在验证minFileCount和包含初始预览文件计数（服务器上载文件）maxFileCount
+    }).on("fileuploaded", function (event, data, previewId, index) {
+        if (!data.response.success) {
+            bootGrowl(data.response.resultValue, "danger");
+        }
+        doLoader("/config");
+    });
+
+    function downLoad() {
+        window.open($basePath + "/config/download");
+    }
+</script>
+</html>

dbsyncer-web/src/main/resources/public/connector/addDqlMysql.html

@@ -10,22 +10,34 @@
         </div>
         <label class="col-sm-2 control-label">密码 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4 ">
-            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require" th:value="${connector?.config?.password}"/>
+            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require"
+                   th:value="${connector?.config?.password}"/>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea id="sql" name="sql" class="form-control" maxlength="8192" dbsyncer-valid="require" rows="10" th:text="${connector?.config?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
+            <textarea id="sql" name="sql" class="form-control" maxlength="8192" dbsyncer-valid="require" rows="10"
+                      th:text="${connector?.config?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
+        </div>
+    </div>
+    <div class="form-group">
+        <div class="col-sm-6"></div>
+        <div class="col-sm-6 text-right">
+            <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer"
+                                                      title="美化SQL"></span>美化SQL</a>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require" placeholder="USER" th:value="${connector?.config?.table}"/>
+            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="USER" th:value="${connector?.config?.table}"/>
         </div>
-        <div class="col-sm-6 text-right">
-            <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer" title="美化SQL"></span>美化SQL</a>
+        <label class="col-sm-2 control-label">主键 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
         </div>
     </div>
     <div class="form-group">

dbsyncer-web/src/main/resources/public/connector/addDqlOracle.html

@@ -10,22 +10,35 @@
         </div>
         <label class="col-sm-2 control-label">密码 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4 ">
-            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require" th:value="${connector?.config?.password}"/>
+            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require"
+                   th:value="${connector?.config?.password}"/>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea id="sql" name="sql" class="form-control dbsyncer_textarea_resize_none" maxlength="8192" dbsyncer-valid="require" rows="10" th:text="${connector?.config?.sql}?:'SELECT T1.*,ROWIDTOCHAR(ROWID) as RID FROM &quot;USER&quot; T1'"></textarea>
+            <textarea id="sql" name="sql" class="form-control dbsyncer_textarea_resize_none" maxlength="8192"
+                      dbsyncer-valid="require" rows="10"
+                      th:text="${connector?.config?.sql}?:'SELECT T1.*,ROWIDTOCHAR(ROWID) as RID FROM &quot;USER&quot; T1'"></textarea>
+        </div>
+    </div>
+    <div class="form-group">
+        <div class="col-sm-6"></div>
+        <div class="col-sm-6 text-right">
+            <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer"
+                                                      title="美化SQL"></span>美化SQL</a>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require" placeholder="USER" th:value="${connector?.config?.table}"/>
+            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="USER" th:value="${connector?.config?.table}"/>
         </div>
-        <div class="col-sm-6 text-right">
-            <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer" title="美化SQL"></span>美化SQL</a>
+        <label class="col-sm-2 control-label">主键 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
         </div>
     </div>
     <div class="form-group">

dbsyncer-web/src/main/resources/public/connector/addDqlPostgreSQL.html

@@ -18,20 +18,30 @@
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea class="form-control dbsyncer_textarea_resize_none" maxlength="8192" dbsyncer-valid="require" id="sql" name="sql" rows="10" th:text="${connector?.config?.sql}?:'SELECT T1.* FROM &quot;USER&quot; T1'"></textarea>
+            <textarea class="form-control dbsyncer_textarea_resize_none" maxlength="8192" dbsyncer-valid="require"
+                      id="sql" name="sql" rows="10"
+                      th:text="${connector?.config?.sql}?:'SELECT T1.* FROM &quot;USER&quot; T1'"></textarea>
         </div>
     </div>
     <div class="form-group">
-        <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
-        <div class="col-sm-4">
-            <input class="form-control" dbsyncer-valid="require" maxlength="32" name="table" placeholder="USER"
-                   th:value="${connector?.config?.table}" type="text"/>
-        </div>
+        <div class="col-sm-6"></div>
         <div class="col-sm-6 text-right">
             <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer"
                                                       title="美化SQL"></span>美化SQL</a>
         </div>
     </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="USER" th:value="${connector?.config?.table}"/>
+        </div>
+        <label class="col-sm-2 control-label">主键 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
+        </div>
+    </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">

dbsyncer-web/src/main/resources/public/connector/addDqlSqlServer.html

@@ -10,22 +10,35 @@
         </div>
         <label class="col-sm-2 control-label">密码 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4 ">
-            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require" th:value="${connector?.config?.password}"/>
+            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require"
+                   th:value="${connector?.config?.password}"/>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">SQL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea id="sql" name="sql" class="form-control dbsyncer_textarea_resize_none" maxlength="1024" dbsyncer-valid="require" rows="10" th:text="${connector?.config?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
+            <textarea id="sql" name="sql" class="form-control dbsyncer_textarea_resize_none" maxlength="1024"
+                      dbsyncer-valid="require" rows="10"
+                      th:text="${connector?.config?.sql}?:'SELECT T1.* FROM USER T1'"></textarea>
+        </div>
+    </div>
+    <div class="form-group">
+        <div class="col-sm-6"></div>
+        <div class="col-sm-6 text-right">
+            <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer"
+                                                      title="美化SQL"></span>美化SQL</a>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">主表 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require" placeholder="USER" th:value="${connector?.config?.table}"/>
+            <input class="form-control" name="table" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="USER" th:value="${connector?.config?.table}"/>
         </div>
-        <div class="col-sm-6 text-right">
-            <a href="javascript:beautifySql();"><span class="fa fa-magic fa-1x fa-flip-horizontal dbsyncer_pointer" title="美化SQL"></span>美化SQL</a>
+        <label class="col-sm-2 control-label">主键 <strong class="driverVerifcateRequired">*</strong></label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="ID" th:value="${connector?.config?.primaryKey}"/>
         </div>
     </div>
     <div class="form-group">

dbsyncer-web/src/main/resources/public/connector/addMysql.html

@@ -10,21 +10,38 @@
         </div>
         <label class="col-sm-2 control-label">密码 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4 ">
-            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require" th:value="${connector?.config?.password}"/>
+            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require"
+                   th:value="${connector?.config?.password}"/>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea name="url" class="form-control dbsyncer_textarea_resize_none" maxlength="1024" dbsyncer-valid="require" rows="5" th:text="${connector?.config?.url} ?: 'jdbc:mysql://127.0.0.1:3306/test?rewriteBatchedStatements=true&useUnicode=true&amp;characterEncoding=UTF8&amp;serverTimezone=Asia/Shanghai&amp;useSSL=false&amp;verifyServerCertificate=false&amp;autoReconnect=true&amp;failOverReadOnly=false'"></textarea>
+            <textarea name="url" class="form-control dbsyncer_textarea_resize_none" maxlength="1024"
+                      dbsyncer-valid="require" rows="5"
+                      th:text="${connector?.config?.url} ?: 'jdbc:mysql://127.0.0.1:3306/test?rewriteBatchedStatements=true&useUnicode=true&amp;characterEncoding=UTF8&amp;serverTimezone=Asia/Shanghai&amp;useSSL=false&amp;verifyServerCertificate=false&amp;autoReconnect=true&amp;failOverReadOnly=false'"></textarea>
         </div>
     </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">主键</label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" placeholder="ID"
+                   th:value="${connector?.config?.primaryKey}"/>
+        </div>
+        <div class="col-sm-6"></div>
+    </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">驱动 </label>
         <div class="col-sm-10">
             <select class="form-control select-control" name="driverClassName">
-                <option value="com.mysql.jdbc.Driver" th:selected="${connector?.config?.driverClassName eq 'com.mysql.jdbc.Driver'}">com.mysql.jdbc.Driver</option>
-                <option value="com.mysql.cj.jdbc.Driver" th:selected="${connector?.config?.driverClassName eq 'com.mysql.cj.jdbc.Driver'}">com.mysql.cj.jdbc.Driver</option>
+                <option value="com.mysql.jdbc.Driver"
+                        th:selected="${connector?.config?.driverClassName eq 'com.mysql.jdbc.Driver'}">
+                    com.mysql.jdbc.Driver
+                </option>
+                <option value="com.mysql.cj.jdbc.Driver"
+                        th:selected="${connector?.config?.driverClassName eq 'com.mysql.cj.jdbc.Driver'}">
+                    com.mysql.cj.jdbc.Driver
+                </option>
             </select>
         </div>
     </div>

dbsyncer-web/src/main/resources/public/connector/addOracle.html

@@ -10,19 +10,31 @@
         </div>
         <label class="col-sm-2 control-label">密码 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4 ">
-            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require" th:value="${connector?.config?.password}"/>
+            <input class="form-control" name="password" type="password" maxlength="32" dbsyncer-valid="require"
+                   th:value="${connector?.config?.password}"/>
         </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">URL <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-10">
-            <textarea name="url" class="form-control dbsyncer_textarea_resize_none" maxlength="1024" dbsyncer-valid="require" rows="5" th:text="${connector?.config?.url} ?: 'jdbc:oracle:thin:@127.0.0.1:1521:ORCL'"></textarea>
+            <textarea name="url" class="form-control dbsyncer_textarea_resize_none" maxlength="1024"
+                      dbsyncer-valid="require" rows="5"
+                      th:text="${connector?.config?.url} ?: 'jdbc:oracle:thin:@127.0.0.1:1521:ORCL'"></textarea>
         </div>
     </div>
+    <div class="form-group">
+        <label class="col-sm-2 control-label">主键</label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" placeholder="ID"
+                   th:value="${connector?.config?.primaryKey}"/>
+        </div>
+        <div class="col-sm-6"></div>
+    </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">驱动 </label>
         <div class="col-sm-10">
-            <input class="form-control" readonly="true" name="driverClassName" type="text" th:value="${connector?.config?.driverClassName} ?: 'oracle.jdbc.OracleDriver'" />
+            <input class="form-control" readonly="true" name="driverClassName" type="text"
+                   th:value="${connector?.config?.driverClassName} ?: 'oracle.jdbc.OracleDriver'"/>
         </div>
     </div>
 </div>

dbsyncer-web/src/main/resources/public/connector/addPostgreSQL.html

@@ -29,7 +29,11 @@
             <input class="form-control" dbsyncer-valid="require" maxlength="32" name="schema" placeholder="public"
                    th:value="${connector?.config?.schema} ?: 'public'" type="text"/>
         </div>
-        <div class="col-sm-6"></div>
+        <label class="col-sm-2 control-label">主键</label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" placeholder="ID"
+                   th:value="${connector?.config?.primaryKey}"/>
+        </div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">删除Slot <i aria-hidden="true" class="fa fa-question-circle fa_gray" title="增量同步，停止驱动自动删除Slot"></i></label>

dbsyncer-web/src/main/resources/public/connector/addSqlServer.html

@@ -22,9 +22,14 @@
     <div class="form-group">
         <label class="col-sm-2 control-label">架构名 <strong class="driverVerifcateRequired">*</strong></label>
         <div class="col-sm-4">
-            <input class="form-control" name="schema" type="text" maxlength="32" dbsyncer-valid="require" placeholder="dbo" th:value="${connector?.config?.schema} ?: 'dbo'"/>
+            <input class="form-control" name="schema" type="text" maxlength="32" dbsyncer-valid="require"
+                   placeholder="dbo" th:value="${connector?.config?.schema} ?: 'dbo'"/>
+        </div>
+        <label class="col-sm-2 control-label">主键</label>
+        <div class="col-sm-4">
+            <input class="form-control" name="primaryKey" type="text" maxlength="32" placeholder="ID"
+                   th:value="${connector?.config?.primaryKey}"/>
         </div>
-        <div class="col-sm-6"></div>
     </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">驱动 </label>

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

@@ -51,11 +51,6 @@
         <div th:replace="mapping/editIncrementQuartz :: content"></div>
     </div>
 
-    <!-- 日志配置 -->
-    <div th:id="mappingIncrementStrategyDQLConfig" class="hidden">
-        <div th:replace="mapping/editIncrementDQL :: content"></div>
-    </div>
-
 </div>
 
 </html>

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

@@ -1,20 +0,0 @@
-<!DOCTYPE html>
-<html xmlns="http://www.w3.org/1999/xhtml"
-      xmlns:th="http://www.thymeleaf.org" lang="zh-CN">
-
-<div th:fragment="content">
-    <!-- 针对DQL的连接配置 -->
-    <div class="form-group" th:if="${#strings.startsWith(mapping?.sourceConnector?.config?.connectorType,'Dql')}">
-        <div class="row">
-            <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 name="incrementStrategyLogTableLabel" type="text" dbsyncer-valid="require" class="form-control" th:value="${mapping?.listener?.tableLabel}?:'T1'">
-                </div>
-            </div>
-            <div class="col-md-8"></div>
-        </div>
-    </div>
-</div>
-
-</html>

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

@@ -23,7 +23,8 @@
                                     </div>
                                     <div class="col-md-8">
                                         <p class="driver_break_word">[[${mapping?.sourceConnector?.name}]]</p>
-                                        <p>数据源表：[[${tableGroup?.sourceTable?.name}]]</p>
+                                        <p class="driver_hidden_word" th:title="${tableGroup?.sourceTable?.name}">
+                                            数据源表：[[${tableGroup?.sourceTable?.name}]]</p>
                                     </div>
                                 </div>
                             </div>

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

@@ -9,14 +9,19 @@
         </div>
         <div>
             <ul id="menu" class="nav navbar-nav">
-                <li class="active"><a href="javascript:void(0);" url="/index"><span class="fa fa-tachometer"></span>驱动管理</a></li>
-                <li><a href="javascript:void(0);" url="/monitor" ><span class="fa fa-line-chart"></span>监控</a></li>
-                <li><a href="javascript:void(0);" url="/upload" ><span class="fa fa-cloud-upload" aria-hidden="true"></span>上传插件</a></li>
+                <li class="active"><a href="javascript:void(0);" url="/index"><span
+                        class="fa fa-tachometer"></span>驱动</a>
+                </li>
+                <li><a href="javascript:void(0);" url="/monitor"><span class="fa fa-line-chart"></span>监控</a></li>
+                <li><a href="javascript:void(0);" url="/plugin"><span class="fa fa-puzzle-piece"
+                                                                      aria-hidden="true"></span>插件</a></li>
                 <li class="dropdown">
-                    <a href="javascript:void(0);" class="dropdown-toggle" data-toggle="dropdown">参数<b class="caret"></b></a>
+                    <a href="javascript:void(0);" class="dropdown-toggle" data-toggle="dropdown">配置<b class="caret"></b></a>
                     <ul class="dropdown-menu">
                         <li><a href="javascript:void(0);" url="/system"><span class="fa fa-cog"></span>系统参数</a></li>
                         <li><a href="javascript:void(0);" url="/pwd"><span class="fa fa-lock"></span>修改密码</a></li>
+                        <li><a href="javascript:void(0);" url="/config"><span class="fa fa-file"
+                                                                              aria-hidden="true"></span>配置管理</a></li>
                     </ul>
                 </li>
             </ul>

dbsyncer-web/src/main/resources/public/upload/upload.html → dbsyncer-web/src/main/resources/public/plugin/plugin.html

@@ -136,24 +136,24 @@ public class MyPlugin implements ConvertService{
     $("#filePlugin").fileinput({
         theme: 'fas',
         language: 'zh',
-        uploadUrl: $basePath + '/upload/upload',
+        uploadUrl: $basePath + '/plugin/plugin',
         enctype: 'multipart/form-data',
-        removeFromPreviewOnError:true, //当选择的文件不符合规则时，例如不是指定后缀文件、大小超出配置等，选择的文件不会出现在预览框中，只会显示错误信息
+        removeFromPreviewOnError: true, //当选择的文件不符合规则时，例如不是指定后缀文件、大小超出配置等，选择的文件不会出现在预览框中，只会显示错误信息
         allowedFileExtensions: ['jar'],
         minFileCount: 0, //每次多次上载允许的最小文件数。如果设置为0，则表示文件数是可选的
         maxFileCount: 5, //表示允许同时上传的最大文件个数 如果设置为0，则表示允许的文件数不受限制
         showPreview: true,
-        showUpload:true,//不展示上传按钮
-        validateInitialCount:true,//是否在验证minFileCount和包含初始预览文件计数（服务器上载文件）maxFileCount
+        showUpload: true,//不展示上传按钮
+        validateInitialCount: true,//是否在验证minFileCount和包含初始预览文件计数（服务器上载文件）maxFileCount
     }).on("fileuploaded", function(event, data, previewId, index) {
         if (!data.response.success) {
             bootGrowl(data.response.resultValue, "danger");
         }
-        doLoader("/upload");
+        doLoader("/plugin");
     });
 
     function downLoad(){
-        window.open($basePath + "/upload/download");
+        window.open($basePath + "/plugin/download");
     }
 </script>
 </html>

dbsyncer-web/src/main/resources/static/css/common.css

@@ -16,8 +16,21 @@
 /**
  * 强制单词换行
  */
-.driver_break_word { word-break: break-all;word-wrap:break-word;white-space:normal}
-.fa_gray {color: gray}
+.driver_break_word {
+    word-break: break-all;
+    word-wrap: break-word;
+    white-space: normal
+}
+
+.driver_hidden_word {
+    white-space: nowrap;
+    overflow: hidden;
+    text-overflow: ellipsis
+}
+
+.fa_gray {
+    color: gray
+}
 .fa_blueviolet {color: blueviolet}
 .dbsyncer_btn-info { background-color: #E7EDF8;}
 .dbsyncer_btn-info.active, .dbsyncer_btn-info.focus, .dbsyncer_btn-info:active, .dbsyncer_btn-info:focus, .dbsyncer_btn-info:hover, .open>.dropdown-toggle.dbsyncer_btn-info {background-color: #D4DDED;}