!59 merge

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

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

@@ -1,5 +1,6 @@
 package org.dbsyncer.biz.checker.impl.connector;
 
+import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.springframework.stereotype.Component;
 
@@ -16,5 +17,8 @@ public class PostgreSQLConfigChecker extends AbstractDataBaseConfigChecker {
     public void modify(DatabaseConfig connectorConfig, Map<String, String> params) {
         super.modify(connectorConfig, params);
         super.modifySchema(connectorConfig, params);
+
+        connectorConfig.getProperties().put("dropSlotOnClose", StringUtil.isNotBlank(params.get("dropSlotOnClose")) ? "true" : "false");
+        connectorConfig.getProperties().put("pluginName", params.get("pluginName"));
     }
 }

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

@@ -37,8 +37,8 @@ public class TableGroupServiceImpl extends BaseServiceImpl implements TableGroup
         assertRunning(manager.getMapping(mappingId));
 
         // table1, table2
-        String[] sourceTableArray = StringUtil.split(params.get("sourceTable"), ",");
-        String[] targetTableArray = StringUtil.split(params.get("targetTable"), ",");
+        String[] sourceTableArray = StringUtil.split(params.get("sourceTable"), "|");
+        String[] targetTableArray = StringUtil.split(params.get("targetTable"), "|");
         int tableSize = sourceTableArray.length;
         Assert.isTrue(tableSize == targetTableArray.length, "数据源表和目标源表关系必须为一组");
 

dbsyncer-common/src/main/java/org/dbsyncer/common/util/DateFormatUtil.java

@@ -1,10 +1,9 @@
 package org.dbsyncer.common.util;
 
-import java.time.Instant;
-import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
+import java.time.*;
+import java.time.format.*;
+import java.time.temporal.ChronoField;
+import java.time.temporal.TemporalAccessor;
 import java.util.Date;
 
 public abstract class DateFormatUtil {
@@ -21,6 +20,50 @@ public abstract class DateFormatUtil {
      * HH:mm:ss
      */
     public static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm:ss");
+
+    private static final DateTimeFormatter TIME_TZ_FORMAT = new DateTimeFormatterBuilder()
+            .append(DateTimeFormatter.ISO_LOCAL_TIME)
+            .appendOffset("+HH:mm", "")
+            .toFormatter();
+
+    private static final DateTimeFormatter NON_ISO_LOCAL_DATE = new DateTimeFormatterBuilder()
+            .appendValue(ChronoField.YEAR_OF_ERA, 4, 10, SignStyle.NEVER)
+            .appendLiteral('-')
+            .appendValue(ChronoField.MONTH_OF_YEAR, 2)
+            .appendLiteral('-')
+            .appendValue(ChronoField.DAY_OF_MONTH, 2)
+            .toFormatter();
+
+    private static final DateTimeFormatter TS_TZ_FORMAT = new DateTimeFormatterBuilder()
+            .append(NON_ISO_LOCAL_DATE)
+            .appendLiteral(' ')
+            .append(DateTimeFormatter.ISO_LOCAL_TIME)
+            .appendOffset("+HH:mm", "")
+            .optionalStart()
+            .appendLiteral(" ")
+            .appendText(ChronoField.ERA, TextStyle.SHORT)
+            .optionalEnd()
+            .toFormatter();
+    private static final DateTimeFormatter TS_TZ_WITH_SECONDS_FORMAT = new DateTimeFormatterBuilder()
+            .append(NON_ISO_LOCAL_DATE)
+            .appendLiteral(' ')
+            .append(DateTimeFormatter.ISO_LOCAL_TIME)
+            .appendOffset("+HH:MM:SS", "")
+            .optionalStart()
+            .appendLiteral(" ")
+            .appendText(ChronoField.ERA, TextStyle.SHORT)
+            .optionalEnd()
+            .toFormatter();
+    private static final DateTimeFormatter TS_FORMAT = new DateTimeFormatterBuilder()
+            .append(NON_ISO_LOCAL_DATE)
+            .appendLiteral(' ')
+            .append(DateTimeFormatter.ISO_LOCAL_TIME)
+            .optionalStart()
+            .appendLiteral(" ")
+            .appendText(ChronoField.ERA, TextStyle.SHORT)
+            .optionalEnd()
+            .toFormatter();
+
     private static ZoneId zoneId = ZoneId.systemDefault();
 
     public static String getCurrentTime() {
@@ -41,4 +84,29 @@ public abstract class DateFormatUtil {
         return Date.from(instant);
     }
 
+    public static LocalDate stringToLocalDate(String s) {
+        return LocalDate.parse(s, DATE_FORMATTER);
+    }
+
+    public static LocalTime stringToLocalTime(String s) {
+        return LocalTime.parse(s, CHINESE_STANDARD_TIME_FORMATTER);
+    }
+
+    public static OffsetTime timeWithTimeZone(String s) {
+        return OffsetTime.parse(s, TIME_TZ_FORMAT).withOffsetSameInstant(ZoneOffset.UTC);
+    }
+
+    public static OffsetDateTime timestampWithTimeZoneToOffsetDateTime(String s) {
+        TemporalAccessor parsedTimestamp;
+        try {
+            parsedTimestamp = TS_TZ_FORMAT.parse(s);
+        } catch (DateTimeParseException e) {
+            parsedTimestamp = TS_TZ_WITH_SECONDS_FORMAT.parse(s);
+        }
+        return OffsetDateTime.from(parsedTimestamp).withOffsetSameInstant(ZoneOffset.UTC);
+    }
+
+    public static Instant timestampToInstant(String s) {
+        return LocalDateTime.from(TS_FORMAT.parse(s)).toInstant(ZoneOffset.UTC);
+    }
 }

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

@@ -1,5 +1,8 @@
 package org.dbsyncer.connector.config;
 
+import java.util.LinkedHashMap;
+import java.util.Map;
+
 /**
  * @author AE86
  * @ClassName: DatabaseConfig
@@ -29,6 +32,17 @@ public class DatabaseConfig extends ConnectorConfig {
     // 构架名
     private String schema;
 
+    // 参数配置
+    private Map<String, String> properties = new LinkedHashMap<>();
+
+    public String getProperty(String key){
+        return properties.get(key);
+    }
+
+    public String getProperty(String key, String defaultValue){
+        return properties.containsKey(key) ? properties.get(key) : defaultValue;
+    }
+
     public String getDriverClassName() {
         return driverClassName;
     }
@@ -84,4 +98,12 @@ public class DatabaseConfig extends ConnectorConfig {
     public void setSchema(String schema) {
         this.schema = schema;
     }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+
+    public void setProperties(Map<String, String> properties) {
+        this.properties = properties;
+    }
 }

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

@@ -54,7 +54,7 @@ public abstract class AbstractDatabaseConnector extends AbstractConnector
 
     @Override
     public String getConnectorMapperCacheKey(DatabaseConfig config) {
-        return String.format("%s-%s", config.getUrl(), config.getUsername());
+        return String.format("%s-%s-%s", config.getConnectorType(), config.getUrl(), config.getUsername());
     }
 
     @Override

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

@@ -30,7 +30,7 @@ public class DatabaseConnectorMapper implements ConnectorMapper<DatabaseConfig,
             throw e;
         } catch (Exception e) {
             logger.error(e.getMessage());
-            throw new ConnectorException(e.getMessage());
+            throw new ConnectorException(e.getMessage(), e.getCause());
         } finally {
             DatabaseUtil.close(connection);
         }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/database/setter/TinyintSetter.java

@@ -21,6 +21,11 @@ public class TinyintSetter extends AbstractSetter<Integer> {
             ps.setShort(i, v);
             return;
         }
+        if (val instanceof Boolean) {
+            Boolean b = (Boolean) val;
+            ps.setBoolean(i, b);
+            return;
+        }
         throw new ConnectorException(String.format("TinyintSetter can not find type [%s], val [%s]", type, val));
     }
 }

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

@@ -80,7 +80,7 @@ public final class ESConnector extends AbstractConnector implements Connector<ES
 
     @Override
     public String getConnectorMapperCacheKey(ESConfig config) {
-        return String.format("%s-%s-%s-%s", config.getUrl(), config.getIndex(), config.getType(), config.getUsername());
+        return String.format("%s-%s-%s-%s-%s", config.getConnectorType(), config.getUrl(), config.getIndex(), config.getType(), config.getUsername());
     }
 
     @Override

dbsyncer-connector/src/main/java/org/dbsyncer/connector/kafka/KafkaConnector.java

@@ -41,7 +41,7 @@ public class KafkaConnector extends AbstractConnector implements Connector<Kafka
 
     @Override
     public String getConnectorMapperCacheKey(KafkaConfig config) {
-        return String.format("%s-%s-%s", config.getBootstrapServers(), config.getTopic(), config.getGroupId());
+        return String.format("%s-%s-%s-%s", config.getConnectorType(), config.getBootstrapServers(), config.getTopic(), config.getGroupId());
     }
 
     @Override

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

@@ -3,7 +3,6 @@ package org.dbsyncer.listener;
 import org.dbsyncer.common.event.Event;
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.common.scheduled.ScheduledTaskService;
-import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.listener.config.ListenerConfig;
@@ -15,6 +14,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.Executor;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @version 1.0.0
@@ -31,64 +31,54 @@ public abstract class AbstractExtractor implements Extractor {
     protected ListenerConfig listenerConfig;
     protected Map<String, String> snapshot;
     protected Set<String> filterTable;
-    private List<Event> watcher;
+    private List<Event> watcher = new CopyOnWriteArrayList<>();
 
     @Override
     public void addListener(Event event) {
         if (null != event) {
-            if (null == watcher) {
-                watcher = new CopyOnWriteArrayList<>();
-            }
             watcher.add(event);
         }
     }
 
     @Override
     public void clearAllListener() {
-        if (null != watcher) {
-            watcher.clear();
-            watcher = null;
-        }
+        watcher.clear();
     }
 
     @Override
     public void changedEvent(RowChangedEvent event) {
-        if (!CollectionUtils.isEmpty(watcher)) {
-            watcher.forEach(w -> w.changedEvent(event));
+        if(null != event){
+            taskExecutor.execute(() -> watcher.forEach(w -> w.changedEvent(event)));
         }
     }
 
     @Override
     public void flushEvent() {
-        if (!CollectionUtils.isEmpty(watcher)) {
-            watcher.forEach(w -> w.flushEvent(snapshot));
-        }
+        watcher.forEach(w -> w.flushEvent(snapshot));
     }
 
     @Override
     public void forceFlushEvent() {
-        if (!CollectionUtils.isEmpty(watcher)) {
-            logger.info("Force flush:{}", snapshot);
-            watcher.forEach(w -> w.forceFlushEvent(snapshot));
-        }
+        logger.info("Force flush:{}", snapshot);
+        watcher.forEach(w -> w.forceFlushEvent(snapshot));
     }
 
     @Override
     public void errorEvent(Exception e) {
-        if (!CollectionUtils.isEmpty(watcher)) {
-            watcher.forEach(w -> w.errorEvent(e));
-        }
+        watcher.forEach(w -> w.errorEvent(e));
     }
 
     @Override
     public void interruptException(Exception e) {
-        if (!CollectionUtils.isEmpty(watcher)) {
-            watcher.forEach(w -> w.interruptException(e));
-        }
+        watcher.forEach(w -> w.interruptException(e));
     }
 
-    protected void asyncSendRowChangedEvent(RowChangedEvent event) {
-        taskExecutor.execute(() -> changedEvent(event));
+    protected void sleepInMills(long timeout) {
+        try {
+            TimeUnit.MILLISECONDS.sleep(timeout);
+        } catch (InterruptedException e) {
+            logger.error(e.getMessage());
+        }
     }
 
     public void setTaskExecutor(Executor taskExecutor) {

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

@@ -6,6 +6,7 @@ import org.dbsyncer.listener.ListenerException;
 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;
@@ -34,7 +35,7 @@ public enum ListenerEnum {
     /**
      * log_PostgreSQL
      */
-//    LOG_POSTGRE_SQL(ListenerTypeEnum.LOG.getType() + ConnectorEnum.POSTGRE_SQL.getType(), PostgreSQLExtractor.class),
+    LOG_POSTGRE_SQL(ListenerTypeEnum.LOG.getType() + ConnectorEnum.POSTGRE_SQL.getType(), PostgreSQLExtractor.class),
     /**
      * log_Kafka
      */

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

@@ -233,37 +233,37 @@ public class MysqlExtractor extends AbstractExtractor {
             }
 
             if (EventType.isUpdate(header.getEventType())) {
+                refresh(header);
                 UpdateRowsEventData data = event.getData();
                 if (isFilterTable(data.getTableId())) {
                     data.getRows().forEach(m -> {
                         List<Object> before = Stream.of(m.getKey()).collect(Collectors.toList());
                         List<Object> after = Stream.of(m.getValue()).collect(Collectors.toList());
-                        asyncSendRowChangedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_UPDATE, before, after));
+                        changedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_UPDATE, before, after));
                     });
                 }
