!60 merge

Merge pull request !60 from AE86/yjwang

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

@@ -1,7 +1,6 @@
 package org.dbsyncer.biz.checker.impl.mapping;
 
 import org.dbsyncer.biz.checker.MappingConfigChecker;
-import org.dbsyncer.common.util.NumberUtil;
 import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.listener.enums.ListenerTypeEnum;
@@ -23,7 +22,7 @@ public class TimingConfigChecker implements MappingConfigChecker {
 
     @Override
     public void modify(Mapping mapping, Map<String, String> params) {
-        String period = params.get("incrementStrategyTimingPeriodExpression");
+        String cron = params.get("incrementStrategyTimingCronExpression");
         String eventFieldName = params.get("incrementStrategyTimingEventFieldName");
         String insert = params.get("incrementStrategyTimingInsert");
         String update = params.get("incrementStrategyTimingUpdate");
@@ -32,8 +31,8 @@ public class TimingConfigChecker implements MappingConfigChecker {
         ListenerConfig config = mapping.getListener();
         Assert.notNull(config, "ListenerConfig can not be null.");
 
-        if (StringUtil.isNotBlank(period)) {
-            config.setPeriod(NumberUtil.toLong(period, 30));
+        if (StringUtil.isNotBlank(cron)) {
+            config.setCron(cron);
         }
         config.setEventFieldName(eventFieldName);
         if (StringUtil.isNotBlank(insert)) {

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

@@ -36,7 +36,10 @@ public abstract class AbstractDQLConnector extends AbstractDatabaseConnector {
     public MetaInfo getMetaInfo(DatabaseConnectorMapper connectorMapper, String tableName) {
         DatabaseConfig cfg = connectorMapper.getConfig();
         String sql = cfg.getSql().toUpperCase();
-        String queryMetaSql = StringUtil.contains(sql, " WHERE ") ? sql + " AND 1!=1 " : sql + " WHERE 1!=1 ";
+        sql = sql.replace("\t", " ");
+        sql = sql.replace("\r", " ");
+        sql = sql.replace("\n", " ");
+        String queryMetaSql = StringUtil.contains(sql, " WHERE ") ? cfg.getSql() + " AND 1!=1 " : cfg.getSql() + " WHERE 1!=1 ";
         return connectorMapper.execute(databaseTemplate -> super.getMetaInfo(databaseTemplate, queryMetaSql, cfg.getTable()));
     }
 

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

@@ -18,10 +18,10 @@ public class ListenerConfig {
     /**
      * 每次读取数
      */
-    private int readNum = 200;
+    private int readNum = 1000;
 
-    // 定时(秒)
-    private long period = 30;
+    // 定时表达式, 格式: [秒] [分] [小时] [日] [月] [周]
+    private String cron = "*/30 * * * * ?";
 
     // 事件字段
     private String eventFieldName = "";
@@ -61,14 +61,6 @@ public class ListenerConfig {
         this.readNum = readNum;
     }
 
-    public long getPeriod() {
-        return period;
-    }
-
-    public void setPeriod(long period) {
-        this.period = period;
-    }
-
     public String getEventFieldName() {
         return eventFieldName;
     }
@@ -108,4 +100,12 @@ public class ListenerConfig {
     public void setTableLabel(String tableLabel) {
         this.tableLabel = tableLabel;
     }
+
+    public String getCron() {
+        return cron;
+    }
+
+    public void setCron(String cron) {
+        this.cron = cron;
+    }
 }

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

@@ -15,25 +15,27 @@ public enum QuartzFilterEnum {
     /**
      * 时间戳(开始)
      */
-    TIME_STAMP_BEGIN("$timestamp_begin$", "系统时间戳(开始)", new TimestampFilter(true)),
+    TIME_STAMP_BEGIN(1, "$timestamp_begin$", "系统时间戳(开始)", new TimestampFilter(true)),
     /**
      * 时间戳(结束)
      */
-    TIME_STAMP_END("$timestamp_end$", "系统时间戳(结束)", new TimestampFilter(false)),
+    TIME_STAMP_END(2, "$timestamp_end$", "系统时间戳(结束)", new TimestampFilter(false)),
     /**
      * 日期(开始)
      */
-    DATE_BEGIN("$date_begin$", "系统日期(开始)", new DateFilter(true)),
+    DATE_BEGIN(3, "$date_begin$", "系统日期(开始)", new DateFilter(true)),
     /**
      * 日期(结束)
      */
-    DATE_END("$date_end$", "系统日期(结束)", new DateFilter(false));
+    DATE_END(4, "$date_end$", "系统日期(结束)", new DateFilter(false));
 
+    private Integer index;
     private String type;
     private String message;
     private QuartzFilter quartzFilter;
 
-    QuartzFilterEnum(String type, String message, QuartzFilter quartzFilter) {
+    QuartzFilterEnum(Integer index, String type, String message, QuartzFilter quartzFilter) {
+        this.index = index;
         this.type = type;
         this.message = message;
         this.quartzFilter = quartzFilter;
@@ -52,6 +54,10 @@ public enum QuartzFilterEnum {
         return null;
     }
 
+    public Integer getIndex() {
+        return index;
+    }
+
     public String getType() {
         return type;
     }

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

@@ -1,6 +1,7 @@
 package org.dbsyncer.listener.postgresql;
 
 import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.listener.postgresql.enums.MessageTypeEnum;
 import org.postgresql.replication.LogSequenceNumber;
 
 import java.nio.ByteBuffer;
@@ -16,7 +17,29 @@ public abstract class AbstractMessageDecoder implements MessageDecoder {
 
     @Override
     public boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn) {
-        return null == lastReceiveLsn || lastReceiveLsn.asLong() == 0 || startLsn.equals(lastReceiveLsn);
+        if (null == lastReceiveLsn || lastReceiveLsn.asLong() == 0 || startLsn.equals(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

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

@@ -2,6 +2,7 @@ package org.dbsyncer.listener.postgresql;
 
 import org.dbsyncer.common.event.RowChangedEvent;
 import org.dbsyncer.connector.config.DatabaseConfig;
+import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.postgresql.replication.LogSequenceNumber;
 import org.postgresql.replication.fluent.logical.ChainedLogicalStreamBuilder;
 
@@ -14,6 +15,9 @@ import java.nio.ByteBuffer;
  */
 public interface MessageDecoder {
 
+    default void postProcessBeforeInitialization(DatabaseConnectorMapper connectorMapper) {
+    }
+
     boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn);
 
     RowChangedEvent processMessage(ByteBuffer buffer);

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

@@ -90,6 +90,7 @@ public class PostgreSQLExtractor extends AbstractExtractor {
 
             messageDecoder = MessageDecoderEnum.getMessageDecoder(config.getProperty(PLUGIN_NAME));
             messageDecoder.setConfig(config);
+            messageDecoder.postProcessBeforeInitialization(connectorMapper);
             dropSlotOnClose = BooleanUtil.toBoolean(config.getProperty(DROP_SLOT_ON_CLOSE, "true"));
 
             connect();

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

@@ -1,15 +1,21 @@
 package org.dbsyncer.listener.postgresql.decoder;
 
 import org.dbsyncer.common.event.RowChangedEvent;
+import org.dbsyncer.connector.constant.ConnectorConstant;
+import org.dbsyncer.connector.database.DatabaseConnectorMapper;
+import org.dbsyncer.listener.ListenerException;
 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;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 
 /**
  * @author AE86
@@ -20,30 +26,30 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
+    private static final LocalDateTime PG_EPOCH = LocalDateTime.of(2000, 1, 1, 0, 0, 0);
+
     @Override
-    public boolean skipMessage(ByteBuffer buffer, LogSequenceNumber startLsn, LogSequenceNumber lastReceiveLsn) {
-        if (super.skipMessage(buffer, startLsn, lastReceiveLsn)) {
-            return true;
+    public void postProcessBeforeInitialization(DatabaseConnectorMapper connectorMapper) {
+        String pubName = getPubName();
+        String selectPublication = String.format("SELECT COUNT(1) FROM pg_publication WHERE pubname = '%s'", pubName);
+        Integer count = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(selectPublication, Integer.class));
+        if (0 < count) {
+            return;
         }
-        int position = buffer.position();
+
+        logger.info("Creating new publication '{}' for plugin '{}'", pubName, getOutputPlugin());
         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);
+            String createPublication = String.format("CREATE PUBLICATION %s FOR ALL TABLES", pubName);
+            logger.info("Creating Publication with statement '{}'", createPublication);
+            connectorMapper.execute(databaseTemplate -> {
+                databaseTemplate.execute(createPublication);
+                return true;
+            });
+        } catch (Exception e) {
+            throw new ListenerException(e.getCause());
         }
+
+        // TODO read table schema
     }
 
     @Override
@@ -51,15 +57,108 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
         if (!buffer.hasArray()) {
             throw new IllegalStateException("Invalid buffer received from PG server during streaming replication");
         }
+
+        RowChangedEvent event = null;
         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)));
+        switch (type) {
+            case UPDATE:
+                event = parseUpdate(buffer);
+                break;
+
+            case INSERT:
+                event = parseInsert(buffer);
+                break;
+
+            case DELETE:
+                event = parseDelete(buffer);
+                break;
+
+            case BEGIN:
+                long beginLsn = buffer.getLong();
+                long beginTs = buffer.getLong();
+                long xid = buffer.getInt();
+                logger.info("Begin LSN {}, timestamp {}, xid {} - {}", beginLsn, PG_EPOCH.plusNanos(beginTs * 1000L), xid, beginTs);
+                break;
+
+            case COMMIT:
+                buffer.get();
+                long commitLsn = buffer.getLong();
+                long commitEndLsn = buffer.getLong();
+                long commitTs = buffer.getLong();
+                logger.info("Commit: LSN {}, end LSN {}, ts {}", commitLsn, commitEndLsn, PG_EPOCH.plusNanos(commitTs * 1000L));
+                break;
+
+            default:
+                logger.info("Type {} not implemented", type.name());
+        }
+
+        if (null != event) {
+            logger.info(event.toString());
         }
+
         return null;
     }
 
+    private RowChangedEvent parseDelete(ByteBuffer buffer) {
+        int relationId = buffer.getInt();
+        logger.info("Delete table {}", relationId);
+
+        List<Object> data = new ArrayList<>();
+        String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
+
+        switch (newTuple) {
+            case "K":
+                readTupleData(buffer, data);
+                break;
+            default:
+                logger.info("K not set, got instead {}", newTuple);
+        }
+        return new RowChangedEvent(String.valueOf(relationId), ConnectorConstant.OPERTION_INSERT, data, Collections.EMPTY_LIST);
+    }
+
+    private RowChangedEvent parseInsert(ByteBuffer buffer) {
+        int relationId = buffer.getInt();
+        logger.info("Insert table {}", relationId);
+
+        List<Object> data = new ArrayList<>();
+        String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
+        switch (newTuple) {
+            case "N":
+                readTupleData(buffer, data);
+                break;
+            default:
+                logger.info("N not set, got instead {}", newTuple);
+        }
+        return new RowChangedEvent(String.valueOf(relationId), ConnectorConstant.OPERTION_INSERT, Collections.EMPTY_LIST, data);
+    }
+
+    private RowChangedEvent parseUpdate(ByteBuffer buffer) {
+        int relationId = buffer.getInt();
+        logger.info("Update table {}", relationId);
+
+        List<Object> data = new ArrayList<>();
+        String newTuple = new String(new byte[]{buffer.get()}, 0, 1);
+        switch (newTuple) {
+            case "K":
+                logger.info("Key update");
+                logger.info("Old Key");
+                readTupleData(buffer, data);
+                break;
+            case "O":
+                logger.info("Value update");
+                logger.info("Old Value");
+                readTupleData(buffer, data);
+                break;
+            case "N":
+                readTupleData(buffer, data);
+                break;
+            default:
+                logger.info("K or O Byte1 not set, got instead {}", newTuple);
+        }
+
+        return new RowChangedEvent(String.valueOf(relationId), ConnectorConstant.OPERTION_UPDATE, Collections.EMPTY_LIST, data);
+    }
+
     @Override
     public String getOutputPlugin() {
         return MessageDecoderEnum.PG_OUTPUT.getType();
@@ -68,13 +167,38 @@ public class PgOutputMessageDecoder extends AbstractMessageDecoder {
     @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()));
+        builder.withSlotOption("publication_names", getPubName());
     }
 
-    private RowChangedEvent parseMessage(String message) {
-        logger.info(message);
+    private String getPubName() {
+        return String.format("dbs_pub_%s_%s", config.getSchema(), config.getUsername());
+    }
 
-        return null;
+    private void readTupleData(ByteBuffer msg, List<Object> data) {
+        short nColumn = msg.getShort();
+        for (int n = 0; n < nColumn; n++) {
+            String tupleContentType = new String(new byte[]{msg.get()}, 0, 1);
+            if (tupleContentType.equals("t")) {
+                int size = msg.getInt();
+                byte[] text = new byte[size];
+
+                for (int z = 0; z < size; z++) {
+                    text[z] = msg.get();
+                }
+                String content = new String(text, 0, size);
+                data.add(content);
+                continue;
+            }
+
+            if (tupleContentType.equals("n")) {
+                data.add(null);
+                continue;
+            }
+
+            if (tupleContentType.equals("u")) {
+                data.add("TOASTED");
+            }
+        }
     }
 
 }

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

@@ -1,7 +1,6 @@
 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;
@@ -9,8 +8,9 @@ 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
@@ -24,37 +24,9 @@ import java.util.List;
  */
 public class TestDecodingMessageDecoder extends AbstractMessageDecoder {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
     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()) {
@@ -109,16 +81,18 @@ public class TestDecodingMessageDecoder extends AbstractMessageDecoder {
         }
 
         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);
+        switch (eventType) {
+            case ConnectorConstant.OPERTION_UPDATE:
+            case ConnectorConstant.OPERTION_INSERT:
+                event = new RowChangedEvent(table, eventType, Collections.EMPTY_LIST, data);
+                break;
+
+            case ConnectorConstant.OPERTION_DELETE:
+                event = new RowChangedEvent(table, eventType, data, Collections.EMPTY_LIST);
+                break;
+
+            default:
+                logger.info("Type {} not implemented", eventType);
         }
         return event;
     }

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

@@ -42,7 +42,6 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
     private Set<String> insert;
     private Set<String> delete;
     private String taskKey;
-    private long period;
     private volatile boolean running;
     private final Lock lock = new ReentrantLock(true);
 
@@ -66,11 +65,10 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
         delete = Stream.of(listenerConfig.getDelete().split(",")).collect(Collectors.toSet());
 
         taskKey = UUIDUtil.getUUID();
-        period = listenerConfig.getPeriod();
         running = true;
-        run();
-        scheduledTaskService.start(taskKey, period * 1000, this);
-        logger.info("启动定时任务:{} >> {}秒", taskKey, period);
+
+        scheduledTaskService.start(taskKey, listenerConfig.getCron(), this);
+        logger.info("启动定时任务:{} >> {}", taskKey, listenerConfig.getCron());
     }
 
     @Override
@@ -137,6 +135,10 @@ public abstract class AbstractQuartzExtractor extends AbstractExtractor implemen
             // 更新记录点
             point.refresh();
 
+            if (data.size() < readNum) {
+                break;
+            }
+
         }
 
         // 持久化

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

@@ -7,8 +7,11 @@ import org.dbsyncer.listener.enums.QuartzFilterEnum;
 import org.springframework.util.Assert;
 
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
@@ -25,10 +28,27 @@ public final class DatabaseQuartzExtractor extends AbstractQuartzExtractor {
     protected Point checkLastPoint(Map<String, String> command, int index) {
         // 检查是否存在系统参数
         final String query = command.get(ConnectorConstant.OPERTION_QUERY);
-        List<QuartzFilterEnum> filterEnums = Stream.of(QuartzFilterEnum.values()).filter(f -> {
-            Assert.isTrue(appearNotMoreThanOnce(query, f.getType()), String.format("系统参数%s存在多个.", f.getType()));
-            return StringUtil.contains(query, f.getType());
-        }).collect(Collectors.toList());
+
+        /**
+         * 排序开始/结束时间，防止系统生成的开始时间大于结束时间，导致无法查询有效范围结果集
+         * <p>fixed：select * from user where end_time > $timestamp_end$ and begin_time <= $timestamp_begin$
+         * <p>normal：select * from user where begin_time > $timestamp_begin$ and end_time <= $timestamp_end$
+         */
+        AtomicBoolean reversed = new AtomicBoolean();
+        AtomicLong lastIndex = new AtomicLong();
+        List<QuartzFilterEnum> filterEnums = Stream.of(QuartzFilterEnum.values())
+                .sorted(Comparator.comparing(QuartzFilterEnum::getIndex))
+                .filter(f -> {
+                    int currentIndex = StringUtil.indexOf(query, f.getType());
+                    Assert.isTrue((currentIndex == StringUtil.lastIndexOf(query, f.getType())), String.format("系统参数%s存在多个.", f.getType()));
+                    boolean exist = StringUtil.contains(query, f.getType());
+                    if (exist && !reversed.get()) {
+                        reversed.set(lastIndex.get() > currentIndex);
+                        lastIndex.set(currentIndex);
+                    }
+                    return exist;
+                }).collect(Collectors.toList());
+
         if (CollectionUtils.isEmpty(filterEnums)) {
             return new Point(command, new ArrayList<>());
         }
@@ -47,7 +67,7 @@ public final class DatabaseQuartzExtractor extends AbstractQuartzExtractor {
             final String key = index + type;
 
             // 开始位置
-            if(f.begin()){
+            if (f.begin()) {
                 if (!snapshot.containsKey(key)) {
                     final Object val = f.getObject();
                     point.addArg(val);
@@ -68,11 +88,11 @@ public final class DatabaseQuartzExtractor extends AbstractQuartzExtractor {
             point.setBeginValue(f.toString(val));
         }
         point.setCommand(ConnectorConstant.OPERTION_QUERY, replaceQuery);
+        if (reversed.get()) {
+            point.reverseArgs();
+        }
 
         return point;
     }
 
-    private boolean appearNotMoreThanOnce(String str, String searchStr) {
-        return StringUtil.indexOf(str, searchStr) == StringUtil.lastIndexOf(str, searchStr);
-    }
 }

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

@@ -2,10 +2,7 @@ package org.dbsyncer.listener.quartz;
 
 import org.dbsyncer.common.util.StringUtil;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 public class Point {
 
@@ -66,4 +63,7 @@ public class Point {
         this.beginValue = beginValue;
     }
 
+    public void reverseArgs() {
+        Collections.reverse(args);
+    }
 }

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

@@ -6,9 +6,10 @@
     <div class="form-group">
         <div class="row">
             <div class="col-md-4">
-                <label class="col-sm-3 control-label text-right">间隔(秒)*</label>
+                <label class="col-sm-3 control-label text-right">Cron*</label>
                 <div class="col-sm-9">
-                    <input name="incrementStrategyTimingPeriodExpression" type="number" min="1" class="form-control" dbsyncer-valid="require" th:value="${mapping?.listener?.period}?:30"/>
+                    <input class="form-control" dbsyncer-valid="require" name="incrementStrategyTimingCronExpression"
+                           th:value="${mapping?.listener?.cron}?:'*/30 * * * * ?'" type="text"/>
                 </div>
             </div>
             <div class="col-md-4">