-                refresh(header);
                 return;
             }
             if (EventType.isWrite(header.getEventType())) {
+                refresh(header);
                 WriteRowsEventData data = event.getData();
                 if (isFilterTable(data.getTableId())) {
                     data.getRows().forEach(m -> {
                         List<Object> after = Stream.of(m).collect(Collectors.toList());
-                        asyncSendRowChangedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, after));
+                        changedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, after));
                     });
                 }
-                refresh(header);
                 return;
             }
             if (EventType.isDelete(header.getEventType())) {
+                refresh(header);
                 DeleteRowsEventData data = event.getData();
                 if (isFilterTable(data.getTableId())) {
                     data.getRows().forEach(m -> {
                         List<Object> before = Stream.of(m).collect(Collectors.toList());
-                        asyncSendRowChangedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_DELETE, before, Collections.EMPTY_LIST));
+                        changedEvent(new RowChangedEvent(getTableName(data.getTableId()), ConnectorConstant.OPERTION_DELETE, before, Collections.EMPTY_LIST));
                     });
                 }
-                refresh(header);
                 return;
             }
 

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

@@ -27,7 +27,7 @@ public class OracleExtractor extends AbstractExtractor {
             String url = config.getUrl();
             client = new DBChangeNotification(username, password, url);
             client.setFilterTable(filterTable);
-            client.addRowEventListener((e) -> asyncSendRowChangedEvent(e));
+            client.addRowEventListener((e) -> changedEvent(e));
             client.start();
         } catch (Exception e) {
             logger.error("启动失败:{}", e.getMessage());

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

@@ -0,0 +1,31 @@
+package org.dbsyncer.listener.postgresql;
+
+import org.dbsyncer.connector.config.DatabaseConfig;
+import org.postgresql.replication.LogSequenceNumber;
+
+import java.nio.ByteBuffer;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/17 23:04
+ */
+public abstract class AbstractMessageDecoder implements MessageDecoder {
+
+    protected DatabaseConfig config;
+
+    @Override
+    public boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn) {
+        return null == lastReceiveLsn || lastReceiveLsn.asLong() == 0 || startLsn.equals(lastReceiveLsn);
+    }
+
+    @Override
+    public String getSlotName() {
+        return String.format("dbs_slot_%s_%s", config.getSchema(), config.getUsername());
+    }
+
+    @Override
+    public void setConfig(DatabaseConfig config) {
+        this.config = config;
+    }
+}

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

@@ -0,0 +1,29 @@
+package org.dbsyncer.listener.postgresql;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.connector.config.DatabaseConfig;
+import org.postgresql.replication.LogSequenceNumber;
+import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
+
+import java.nio.ByteBuffer;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/17 22:59
+ */
+public interface MessageDecoder {
+
+    boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn);
+
+    RowChangedEvent processMessage(ByteBuffer buffer);
+
+    String getSlotName();
+
+    String getOutputPlugin();
+
+    void withSlotOption(ChainedLogicalStreamBuilder builder);
+
+    void setConfig(DatabaseConfig config);
+
+}

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

@@ -1,18 +1,32 @@
 package org.dbsyncer.listener.postgresql;
 
+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.ListenerException;
+import org.dbsyncer.listener.postgresql.enums.MessageDecoderEnum;
 import org.postgresql.PGConnection;
+import org.postgresql.PGProperty;
+import org.postgresql.replication.LogSequenceNumber;
 import org.postgresql.replication.PGReplicationStream;
+import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
+import org.postgresql.util.PSQLException;
+import org.postgresql.util.PSQLState;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.util.Assert;
 
+import java.nio.ByteBuffer;
 import java.sql.Connection;
+import java.sql.DriverManager;
 import java.sql.SQLException;
+import java.time.Instant;
 import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -23,19 +37,26 @@ import java.util.concurrent.locks.ReentrantLock;
  */
 public class PostgreSQLExtractor extends AbstractExtractor {
 
-    private static final String GET_VALIDATION = "SELECT 1";
-    private static final String GET_ROLE = "SELECT r.rolcanlogin AS rolcanlogin, r.rolreplication AS rolreplication, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rds_superuser') AS BOOL) IS TRUE AS aws_superuser, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsadmin') AS BOOL) IS TRUE AS aws_admin, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsrepladmin') AS BOOL) IS TRUE AS aws_repladmin FROM pg_roles r WHERE r.rolname = current_user";
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    private static final String GET_SLOT = "select count(1) from pg_replication_slots where database = ? and slot_name = ? and plugin = ?";
+    private static final String GET_ROLE = "SELECT r.rolcanlogin AS login, r.rolreplication AS replication, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rds_superuser') AS BOOL) IS TRUE AS superuser, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsadmin') AS BOOL) IS TRUE AS admin, CAST(array_position(ARRAY(SELECT b.rolname FROM pg_catalog.pg_auth_members m JOIN pg_catalog.pg_roles b ON (m.roleid = b.oid) WHERE m.member = r.oid), 'rdsrepladmin') AS BOOL) IS TRUE AS rep_admin FROM pg_roles r WHERE r.rolname = current_user";
+    private static final String GET_DATABASE = "SELECT current_database()";
     private static final String GET_WAL_LEVEL = "SHOW WAL_LEVEL";
     private static final String DEFAULT_WAL_LEVEL = "logical";
-    private static final String DEFAULT_SLOT_NAME = "DBSYNCER_SLOT";
-    private static final String DEFAULT_PLUGIN_NAME = "wal2json";
-    private final Logger logger = LoggerFactory.getLogger(getClass());
+    private static final String PLUGIN_NAME = "pluginName";
+    private static final String LSN_POSITION = "position";
+    private static final String DROP_SLOT_ON_CLOSE = "dropSlotOnClose";
     private final Lock connectLock = new ReentrantLock();
     private volatile boolean connected;
-    private Connection connection;
-    private PGReplicationStream stream;
     private DatabaseConfig config;
     private DatabaseConnectorMapper connectorMapper;
+    private Connection connection;
+    private PGReplicationStream stream;
+    private boolean dropSlotOnClose;
+    private MessageDecoder messageDecoder;
+    private Worker worker;
+    private LogSequenceNumber startLsn;
 
     @Override
     public void start() {
@@ -46,10 +67,8 @@ public class PostgreSQLExtractor extends AbstractExtractor {
                 return;
             }
 
-            connect();
-
-            connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_VALIDATION, Integer.class));
-            logger.info("Successfully tested connection for {} with user '{}'", config.getUrl(), config.getUsername());
+            connectorMapper = (DatabaseConnectorMapper) connectorFactory.connect(connectorConfig);
+            config = connectorMapper.getConfig();
 
             final String walLevel = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_WAL_LEVEL, String.class));
             if (!DEFAULT_WAL_LEVEL.equals(walLevel)) {
@@ -57,59 +76,227 @@ public class PostgreSQLExtractor extends AbstractExtractor {
             }
 
             final boolean hasAuth = connectorMapper.execute(databaseTemplate -> {
-                Map rs = databaseTemplate.queryForObject(GET_ROLE, Map.class);
-                Boolean login = (Boolean) rs.getOrDefault("rolcanlogin", false);
-                Boolean replication = (Boolean) rs.getOrDefault("rolreplication", false);
-                Boolean superuser = (Boolean) rs.getOrDefault("aws_superuser", false);
-                Boolean admin = (Boolean) rs.getOrDefault("aws_admin", false);
-                Boolean replicationAdmin = (Boolean) rs.getOrDefault("aws_repladmin", false);
-                return login && (replication || superuser || admin || replicationAdmin);
+                Map rs = databaseTemplate.queryForMap(GET_ROLE);
+                Boolean login = (Boolean) rs.getOrDefault("login", false);
+                Boolean replication = (Boolean) rs.getOrDefault("replication", false);
+                Boolean superuser = (Boolean) rs.getOrDefault("superuser", false);
+                Boolean admin = (Boolean) rs.getOrDefault("admin", false);
+                Boolean repAdmin = (Boolean) rs.getOrDefault("rep_admin", false);
+                return login && (replication || superuser || admin || repAdmin);
             });
             if (!hasAuth) {
                 throw new ListenerException(String.format("Postgres roles LOGIN and REPLICATION are not assigned to user: %s", config.getUsername()));
             }
+
+            messageDecoder = MessageDecoderEnum.getMessageDecoder(config.getProperty(PLUGIN_NAME));
+            messageDecoder.setConfig(config);
+            dropSlotOnClose = BooleanUtil.toBoolean(config.getProperty(DROP_SLOT_ON_CLOSE, "true"));
+
+            connect();
             connected = true;
+
+            worker = new Worker();
+            worker.setName(new StringBuilder("wal-parser-").append(config.getUrl()).append("_").append(RandomUtil.nextInt(1, 100)).toString());
+            worker.setDaemon(false);
+            worker.start();
         } catch (Exception e) {
             logger.error("启动失败:{}", e.getMessage());
+            DatabaseUtil.close(stream);
+            DatabaseUtil.close(connection);
             throw new ListenerException(e);
         } finally {
             connectLock.unlock();
-            close();
         }
     }
 
     @Override
     public void close() {
         try {
-            connectLock.lock();
             connected = false;
+            if (null != worker && !worker.isInterrupted()) {
+                worker.interrupt();
+                worker = null;
+            }
             DatabaseUtil.close(stream);
             DatabaseUtil.close(connection);
+            dropReplicationSlot();
         } catch (Exception e) {
             logger.error("关闭失败:{}", e.getMessage());
-        } finally {
-            connectLock.unlock();
         }
     }
 
     private void connect() throws SQLException {
-        if (connectorFactory.isAlive(connectorConfig)) {
-            config = (DatabaseConfig) connectorConfig;
-            connectorMapper = (DatabaseConnectorMapper) connectorFactory.connect(config);
+        Properties props = new Properties();
+        PGProperty.USER.set(props, config.getUsername());
+        PGProperty.PASSWORD.set(props, config.getPassword());
+        // Postgres 9.4发布逻辑复制功能
+        PGProperty.ASSUME_MIN_SERVER_VERSION.set(props, "9.4");
+        PGProperty.REPLICATION.set(props, "database");
+        PGProperty.PREFER_QUERY_MODE.set(props, "simple");
+        connection = DriverManager.getConnection(config.getUrl(), props);
+        Assert.notNull(connection, "Unable to get connection.");
+
+        PGConnection pgConnection = connection.unwrap(PGConnection.class);
+        createReplicationSlot(pgConnection);
+        createReplicationStream(pgConnection);
+
+        sleepInMills(10L);
+    }
+
+    private LogSequenceNumber readLastLsn() throws SQLException {
+        if (!snapshot.containsKey(LSN_POSITION)) {
+            LogSequenceNumber lsn = currentXLogLocation();
+            if (null == lsn || lsn.asLong() == 0) {
+                throw new ListenerException("No maximum LSN recorded in the database");
+            }
+            snapshot.put(LSN_POSITION, lsn.asString());
+        }
 
-            connection = DatabaseUtil.getConnection(config.getDriverClassName(), config.getUrl(), config.getUsername(), config.getPassword());
-            PGConnection replConnection = connection.unwrap(PGConnection.class);
-            replConnection.getReplicationAPI()
+        return LogSequenceNumber.valueOf(snapshot.get(LSN_POSITION));
+    }
+
+    private void createReplicationStream(PGConnection pgConnection) throws SQLException {
+        this.startLsn = readLastLsn();
+        ChainedLogicalStreamBuilder streamBuilder = pgConnection
+                .getReplicationAPI()
+                .replicationStream()
+                .logical()
+                .withSlotName(messageDecoder.getSlotName())
+                .withStartPosition(startLsn)
+                .withStatusInterval(10, TimeUnit.SECONDS);
+
+        messageDecoder.withSlotOption(streamBuilder);
+        this.stream = streamBuilder.start();
+    }
+
+    private void createReplicationSlot(PGConnection pgConnection) throws SQLException {
+        String database = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_DATABASE, String.class));
+        String slotName = messageDecoder.getSlotName();
+        String plugin = messageDecoder.getOutputPlugin();
+        boolean existSlot = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(GET_SLOT, new Object[]{database, slotName, plugin}, Integer.class) > 0);
+        if (!existSlot) {
+            pgConnection.getReplicationAPI()
                     .createReplicationSlot()
                     .logical()
-                    .withSlotName(DEFAULT_SLOT_NAME)
-                    .withOutputPlugin(DEFAULT_PLUGIN_NAME)
+                    .withSlotName(slotName)
+                    .withOutputPlugin(plugin)
                     .make();
-            stream = replConnection.getReplicationAPI()
-                    .replicationStream()
-                    .logical()
-                    .withSlotName(DEFAULT_SLOT_NAME)
-                    .start();
         }
     }
-}
+
+    private void dropReplicationSlot() {
+        if (!dropSlotOnClose) {
+            return;
+        }
+
+        final String slotName = messageDecoder.getSlotName();
+        final int ATTEMPTS = 3;
+        for (int i = 0; i < ATTEMPTS; i++) {
+            try {
+                connectorMapper.execute(databaseTemplate -> {
+                    databaseTemplate.execute(String.format("select pg_drop_replication_slot('%s')", slotName));
+                    return true;
+                });
+                break;
+            } catch (Exception e) {
+                if (e.getCause() instanceof PSQLException) {
+                    PSQLException ex = (PSQLException) e.getCause();
+                    if (PSQLState.OBJECT_IN_USE.getState().equals(ex.getSQLState())) {
+                        if (i < ATTEMPTS - 1) {
+                            logger.debug("Cannot drop replication slot '{}' because it's still in use", slotName);
+                            continue;
+                        }
+                        logger.warn("Cannot drop replication slot '{}' because it's still in use", slotName);
+                        break;
+                    }
+
+                    if (PSQLState.UNDEFINED_OBJECT.getState().equals(ex.getSQLState())) {
+                        logger.debug("Replication slot {} has already been dropped", slotName);
+                        break;
+                    }
+
+                    logger.error("Unexpected error while attempting to drop replication slot", ex);
+                }
+            }
+        }
+    }
+
+    private LogSequenceNumber currentXLogLocation() throws SQLException {
+        int majorVersion = connection.getMetaData().getDatabaseMajorVersion();
+        String sql = majorVersion >= 10 ? "select * from pg_current_wal_lsn()" : "select * from pg_current_xlog_location()";
+        return connectorMapper.execute(databaseTemplate -> LogSequenceNumber.valueOf(databaseTemplate.queryForObject(sql, String.class)));
+    }
+
+    private void recover() {
+        connectLock.lock();
+        try {
+            long s = Instant.now().toEpochMilli();
+            DatabaseUtil.close(stream);
+            DatabaseUtil.close(connection);
+            stream = null;
+            connection = null;
+
+            while (connected) {
+                try {
+                    connect();
+                    break;
+                } catch (Exception e) {
+                    logger.error("Recover streaming occurred error");
+                    DatabaseUtil.close(stream);
+                    DatabaseUtil.close(connection);
+                    sleepInMills(3000L);
+                }
+            }
+            long e = Instant.now().toEpochMilli();
+            logger.info("Recover logical replication success, slot:{}, plugin:{}, cost:{}seconds", messageDecoder.getSlotName(), messageDecoder.getOutputPlugin(), (e - s) / 1000);
+        } finally {
+            connectLock.unlock();
+        }
+    }
+
+    private void flushLsn(LogSequenceNumber lsn) {
+        if (null != lsn && lsn.asLong() > 0) {
+            snapshot.put(LSN_POSITION, lsn.asString());
+        }
+    }
+
+    final class Worker extends Thread {
+
+        @Override
+        public void run() {
+            while (!isInterrupted() && connected) {
+                try {
+                    // non blocking receive message
+                    ByteBuffer msg = stream.readPending();
+
+                    if (msg == null) {
+                        sleepInMills(10L);
+                        continue;
+                    }
+
+                    LogSequenceNumber lsn = stream.getLastReceiveLSN();
+                    if (messageDecoder.skipMessage(msg, startLsn, lsn)) {
+                        continue;
+                    }
+
+                    flushLsn(lsn);
+                    // process decoder
+                    changedEvent(messageDecoder.processMessage(msg));
+                    forceFlushEvent();
+
+                    // feedback
+                    stream.setAppliedLSN(lsn);
+                    stream.setFlushedLSN(lsn);
+                    stream.forceUpdateStatus();
+                } catch (IllegalStateException e) {
+                    logger.error(e.getMessage());
+                } catch (Exception e) {
+                    logger.error(e.getMessage());
+                    recover();
+                }
+            }
+        }
+
+    }
+
+}

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

@@ -0,0 +1,177 @@
+package org.dbsyncer.listener.postgresql.column;
+
+import org.dbsyncer.common.util.DateFormatUtil;
+import org.dbsyncer.listener.ListenerException;
+import org.postgresql.PGStatement;
+import org.postgresql.geometric.*;
+import org.postgresql.util.PGInterval;
+import org.postgresql.util.PGmoney;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.sql.SQLException;
+import java.time.*;
+import java.util.concurrent.TimeUnit;
+
+public abstract class AbstractColumnValue implements ColumnValue {
+
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public LocalDate asLocalDate() {
+        return DateFormatUtil.stringToLocalDate(asString());
+    }
+
+    @Override
+    public Object asTime() {
+        return asString();
+    }
+
+    @Override
+    public Object asLocalTime() {
+        return DateFormatUtil.stringToLocalTime(asString());
+    }
+
+    @Override
+    public OffsetTime asOffsetTimeUtc() {
+        return DateFormatUtil.timeWithTimeZone(asString());
+    }
+
+    @Override
+    public OffsetDateTime asOffsetDateTimeAtUtc() {
+        if ("infinity".equals(asString())) {
+            return OffsetDateTime.ofInstant(toInstantFromMillis(PGStatement.DATE_POSITIVE_INFINITY), ZoneOffset.UTC);
+        } else if ("-infinity".equals(asString())) {
+            return OffsetDateTime.ofInstant(toInstantFromMillis(PGStatement.DATE_NEGATIVE_INFINITY), ZoneOffset.UTC);
+        }
+        return DateFormatUtil.timestampWithTimeZoneToOffsetDateTime(asString());
+    }
+
+    @Override
+    public Instant asInstant() {
+        if ("infinity".equals(asString())) {
+            return toInstantFromMicros(PGStatement.DATE_POSITIVE_INFINITY);
+        } else if ("-infinity".equals(asString())) {
+            return toInstantFromMicros(PGStatement.DATE_NEGATIVE_INFINITY);
+        }
+        return DateFormatUtil.timestampToInstant(asString());
+    }
+
+    @Override
+    public PGbox asBox() {
+        try {
+            return new PGbox(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGcircle asCircle() {
+        try {
+            return new PGcircle(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse circle {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public Object asInterval() {
+        try {
+            return new PGInterval(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGline asLine() {
+        try {
+            return new PGline(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGlseg asLseg() {
+        try {
+            return new PGlseg(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGmoney asMoney() {
+        try {
+            final String value = asString();
+            if (value != null && value.startsWith("-")) {
+                final String negativeMoney = "(" + value.substring(1) + ")";
+                return new PGmoney(negativeMoney);
+            }
+            return new PGmoney(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse money {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGpath asPath() {
+        try {
+            return new PGpath(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGpoint asPoint() {
+        try {
+            return new PGpoint(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public PGpolygon asPolygon() {
+        try {
+            return new PGpolygon(asString());
+        } catch (final SQLException e) {
+            logger.error("Failed to parse point {}, {}", asString(), e);
+            throw new ListenerException(e);
+        }
+    }
+
+    @Override
+    public boolean isArray() {
+        return false;
+    }
+
+    @Override
+    public Object asArray() {
+        return null;
+    }
+
+    private Instant toInstantFromMicros(long microsSinceEpoch) {
+        return Instant.ofEpochSecond(
+                TimeUnit.MICROSECONDS.toSeconds(microsSinceEpoch),
+                TimeUnit.MICROSECONDS.toNanos(microsSinceEpoch % TimeUnit.SECONDS.toMicros(1)));
+    }
+
+    private Instant toInstantFromMillis(long millisecondSinceEpoch) {
+        return Instant.ofEpochSecond(
+                TimeUnit.MILLISECONDS.toSeconds(millisecondSinceEpoch),
+                TimeUnit.MILLISECONDS.toNanos(millisecondSinceEpoch % TimeUnit.SECONDS.toMillis(1)));
+    }
+
+}

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

@@ -0,0 +1,71 @@
+package org.dbsyncer.listener.postgresql.column;
+
+import org.postgresql.geometric.*;
+import org.postgresql.util.PGmoney;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.OffsetDateTime;
+import java.time.OffsetTime;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/22 22:39
+ * @see org.postgresql.jdbc.TypeInfoCache
+ */
+public interface ColumnValue {
+
+    boolean isNull();
+
+    String asString();
+
+    Boolean asBoolean();
+
+    Integer asInteger();
+
+    Long asLong();
+
+    Float asFloat();
+
+    Double asDouble();
+
+    Object asDecimal();
+
+    LocalDate asLocalDate();
+
+    OffsetDateTime asOffsetDateTimeAtUtc();
+
+    Instant asInstant();
+
+    Object asTime();
+
+    Object asLocalTime();
+
+    OffsetTime asOffsetTimeUtc();
+
+    byte[] asByteArray();
+
+    PGbox asBox();
+
+    PGcircle asCircle();
+
+    Object asInterval();
+
+    PGline asLine();
+
+    Object asLseg();
+
+    PGmoney asMoney();
+
+    PGpath asPath();
+
+    PGpoint asPoint();
+
+    PGpolygon asPolygon();
+
+    boolean isArray();
+
+    Object asArray();
+
+}

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

@@ -0,0 +1,158 @@
+package org.dbsyncer.listener.postgresql.column;
+
+import org.postgresql.util.PGmoney;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/23 22:45
+ */
+public class ColumnValueResolver {
+
+    /**
+     * Resolve the value of a {@link ColumnValue}.
+     *
+     * @param type
+     * @param value
+     * @return
+     */
+    public Object resolveValue(String type, ColumnValue value) {
+        if (value.isNull()) {
+            // nulls are null
+            return null;
+        }
+
+        switch (type) {
+            // include all types from https://www.postgresql.org/docs/current/static/datatype.html#DATATYPE-TABLE
+            // plus aliases from the shorter names produced by older wal2json
+            case "boolean":
+            case "bool":
+                return value.asBoolean();
+
+            case "hstore":
+                return value.asString();
+
+            case "integer":
+            case "int":
+            case "int4":
+            case "smallint":
+            case "int2":
+            case "smallserial":
+            case "serial":
+            case "serial2":
+            case "serial4":
+                return value.asInteger();
+
+            case "bigint":
+            case "bigserial":
+            case "int8":
+            case "oid":
+                return value.asLong();
+
+            case "real":
+            case "float4":
+                return value.asFloat();
+
+            case "double precision":
+            case "float8":
+                return value.asDouble();
+
+            case "numeric":
+            case "decimal":
+                return value.asDecimal();
+
+            case "character":
+            case "char":
+            case "character varying":
+            case "varchar":
+            case "bpchar":
+            case "text":
+                return value.asString();
+
+            case "date":
+                return value.asLocalDate();
+
+            case "timestamp with time zone":
+            case "timestamptz":
+                return value.asOffsetDateTimeAtUtc();
+
+            case "timestamp":
+            case "timestamp without time zone":
+                return value.asInstant();
+
+            case "time":
+                return value.asTime();
+
+            case "time without time zone":
+                return value.asLocalTime();
+
+            case "time with time zone":
+            case "timetz":
+                return value.asOffsetTimeUtc();
+
+            case "bytea":
+                return value.asByteArray();
+
+            // these are all PG-specific types and we use the JDBC representations
+            // note that, with the exception of point, no converters for these types are implemented yet,
+            // i.e. those values won't actually be propagated to the outbound message until that's the case
+            case "box":
+                return value.asBox();
+            case "circle":
+                return value.asCircle();
+            case "interval":
+                return value.asInterval();
+            case "line":
+                return value.asLine();
+            case "lseg":
+                return value.asLseg();
+            case "money":
+                final Object v = value.asMoney();
+                return (v instanceof PGmoney) ? ((PGmoney) v).val : v;
+            case "path":
+                return value.asPath();
+            case "point":
+                return value.asPoint();
+            case "polygon":
+                return value.asPolygon();
+
+            // PostGIS types are HexEWKB strings
+            // ValueConverter turns them into the correct types
+            case "geometry":
+            case "geography":
+                return value.asString();
+
+            case "citext":
+            case "bit":
+            case "bit varying":
+            case "varbit":
+            case "json":
+            case "jsonb":
+            case "xml":
+            case "uuid":
+            case "tsrange":
+            case "tstzrange":
+            case "daterange":
+            case "inet":
+            case "cidr":
+            case "macaddr":
+            case "macaddr8":
+            case "int4range":
+            case "numrange":
+            case "int8range":
+                return value.asString();
+
+            // catch-all for other known/builtin PG types
+            // TODO: improve with more specific/useful classes here?
+            case "pg_lsn":
+            case "tsquery":
+            case "tsvector":
+            case "txid_snapshot":
+                // catch-all for unknown (extension module/custom) types
+            default:
+                return null;
+        }
+
+    }
+
+}

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

@@ -0,0 +1,64 @@
+package org.dbsyncer.listener.postgresql.column;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/24 18:22
+ */
+public final class Lexer {
+    private final char[] array;
+    private final int length;
+    private int pos = 0;
+    private String token;
+
+    public Lexer(String input) {
+        this.array = input.toCharArray();
+        this.length = this.array.length;
+    }
+
+    public String token() {
+        return token;
+    }
+
+    public String nextToken(char comma) {
+        if (pos < length) {
+            StringBuilder out = new StringBuilder(16);
+            while (pos < length && array[pos] != comma) {
+                out.append(array[pos]);
+                pos++;
+            }
+            pos++;
+            return token = out.toString();
+        }
+        return token = null;
+    }
+
+    public String nextTokenToQuote() {
+        if (pos < length) {
+            int commaCount = 1;
+            StringBuilder out = new StringBuilder(16);
+            while (!((pos == length - 1 || (array[pos + 1] == ' ' && commaCount % 2 == 1)) && array[pos] == '\'')) {
+                if (array[pos] == '\'') {
+                    commaCount++;
+                }
+                out.append(array[pos]);
+                pos++;
+            }
+            pos++;
+            return token = out.toString();
+        }
+        return token = null;
+    }
+
+    public void skip(int skip) {
+        this.pos += skip;
+    }
+
+    public char current() {
+        return array[pos];
+    }
+
+    public boolean hasNext() {
+        return pos < length;
+    }
+}

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

@@ -0,0 +1,59 @@
+package org.dbsyncer.listener.postgresql.column;
+
+import org.dbsyncer.common.util.StringUtil;
+
+import java.math.BigDecimal;
+
+public final class TestDecodingColumnValue extends AbstractColumnValue {
+
+    private String value;
+
+    public TestDecodingColumnValue(String value) {
+        this.value = value;
+    }
+
+    @Override
+    public boolean isNull() {
+        return value == null;
+    }
+
+    @Override
+    public String asString() {
+        return value;
+    }
+
+    @Override
+    public Boolean asBoolean() {
+        return "t".equalsIgnoreCase(value);
+    }
+
+    @Override
+    public Integer asInteger() {
+        return Integer.valueOf(value);
+    }
+
+    @Override
+    public Long asLong() {
+        return Long.valueOf(value);
+    }
+
+    @Override
+    public Float asFloat() {
+        return Float.valueOf(value);
+    }
+
+    @Override
+    public Double asDouble() {
+        return Double.valueOf(value);
+    }
+
+    @Override
+    public Object asDecimal() {
+        return new BigDecimal(value);
+    }
+
+    @Override
+    public byte[] asByteArray() {
+        return StringUtil.hexStringToByteArray(value.substring(2));
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/decoder/PgOutputMessageDecoder.java

@@ -0,0 +1,80 @@
+package org.dbsyncer.listener.postgresql.decoder;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.listener.postgresql.AbstractMessageDecoder;
+import org.dbsyncer.listener.postgresql.enums.MessageDecoderEnum;
+import org.dbsyncer.listener.postgresql.enums.MessageTypeEnum;
+import org.postgresql.replication.LogSequenceNumber;
+import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.ByteBuffer;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/17 23:00
+ */
+public class PgOutputMessageDecoder extends AbstractMessageDecoder {
+
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Override
+    public boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn) {
+        if (super.skipMessage(buffer, startLsn, lastReceiveLsn)) {
+            return true;
+        }
+        int position = buffer.position();
+        try {
+            MessageTypeEnum type = MessageTypeEnum.getType((char) buffer.get());
+            switch (type) {
+                case BEGIN:
+                case COMMIT:
+                case RELATION:
+                case TRUNCATE:
+                case TYPE:
+                case ORIGIN:
+                case NONE:
+                    return true;
+                default:
+                    // TABLE|INSERT|UPDATE|DELETE
+                    return false;
+            }
+        } finally {
+            buffer.position(position);
+        }
+    }
+
+    @Override
+    public RowChangedEvent processMessage(ByteBuffer buffer) {
+        if (!buffer.hasArray()) {
+            throw new IllegalStateException("Invalid buffer received from PG server during streaming replication");
+        }
+        MessageTypeEnum type = MessageTypeEnum.getType((char) buffer.get());
+        if (MessageTypeEnum.TABLE == type) {
+            int offset = buffer.arrayOffset();
+            byte[] source = buffer.array();
+            return parseMessage(new String(source, offset, (source.length - offset)));
+        }
+        return null;
+    }
+
+    @Override
+    public String getOutputPlugin() {
+        return MessageDecoderEnum.PG_OUTPUT.getType();
+    }
+
+    @Override
+    public void withSlotOption(ChainedLogicalStreamBuilder builder) {
+        builder.withSlotOption("proto_version", 1);
+        builder.withSlotOption("publication_names", String.format("dbs_pub_%s_%s", config.getSchema(), config.getUsername()));
+    }
+
+    private RowChangedEvent parseMessage(String message) {
+        logger.info(message);
+
+        return null;
+    }
+
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/decoder/TestDecodingMessageDecoder.java

@@ -0,0 +1,149 @@
+package org.dbsyncer.listener.postgresql.decoder;
+
+import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.listener.postgresql.AbstractMessageDecoder;
+import org.dbsyncer.listener.postgresql.column.ColumnValueResolver;
+import org.dbsyncer.listener.postgresql.column.Lexer;
+import org.dbsyncer.listener.postgresql.column.TestDecodingColumnValue;
+import org.dbsyncer.listener.postgresql.enums.MessageDecoderEnum;
+import org.dbsyncer.listener.postgresql.enums.MessageTypeEnum;
+import org.postgresql.replication.LogSequenceNumber;
+import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
+
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/17 23:00
+ */
+public class TestDecodingMessageDecoder extends AbstractMessageDecoder {
+
+    private static final ColumnValueResolver resolver = new ColumnValueResolver();
+
+    @Override
+    public boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn) {
+        if (super.skipMessage(buffer, startLsn, lastReceiveLsn)) {
+            return true;
+        }
+        int position = buffer.position();
+        try {
+            MessageTypeEnum type = MessageTypeEnum.getType((char) buffer.get());
+            switch (type) {
+                case BEGIN:
+                case COMMIT:
+                case RELATION:
+                case TRUNCATE:
+                case TYPE:
+                case ORIGIN:
+                case INSERT:
+                case UPDATE:
+                case DELETE:
+                case NONE:
+                    return true;
+                default:
+                    // TABLE
+                    return false;
+            }
+        } finally {
+            buffer.position(position);
+        }
+    }
+
+    @Override
+    public RowChangedEvent processMessage(ByteBuffer buffer) {
+        if (!buffer.hasArray()) {
+            throw new IllegalStateException("Invalid buffer received from PG server during streaming replication");
+        }
+        MessageTypeEnum type = MessageTypeEnum.getType((char) buffer.get());
+        if (MessageTypeEnum.TABLE == type) {
+            int offset = buffer.arrayOffset();
+            byte[] source = buffer.array();
+            return parseMessage(new String(source, offset, (source.length - offset)));
+        }
+        return null;
+    }
+
+    @Override
+    public String getOutputPlugin() {
+        return MessageDecoderEnum.TEST_DECODING.getType();
+    }
+
+    @Override
+    public void withSlotOption(ChainedLogicalStreamBuilder builder) {
+        builder.withSlotOption("include-xids", true)
+                .withSlotOption("skip-empty-xacts", true);
+    }
+
+    private RowChangedEvent parseMessage(String message) {
+        Lexer lexer = new Lexer(message);
+
+        // table
+        lexer.nextToken(' ');
+        // schemaName
+        lexer.nextToken('.');
+        // tableName
+        lexer.skip(1);
+        String table = lexer.nextToken('"');
+        lexer.skip(2);
+        // eventType
+        String eventType = lexer.nextToken(':');
+        lexer.skip(1);
+
+        List<Object> data = new ArrayList<>();
+        while (lexer.hasNext()) {
+            String name = parseName(lexer);
+            if ("(no-tuple-data)".equals(name)) {
+                // 删除时,无主键,不能同步
+                return null;
+            }
+            String type = parseType(lexer);
+            lexer.skip(1);
+            String value = parseValue(lexer);
+            data.add(resolver.resolveValue(type, new TestDecodingColumnValue(value)));
+        }
+
+        RowChangedEvent event = null;
+        if (StringUtil.equals(ConnectorConstant.OPERTION_UPDATE, eventType)) {
+            event = new RowChangedEvent(table, ConnectorConstant.OPERTION_UPDATE, Collections.EMPTY_LIST, data);
+        }
+
+        if (StringUtil.equals(ConnectorConstant.OPERTION_INSERT, eventType)) {
+            event = new RowChangedEvent(table, ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, data);
+        }
+
+        if (StringUtil.equals(ConnectorConstant.OPERTION_DELETE, eventType)) {
+            event = new RowChangedEvent(table, ConnectorConstant.OPERTION_DELETE, data, Collections.EMPTY_LIST);
+        }
+        return event;
+    }
+
+    private String parseName(Lexer lexer) {
+        if (lexer.current() == ' ') {
+            lexer.skip(1);
+        }
+        lexer.nextToken('[');
+        return lexer.token();
+    }
+
+    private String parseType(Lexer lexer) {
+        lexer.nextToken(']');
+        return lexer.token();
+    }
+
+    private String parseValue(Lexer lexer) {
+        if (lexer.current() == '\'') {
+            lexer.skip(1);
+            lexer.nextTokenToQuote();
+            return lexer.token();
+        }
+        lexer.nextToken(' ');
+        return lexer.token();
+    }
+
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/enums/MessageDecoderEnum.java

@@ -0,0 +1,50 @@
+package org.dbsyncer.listener.postgresql.enums;
+
+import org.dbsyncer.common.util.StringUtil;
+import org.dbsyncer.listener.ListenerException;
+import org.dbsyncer.listener.postgresql.MessageDecoder;
+import org.dbsyncer.listener.postgresql.decoder.PgOutputMessageDecoder;
+import org.dbsyncer.listener.postgresql.decoder.TestDecodingMessageDecoder;
+
+/**
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/4/17 23:05
+ */
+public enum MessageDecoderEnum {
+
+    /**
+     * 插件：TEST_DECODING
+     */
+    TEST_DECODING("test_decoding", TestDecodingMessageDecoder.class),
+
+    /**
+     * 插件：PG_OUTPUT
+     */
+    PG_OUTPUT("pgoutput", PgOutputMessageDecoder.class);
+
+    private String type;
+    private Class<?> clazz;
+
+    MessageDecoderEnum(String type, Class<?> clazz) {
+        this.type = type;
+        this.clazz = clazz;
+    }
+
+    public static MessageDecoder getMessageDecoder(String type) throws ListenerException, IllegalAccessException, InstantiationException {
+        for (MessageDecoderEnum e : MessageDecoderEnum.values()) {
+            if (StringUtil.equals(type, e.getType())) {
+                return (MessageDecoder) e.getClazz().newInstance();
+            }
+        }
+        return (MessageDecoder) TEST_DECODING.getClazz().newInstance();
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public Class<?> getClazz() {
+        return clazz;
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/postgresql/enums/MessageTypeEnum.java

@@ -0,0 +1,42 @@
+package org.dbsyncer.listener.postgresql.enums;
+
+public enum MessageTypeEnum {
+    BEGIN,
+    COMMIT,
+    TABLE,
+    INSERT,
+    UPDATE,
+    DELETE,
+    RELATION,
+    TRUNCATE,
+    TYPE,
+    ORIGIN,
+    NONE;
+
+    public static MessageTypeEnum getType(char type) {
+        switch (type) {
+            case 'B':
+                return BEGIN;
+            case 'C':
+                return COMMIT;
+            case 't':
+                return TABLE;
+            case 'I':
+                return INSERT;
+            case 'U':
+                return UPDATE;
+            case 'D':
+                return DELETE;
+            case 'R':
+                return RELATION;
+            case 'Y':
+                return TYPE;
+            case 'O':
+                return ORIGIN;
+            case 'T':
+                return TRUNCATE;
+            default:
+                return NONE;
+        }
+    }
+}

dbsyncer-listener/src/main/java/org/dbsyncer/listener/quartz/AbstractQuartzExtractor.java

@@ -17,7 +17,8 @@ import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -42,7 +43,8 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
     private Set<String> delete;
     private String taskKey;
     private long period;
-    private AtomicBoolean running;
+    private volatile boolean running;
+    private final Lock lock = new ReentrantLock(true);
 
     /**
      * 获取增量参数
@@ -65,7 +67,7 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
 
         taskKey = UUIDUtil.getUUID();
         period = listenerConfig.getPeriod();
-        running = new AtomicBoolean();
+        running = true;
         run();
         scheduledTaskService.start(taskKey, period * 1000, this);
         logger.info("启动定时任务:{} >> {}秒", taskKey, period);
@@ -73,24 +75,29 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
 
     @Override
     public void run() {
+        final Lock taskLock = lock;
+        boolean locked = false;
         try {
-            if (running.compareAndSet(false, true)) {
-                // 依次执行同步映射关系
+            locked = taskLock.tryLock();
+            if (locked) {
                 for (int i = 0; i < commandSize; i++) {
                     execute(commands.get(i), i);
                 }
-                running.compareAndSet(true, false);
             }
         } catch (Exception e) {
-            running.compareAndSet(true, false);
             errorEvent(e);
             logger.error(e.getMessage());
+        } finally {
+            if (locked) {
+                taskLock.unlock();
+            }
         }
     }
 
     @Override
     public void close() {
         scheduledTaskService.stop(taskKey);
+        running = false;
     }
 
     private void execute(Map<String, String> command, int index) {
@@ -98,7 +105,7 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
         ConnectorMapper connectionMapper = connectorFactory.connect(connectorConfig);
         Point point = checkLastPoint(command, index);
         int pageIndex = 1;
-        for (; ; ) {
+        while (running) {
             Result reader = connectorFactory.reader(connectionMapper, new ReaderConfig(point.getCommand(), point.getArgs(), pageIndex++, readNum));
             List<Map> data = reader.getSuccessData();
             if (CollectionUtils.isEmpty(data)) {

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

@@ -46,7 +46,6 @@ public class SqlServerExtractor extends AbstractExtractor {
     private static final String GET_ALL_CHANGES_FOR_TABLE = "SELECT * FROM cdc.[fn_cdc_get_all_changes_#](?, ?, N'all update old') order by [__$start_lsn] ASC, [__$seqval] ASC, [__$operation] ASC";
 
     private static final String LSN_POSITION = "position";
-    private static final long DEFAULT_POLL_INTERVAL_MILLIS = 300;
     private static final int OFFSET_COLUMNS = 4;
     private final Lock connectLock = new ReentrantLock();
     private volatile boolean connected;
@@ -114,9 +113,9 @@ public class SqlServerExtractor extends AbstractExtractor {
     }
 
     private void connect() {
-        DatabaseConfig cfg = (DatabaseConfig) connectorConfig;
-        if (connectorFactory.isAlive(cfg)) {
-            connectorMapper = (DatabaseConnectorMapper) connectorFactory.connect(cfg);
+        if (connectorFactory.isAlive(connectorConfig)) {
+            connectorMapper = (DatabaseConnectorMapper) connectorFactory.connect(connectorConfig);
+            DatabaseConfig cfg = connectorMapper.getConfig();
             serverName = cfg.getUrl();
             schema = cfg.getSchema();
         }
@@ -321,21 +320,17 @@ public class SqlServerExtractor extends AbstractExtractor {
                 try {
                     Lsn stopLsn = queryAndMap(GET_MAX_LSN, rs -> new Lsn(rs.getBytes(1)));
                     if (null == stopLsn || !stopLsn.isAvailable() || stopLsn.compareTo(lastLsn) <= 0) {
-                        TimeUnit.MILLISECONDS.sleep(DEFAULT_POLL_INTERVAL_MILLIS);
+                        sleepInMills(500L);
                         continue;
                     }
 
-                    pull(stopLsn);
-
                     lastLsn = stopLsn;
                     snapshot.put(LSN_POSITION, lastLsn.toString());
+
+                    pull(stopLsn);
                 } catch (Exception e) {
                     logger.error(e.getMessage());
-                    try {
-                        TimeUnit.SECONDS.sleep(1);
-                    } catch (InterruptedException ex) {
-                        logger.error(ex.getMessage());
-                    }
+                    sleepInMills(1000L);
                 }
             }
         }

dbsyncer-listener/src/main/test/PGReplicationTest.java

@@ -1,16 +1,14 @@
-import org.dbsyncer.connector.util.DatabaseUtil;
 import org.junit.Test;
 import org.postgresql.PGConnection;
+import org.postgresql.PGProperty;
 import org.postgresql.replication.LogSequenceNumber;
 import org.postgresql.replication.PGReplicationStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
-import java.sql.Connection;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.SQLException;
+import java.sql.*;
+import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 
 /**
@@ -29,18 +27,49 @@ public class PGReplicationTest {
         String driverClassNam = "org.postgresql.Driver";
         String username = "postgres";
         String password = "123456";
-        connection = DatabaseUtil.getConnection(driverClassNam, url, username, password);
+//        String slotName = "test_slot";
+        String slotName = "test_pgoutput";
+//        String outputPlugin = "test_decoding";
+        String outputPlugin = "pgoutput";
+        String publicationName = "mypub";
 
-        LogSequenceNumber currentLSN = query("SELECT pg_current_wal_lsn()", rs -> LogSequenceNumber.valueOf(rs.getString(1)));
+        Properties props = new Properties();
+        PGProperty.USER.set(props, username);
+        PGProperty.PASSWORD.set(props, password);
+        PGProperty.ASSUME_MIN_SERVER_VERSION.set(props, "9.4");
+        PGProperty.REPLICATION.set(props, "database");
+        PGProperty.PREFER_QUERY_MODE.set(props, "simple");
+        connection = DriverManager.getConnection(url, props);
 
-        PGConnection replConnection = connection.unwrap(PGConnection.class);
-        PGReplicationStream stream = replConnection
+        LogSequenceNumber lsn = currentXLogLocation();
+
+        PGConnection pgConnection = connection.unwrap(PGConnection.class);
+
+//        pgConnection.getReplicationAPI()
+//                    .createReplicationSlot()
+//                    .logical()
+//                    .withSlotName(slotName)
+//                    .withOutputPlugin(outputPlugin)
+//                    .make();
+
+        PGReplicationStream stream = pgConnection
                 .getReplicationAPI()
                 .replicationStream()
                 .logical()
-                .withSlotName("test_slot")
-                .withStartPosition(currentLSN)
+                .withSlotName(slotName)
+//                .withSlotOption("include-xids", true)
+//                .withSlotOption("skip-empty-xacts", true)
+                .withSlotOption("proto_version", 1)
+                .withSlotOption("publication_names", publicationName)
+                .withStatusInterval(5, TimeUnit.SECONDS)
+                .withStartPosition(lsn)
                 .start();
+
+        try {
+            Thread.sleep(10);
+        } catch (Exception e) {
+        }
+        stream.forceUpdateStatus();
         while (true) {
             //non blocking receive message
             ByteBuffer msg = stream.readPending();
@@ -57,6 +86,18 @@ public class PGReplicationTest {
 
     }
 
+    /**
+     * Returns the current position in the server tx log.
+     *
+     * @return a long value, never negative
+     * @throws SQLException if anything unexpected fails.
+     */
+    public LogSequenceNumber currentXLogLocation() throws SQLException {
+        int majorVersion = connection.getMetaData().getDatabaseMajorVersion();
+        String sql = majorVersion >= 10 ? "select * from pg_current_wal_lsn()" : "select * from pg_current_xlog_location()";
+        return query(sql, rs -> LogSequenceNumber.valueOf(rs.getString(1)));
+    }
+
     public <T> T query(String sql, ResultSetMapper mapper) {
         PreparedStatement ps = null;
         ResultSet rs = null;
@@ -76,6 +117,20 @@ public class PGReplicationTest {
         return apply;
     }
 
+    public boolean execute(String sql) {
+        PreparedStatement ps = null;
+        boolean execute = false;
+        try {
+            ps = connection.prepareStatement(sql);
+            execute = ps.execute();
+        } catch (Exception e) {
+            logger.error(e.getMessage());
+        } finally {
+            close(ps);
+        }
+        return execute;
+    }
+
     private void close(AutoCloseable closeable) {
         if (null != closeable) {
             try {

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

@@ -50,6 +50,25 @@
         </div>
         <div class="col-sm-6"></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>
+        <div class="col-sm-4">
+            <input id="dropSlotOnCloseSwitch" name="dropSlotOnClose"
+                   th:checked="${#maps.isEmpty(connector?.config?.properties) or connector?.config?.properties?.dropSlotOnClose eq 'true'}"
+                   type="checkbox">
+        </div>
+        <label class="col-sm-2 control-label">插件</label>
+        <div class="col-sm-4"
+             th:if="${#maps.isEmpty(connector?.config?.properties) or not #maps.containsKey(connector?.config?.properties, 'pluginName')}">
+            <input class="form-control" maxlength="32" name="pluginName" th:value="'test_decoding'" type="text"/>
+        </div>
+        <div class="col-sm-4"
+             th:if="${not #maps.isEmpty(connector?.config?.properties) and #maps.containsKey(connector?.config?.properties, 'pluginName')}">
+            <input class="form-control" maxlength="32" name="pluginName"
+                   th:value="${connector?.config?.properties?.pluginName}" type="text"/>
+        </div>
+    </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">驱动 </label>
         <div class="col-sm-10">
@@ -57,6 +76,18 @@
                    th:value="${connector?.config?.driverClassName} ?: 'org.postgresql.Driver'" type="text"/>
         </div>
     </div>
+
+    <script type="text/javascript">
+        $(function () {
+            $('#dropSlotOnCloseSwitch').bootstrapSwitch({
+                onText: "Yes",
+                offText: "No",
+                onColor: "success",
+                offColor: "info",
+                size: "normal"
+            });
+        })
+    </script>
 </div>
 
 </html>

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

@@ -31,6 +31,25 @@
         </div>
         <div class="col-sm-6"></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>
+        <div class="col-sm-4">
+            <input id="dropSlotOnCloseSwitch" name="dropSlotOnClose"
+                   th:checked="${#maps.isEmpty(connector?.config?.properties) or connector?.config?.properties?.dropSlotOnClose eq 'true'}"
+                   type="checkbox">
+        </div>
+        <label class="col-sm-2 control-label">插件</label>
+        <div class="col-sm-4"
+             th:if="${#maps.isEmpty(connector?.config?.properties) or not #maps.containsKey(connector?.config?.properties, 'pluginName')}">
+            <input class="form-control" maxlength="32" name="pluginName" th:value="'test_decoding'" type="text"/>
+        </div>
+        <div class="col-sm-4"
+             th:if="${not #maps.isEmpty(connector?.config?.properties) and #maps.containsKey(connector?.config?.properties, 'pluginName')}">
+            <input class="form-control" maxlength="32" name="pluginName"
+                   th:value="${connector?.config?.properties?.pluginName}" type="text"/>
+        </div>
+    </div>
     <div class="form-group">
         <label class="col-sm-2 control-label">驱动 </label>
         <div class="col-sm-10">
@@ -41,8 +60,13 @@
 
     <script type="text/javascript">
         $(function () {
-            // 初始化select插件
-            initSelectIndex($(".select-control"), 1);
+            $('#dropSlotOnCloseSwitch').bootstrapSwitch({
+                onText: "Yes",
+                offText: "No",
+                onColor: "success",
+                offColor: "info",
+                size: "normal"
+            });
         })
     </script>
 </div>

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

@@ -18,6 +18,7 @@
     <link type="text/css" rel="stylesheet" th:href="@{/plugins/css/bootstrap-dialog/bootstrap-dialog.min.css}"/>
     <link type="text/css" rel="stylesheet" th:href="@{/plugins/css/bootstrap-fileinput/fileinput.min.css}"/>
     <link type="text/css" rel="stylesheet" th:href="@{/plugins/css/bootstrap-select/bootstrap-select.min.css}"/>
+    <link rel="stylesheet" th:href="@{/plugins/css/bootstrap-switch/bootstrap-switch.min.css}" type="text/css"/>
     <link type="text/css" rel="stylesheet" th:href="@{/plugins/css/icheck/all.css}"/>
     <link type="text/css" rel="stylesheet" th:href="@{/plugins/css/loading-plus/loading-plus.css}"/>
     <link type="text/css" rel="stylesheet" th:href="@{/css/common.css}">
@@ -46,6 +47,7 @@
 <script th:src="@{/plugins/js/bootstrap-fileinput/theme.min.js}"></script>
 <script th:src="@{/plugins/js/bootstrap-fileinput/zh.js}"></script>
 <script th:src="@{/plugins/js/bootstrap-select/bootstrap-select.min.js}"></script>
+<script th:src="@{/plugins/js/bootstrap-switch/bootstrap-switch.min.js}"></script>
 <script th:src="@{/plugins/js/icheck/icheck.min.js}"></script>
 <script th:src="@{/plugins/js/loading-plus/loading-plus.js}"></script>
 <script th:src="@{/plugins/js/placeholder/jquery.placeholder.js}"></script>

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

@@ -61,7 +61,7 @@
                             <div class="form-group">
                                 <div class="col-sm-2 text-right"><label>匹配相似表</label></div>
                                 <div class="col-sm-10">
-                                    <input id="autoMatchTableSelect" name="autoMatchTable" class="form-control" type="checkbox" />
+                                    <input id="autoMatchTableSwitch" name="autoMatchTable" type="checkbox">
                                 </div>
                             </div>
 

dbsyncer-web/src/main/resources/static/js/mapping/add.js

@@ -11,10 +11,12 @@ function submit(data) {
 
 // 绑定匹配相似表复选框事件
 function bindAutoMatchTableCheckBoxClick(){
-    $('#autoMatchTableSelect').iCheck({
-        checkboxClass: 'icheckbox_square-blue',
-        labelHover: false,
-        cursor: true
+    $('#autoMatchTableSwitch').bootstrapSwitch({
+        onText: "Yes",
+        offText: "No",
+        onColor: "success",
+        offColor: "info",
+        size: "normal"
     });
 }
 

dbsyncer-web/src/main/resources/static/js/mapping/edit.js

@@ -129,18 +129,18 @@ function bindMappingTableGroupAddClick($sourceSelect, $targetSelect) {
         // 如果存在多个选择，只筛选相似表
         var sLen = m.sourceTable.length;
         var tLen = m.targetTable.length;
-        if(1 < sLen || 1 < tLen){
+        if (1 < sLen || 1 < tLen) {
             var mark = [];
-            for(j = 0; j < sLen; j++) {
-                if(-1 != m.targetTable.indexOf(m.sourceTable[j])){
+            for (j = 0; j < sLen; j++) {
+                if (-1 != m.targetTable.indexOf(m.sourceTable[j])) {
                     mark.push(m.sourceTable[j]);
                 }
             }
             m.sourceTable = mark;
             m.targetTable = mark;
         }
-        m.sourceTable = m.sourceTable.join();
-        m.targetTable = m.targetTable.join();
+        m.sourceTable = m.sourceTable.join('|');
+        m.targetTable = m.targetTable.join('|');
 
         doPoster("/tableGroup/add", m, function (data) {
             if (data.success == true) {

dbsyncer-web/src/main/resources/static/plugins/css/bootstrap-icheck/flat/icheck-allSkins.css

@@ -1,530 +0,0 @@
-/* iCheck plugin Flat skin
------------------------------------ */
-.icheckbox_flat,
-.iradio_flat {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(flat.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat {
-    background-position: 0 0;
-}
-    .icheckbox_flat.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat {
-    background-position: -88px 0;
-}
-    .iradio_flat.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat,
-    .iradio_flat {
-        background-image: url(flat@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* red */
-.icheckbox_flat-red,
-.iradio_flat-red {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(red.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-red {
-    background-position: 0 0;
-}
-    .icheckbox_flat-red.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-red.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-red.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-red {
-    background-position: -88px 0;
-}
-    .iradio_flat-red.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-red.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-red.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-red,
-    .iradio_flat-red {
-        background-image: url(red@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* green */
-.icheckbox_flat-green,
-.iradio_flat-green {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(green.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-green {
-    background-position: 0 0;
-}
-    .icheckbox_flat-green.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-green.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-green.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-green {
-    background-position: -88px 0;
-}
-    .iradio_flat-green.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-green.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-green.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-green,
-    .iradio_flat-green {
-        background-image: url(green@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* blue */
-.icheckbox_flat-blue,
-.iradio_flat-blue {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(blue.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-blue {
-    background-position: 0 0;
-}
-    .icheckbox_flat-blue.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-blue.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-blue.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-blue {
-    background-position: -88px 0;
-}
-    .iradio_flat-blue.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-blue.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-blue.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-blue,
-    .iradio_flat-blue {
-        background-image: url(blue@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* aero */
-.icheckbox_flat-aero,
-.iradio_flat-aero {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(aero.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-aero {
-    background-position: 0 0;
-}
-    .icheckbox_flat-aero.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-aero.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-aero.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-aero {
-    background-position: -88px 0;
-}
-    .iradio_flat-aero.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-aero.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-aero.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-aero,
-    .iradio_flat-aero {
-        background-image: url(aero@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* grey */
-.icheckbox_flat-grey,
-.iradio_flat-grey {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(grey.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-grey {
-    background-position: 0 0;
-}
-    .icheckbox_flat-grey.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-grey.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-grey.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-grey {
-    background-position: -88px 0;
-}
-    .iradio_flat-grey.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-grey.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-grey.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-grey,
-    .iradio_flat-grey {
-        background-image: url(grey@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* orange */
-.icheckbox_flat-orange,
-.iradio_flat-orange {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(orange.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-orange {
-    background-position: 0 0;
-}
-    .icheckbox_flat-orange.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-orange.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-orange.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-orange {
-    background-position: -88px 0;
-}
-    .iradio_flat-orange.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-orange.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-orange.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-orange,
-    .iradio_flat-orange {
-        background-image: url(orange@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* yellow */
-.icheckbox_flat-yellow,
-.iradio_flat-yellow {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(yellow.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-yellow {
-    background-position: 0 0;
-}
-    .icheckbox_flat-yellow.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-yellow.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-yellow.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-yellow {
-    background-position: -88px 0;
-}
-    .iradio_flat-yellow.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-yellow.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-yellow.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-yellow,
-    .iradio_flat-yellow {
-        background-image: url(yellow@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* pink */
-.icheckbox_flat-pink,
-.iradio_flat-pink {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(pink.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-pink {
-    background-position: 0 0;
-}
-    .icheckbox_flat-pink.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-pink.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-pink.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-pink {
-    background-position: -88px 0;
-}
-    .iradio_flat-pink.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-pink.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-pink.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-pink,
-    .iradio_flat-pink {
-        background-image: url(pink@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}
-
-/* purple */
-.icheckbox_flat-purple,
-.iradio_flat-purple {
-    display: inline-block;
-    *display: inline;
-    vertical-align: middle;
-    margin: 0;
-    padding: 0;
-    width: 20px;
-    height: 20px;
-    background: url(purple.png) no-repeat;
-    border: none;
-    cursor: pointer;
-}
-
-.icheckbox_flat-purple {
-    background-position: 0 0;
-}
-    .icheckbox_flat-purple.checked {
-        background-position: -22px 0;
-    }
-    .icheckbox_flat-purple.disabled {
-        background-position: -44px 0;
-        cursor: default;
-    }
-    .icheckbox_flat-purple.checked.disabled {
-        background-position: -66px 0;
-    }
-
-.iradio_flat-purple {
-    background-position: -88px 0;
-}
-    .iradio_flat-purple.checked {
-        background-position: -110px 0;
-    }
-    .iradio_flat-purple.disabled {
-        background-position: -132px 0;
-        cursor: default;
-    }
-    .iradio_flat-purple.checked.disabled {
-        background-position: -154px 0;
-    }
-
-/* HiDPI support */
-@media (-o-min-device-pixel-ratio: 5/4), (-webkit-min-device-pixel-ratio: 1.25), (min-resolution: 120dpi) {
-    .icheckbox_flat-purple,
-    .iradio_flat-purple {
-        background-image: url(purple@2x.png);
-        -webkit-background-size: 176px 22px;
-        background-size: 176px 22px;
-    }
-}

dbsyncer-web/src/main/resources/static/plugins/js/bootstrap-icheck/bootstrap-icheck.min.js

@@ -1,11 +0,0 @@
-/*! iCheck v1.0.2 by Damir Sultanov, http://git.io/arlzeA, MIT Licensed */
-(function(f){function A(a,b,d){var c=a[0],g=/er/.test(d)?_indeterminate:/bl/.test(d)?n:k,e=d==_update?{checked:c[k],disabled:c[n],indeterminate:"true"==a.attr(_indeterminate)||"false"==a.attr(_determinate)}:c[g];if(/^(ch|di|in)/.test(d)&&!e)x(a,g);else if(/^(un|en|de)/.test(d)&&e)q(a,g);else if(d==_update)for(var f in e)e[f]?x(a,f,!0):q(a,f,!0);else if(!b||"toggle"==d){if(!b)a[_callback]("ifClicked");e?c[_type]!==r&&q(a,g):x(a,g)}}function x(a,b,d){var c=a[0],g=a.parent(),e=b==k,u=b==_indeterminate,
-v=b==n,s=u?_determinate:e?y:"enabled",F=l(a,s+t(c[_type])),B=l(a,b+t(c[_type]));if(!0!==c[b]){if(!d&&b==k&&c[_type]==r&&c.name){var w=a.closest("form"),p='input[name="'+c.name+'"]',p=w.length?w.find(p):f(p);p.each(function(){this!==c&&f(this).data(m)&&q(f(this),b)})}u?(c[b]=!0,c[k]&&q(a,k,"force")):(d||(c[b]=!0),e&&c[_indeterminate]&&q(a,_indeterminate,!1));D(a,e,b,d)}c[n]&&l(a,_cursor,!0)&&g.find("."+C).css(_cursor,"default");g[_add](B||l(a,b)||"");g.attr("role")&&!u&&g.attr("aria-"+(v?n:k),"true");
-g[_remove](F||l(a,s)||"")}function q(a,b,d){var c=a[0],g=a.parent(),e=b==k,f=b==_indeterminate,m=b==n,s=f?_determinate:e?y:"enabled",q=l(a,s+t(c[_type])),r=l(a,b+t(c[_type]));if(!1!==c[b]){if(f||!d||"force"==d)c[b]=!1;D(a,e,s,d)}!c[n]&&l(a,_cursor,!0)&&g.find("."+C).css(_cursor,"pointer");g[_remove](r||l(a,b)||"");g.attr("role")&&!f&&g.attr("aria-"+(m?n:k),"false");g[_add](q||l(a,s)||"")}function E(a,b){if(a.data(m)){a.parent().html(a.attr("style",a.data(m).s||""));if(b)a[_callback](b);a.off(".i").unwrap();
-f(_label+'[for="'+a[0].id+'"]').add(a.closest(_label)).off(".i")}}function l(a,b,f){if(a.data(m))return a.data(m).o[b+(f?"":"Class")]}function t(a){return a.charAt(0).toUpperCase()+a.slice(1)}function D(a,b,f,c){if(!c){if(b)a[_callback]("ifToggled");a[_callback]("ifChanged")[_callback]("if"+t(f))}}var m="iCheck",C=m+"-helper",r="radio",k="checked",y="un"+k,n="disabled";_determinate="determinate";_indeterminate="in"+_determinate;_update="update";_type="type";_click="click";_touch="touchbegin.i touchend.i";
-_add="addClass";_remove="removeClass";_callback="trigger";_label="label";_cursor="cursor";_mobile=/ipad|iphone|ipod|android|blackberry|windows phone|opera mini|silk/i.test(navigator.userAgent);f.fn[m]=function(a,b){var d='input[type="checkbox"], input[type="'+r+'"]',c=f(),g=function(a){a.each(function(){var a=f(this);c=a.is(d)?c.add(a):c.add(a.find(d))})};if(/^(check|uncheck|toggle|indeterminate|determinate|disable|enable|update|destroy)$/i.test(a))return a=a.toLowerCase(),g(this),c.each(function(){var c=
-f(this);"destroy"==a?E(c,"ifDestroyed"):A(c,!0,a);f.isFunction(b)&&b()});if("object"!=typeof a&&a)return this;var e=f.extend({checkedClass:k,disabledClass:n,indeterminateClass:_indeterminate,labelHover:!0},a),l=e.handle,v=e.hoverClass||"hover",s=e.focusClass||"focus",t=e.activeClass||"active",B=!!e.labelHover,w=e.labelHoverClass||"hover",p=(""+e.increaseArea).replace("%","")|0;if("checkbox"==l||l==r)d='input[type="'+l+'"]';-50>p&&(p=-50);g(this);return c.each(function(){var a=f(this);E(a);var c=this,
-b=c.id,g=-p+"%",d=100+2*p+"%",d={position:"absolute",top:g,left:g,display:"block",width:d,height:d,margin:0,padding:0,background:"#fff",border:0,opacity:0},g=_mobile?{position:"absolute",visibility:"hidden"}:p?d:{position:"absolute",opacity:0},l="checkbox"==c[_type]?e.checkboxClass||"icheckbox":e.radioClass||"i"+r,z=f(_label+'[for="'+b+'"]').add(a.closest(_label)),u=!!e.aria,y=m+"-"+Math.random().toString(36).substr(2,6),h='<div class="'+l+'" '+(u?'role="'+c[_type]+'" ':"");u&&z.each(function(){h+=
-'aria-labelledby="';this.id?h+=this.id:(this.id=y,h+=y);h+='"'});h=a.wrap(h+"/>")[_callback]("ifCreated").parent().append(e.insert);d=f('<ins class="'+C+'"/>').css(d).appendTo(h);a.data(m,{o:e,s:a.attr("style")}).css(g);e.inheritClass&&h[_add](c.className||"");e.inheritID&&b&&h.attr("id",m+"-"+b);"static"==h.css("position")&&h.css("position","relative");A(a,!0,_update);if(z.length)z.on(_click+".i mouseover.i mouseout.i "+_touch,function(b){var d=b[_type],e=f(this);if(!c[n]){if(d==_click){if(f(b.target).is("a"))return;
-A(a,!1,!0)}else B&&(/ut|nd/.test(d)?(h[_remove](v),e[_remove](w)):(h[_add](v),e[_add](w)));if(_mobile)b.stopPropagation();else return!1}});a.on(_click+".i focus.i blur.i keyup.i keydown.i keypress.i",function(b){var d=b[_type];b=b.keyCode;if(d==_click)return!1;if("keydown"==d&&32==b)return c[_type]==r&&c[k]||(c[k]?q(a,k):x(a,k)),!1;if("keyup"==d&&c[_type]==r)!c[k]&&x(a,k);else if(/us|ur/.test(d))h["blur"==d?_remove:_add](s)});d.on(_click+" mousedown mouseup mouseover mouseout "+_touch,function(b){var d=
-b[_type],e=/wn|up/.test(d)?t:v;if(!c[n]){if(d==_click)A(a,!1,!0);else{if(/wn|er|in/.test(d))h[_add](e);else h[_remove](e+" "+t);if(z.length&&B&&e==v)z[/ut|nd/.test(d)?_remove:_add](w)}if(_mobile)b.stopPropagation();else return!1}})})}})(window.jQuery||window.Zepto);

pom.xml

@@ -49,6 +49,7 @@
         <kafka.version>0.9.0.0</kafka.version>
         <json.version>20090211</json.version>
         <fastjson.version>1.2.75</fastjson.version>
+        <log4j2.version>2.17.1</log4j2.version>
         <junit.version>4.12</junit.version>
     </properties>
 
@@ -170,6 +171,33 @@
                 <version>${kafka.version}</version>
             </dependency>
 
+            <!-- log4j -->
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-api</artifactId>
+                <version>${log4j2.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-slf4j-impl</artifactId>
+                <version>${log4j2.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-to-slf4j</artifactId>
+                <version>${log4j2.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-core</artifactId>
+                <version>${log4j2.version}</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.logging.log4j</groupId>
+                <artifactId>log4j-jul</artifactId>
+                <version>${log4j2.version}</version>
+            </dependency>
+
             <dependency>
                 <groupId>junit</groupId>
                 <artifactId>junit</artifactId>