!71 merge

Merge pull request !71 from AE86/yjwang

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

@@ -21,6 +21,10 @@ public abstract class DateFormatUtil {
      * HH:mm:ss
      */
     public static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm:ss");
+    /**
+     * 默认时区
+     */
+    private static final ZoneId zoneId = ZoneId.systemDefault();
 
     private static final DateTimeFormatter TIME_TZ_FORMAT = new DateTimeFormatterBuilder()
             .append(DateTimeFormatter.ISO_LOCAL_TIME)
@@ -64,14 +68,18 @@ public abstract class DateFormatUtil {
         return date.toLocalDate().format(DATE_FORMATTER);
     }
 
-    public static String timestampToString(Timestamp timestamp) {
-        return timestamp.toLocalDateTime().format(CHINESE_STANDARD_TIME_FORMATTER);
+    public static String dateToString(java.util.Date date) {
+        return date.toInstant().atZone(zoneId).toLocalDateTime().format(CHINESE_STANDARD_TIME_FORMATTER);
     }
 
     public static Date stringToDate(String s) {
         return Date.valueOf(LocalDate.parse(s, DATE_FORMATTER));
     }
 
+    public static String timestampToString(Timestamp timestamp) {
+        return timestamp.toLocalDateTime().format(CHINESE_STANDARD_TIME_FORMATTER);
+    }
+
     public static LocalTime stringToLocalTime(String s) {
         return LocalTime.parse(s, CHINESE_STANDARD_TIME_FORMATTER);
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/AutoConfiguration.java → dbsyncer-connector/src/main/java/org/dbsyncer/connector/ConnectorConfiguration.java

@@ -9,7 +9,7 @@ import org.springframework.context.annotation.Configuration;
  * @date 2019/9/19 23:17
  */
 @Configuration
-public class AutoConfiguration {
+public class ConnectorConfiguration {
 
     @Bean
     public ConnectorFactory connectorFactory() {

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

@@ -1,11 +1,13 @@
 package org.dbsyncer.connector.database.setter;
 
+import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.database.AbstractSetter;
 
 import java.sql.Date;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.Timestamp;
+import java.time.LocalDateTime;
 
 public class TimestampSetter extends AbstractSetter<Timestamp> {
 
@@ -21,7 +23,20 @@ public class TimestampSetter extends AbstractSetter<Timestamp> {
             ps.setTimestamp(i, new Timestamp(date.getTime()));
             return;
         }
-        ps.setTimestamp(i, Timestamp.valueOf(String.valueOf(val)));
+
+        if (val instanceof LocalDateTime) {
+            LocalDateTime dateTime = (LocalDateTime) val;
+            ps.setTimestamp(i, Timestamp.valueOf(dateTime));
+            return;
+        }
+
+        if (val instanceof String) {
+            String s = (String) val;
+            ps.setTimestamp(i, Timestamp.valueOf(s));
+            return;
+        }
+
+        throw new ConnectorException(String.format("TimestampSetter can not find type [%s], val [%s]", type, val));
     }
 
 }

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

@@ -1,10 +1,15 @@
 package org.dbsyncer.connector.database.setter;
 
+import org.dbsyncer.common.util.DateFormatUtil;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.database.AbstractSetter;
 
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Date;
 
 public class VarcharSetter extends AbstractSetter<String> {
 
@@ -19,6 +24,23 @@ public class VarcharSetter extends AbstractSetter<String> {
             ps.setString(i, new String((byte[]) val));
             return;
         }
+
+        // TODO 1.2.0迭代 dbs将统一schema规范，统一转换处理，减少case
+        if (val instanceof LocalDateTime) {
+            ps.setString(i, ((LocalDateTime) val).format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
+            return;
+        }
+
+        if (val instanceof LocalDate) {
+            ps.setString(i, ((LocalDate) val).format(DateTimeFormatter.ofPattern("yyyy-MM-dd")));
+            return;
+        }
+
+        if (val instanceof Date) {
+            ps.setString(i, DateFormatUtil.dateToString((Date) val));
+            return;
+        }
+
         throw new ConnectorException(String.format("VarcharSetter can not find type [%s], val [%s]", type, val));
     }
 }

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

@@ -33,7 +33,8 @@ public final class SqlServerConnector extends AbstractDatabaseConnector {
     protected String getQueryCountSql(CommandConfig commandConfig, String schema, String quotation, String queryFilterSql) {
         // 有过滤条件，走默认方式
         if (StringUtil.isNotBlank(queryFilterSql)) {
-            return super.getQueryCountSql(commandConfig, schema, quotation, queryFilterSql);
+            String table = commandConfig.getTable().getName();
+            return new StringBuilder("SELECT COUNT(1) FROM ").append(schema).append(quotation).append(table).append(quotation).append(queryFilterSql).toString();
         }
 
         String table = commandConfig.getTable().getName();

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

@@ -1,3 +1,3 @@
 # 在这里开启自动状态，遵循spring 扫描规则
 org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
-org.dbsyncer.connector.AutoConfiguration
+org.dbsyncer.connector.ConnectorConfiguration

dbsyncer-monitor/src/main/java/org/dbsyncer/monitor/MonitorFactory.java

@@ -201,6 +201,7 @@ public class MonitorFactory implements Monitor {
             Query query = new Query(1, 1);
             operation.apply(query);
             metaAll.forEach(meta -> {
+                query.setQueryTotal(true);
                 Paging paging = manager.queryData(query, meta.getId());
                 total.getAndAdd(paging.getTotal());
             });

dbsyncer-parser/src/main/java/org/dbsyncer/parser/config/ParserStrategyConfiguration.java → dbsyncer-parser/src/main/java/org/dbsyncer/parser/ParserStrategyConfiguration.java

@@ -1,4 +1,4 @@
-package org.dbsyncer.parser.config;
+package org.dbsyncer.parser;
 
 import org.dbsyncer.parser.strategy.FlushStrategy;
 import org.dbsyncer.parser.strategy.ParserStrategy;

dbsyncer-parser/src/main/java/org/dbsyncer/parser/convert/handler/NumberToStringHandler.java

@@ -0,0 +1,21 @@
+package org.dbsyncer.parser.convert.handler;
+
+import org.dbsyncer.parser.convert.AbstractHandler;
+
+import java.sql.Timestamp;
+
+/**
+ * Number转String
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2022/7/20 23:04
+ */
+public class NumberToStringHandler extends AbstractHandler {
+
+    @Override
+    public Object convert(String args, Object value) {
+        return String.valueOf(value);
+    }
+
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/enums/ConvertEnum.java

@@ -46,6 +46,10 @@ public enum ConvertEnum {
      * String转Timestamp
      */
     STRING_TO_TIMESTAMP("STRING_TO_TIMESTAMP", "String转Timestamp", 0, new StringToTimestampHandler()),
+    /**
+     * Number转String
+     */
+    NUMBER_TO_STRING("NUMBER_TO_STRING", "Number转String", 0, new NumberToStringHandler()),
     /**
      * Byte[]转String
      */

dbsyncer-storage/src/main/java/org/dbsyncer/storage/config/StorageSupportConfiguration.java → dbsyncer-storage/src/main/java/org/dbsyncer/storage/StorageSupportConfiguration.java

@@ -1,6 +1,5 @@
-package org.dbsyncer.storage.config;
+package org.dbsyncer.storage;
 
-import org.dbsyncer.storage.StorageService;
 import org.dbsyncer.storage.support.DiskStorageServiceImpl;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/AbstractBinlogActuator.java

@@ -1,38 +0,0 @@
-package org.dbsyncer.storage.binlog;
-
-import org.dbsyncer.storage.enums.BinlogStatusEnum;
-import org.dbsyncer.storage.model.BinlogIndex;
-
-import java.time.LocalDateTime;
-
-public abstract class AbstractBinlogActuator implements BinlogActuator {
-
-    private BinlogIndex binlogIndex;
-
-    private BinlogStatusEnum status = BinlogStatusEnum.RUNNING;
-
-    protected void initBinlogIndex(BinlogIndex binlogIndex) {
-        binlogIndex.addLock(this);
-        this.binlogIndex = binlogIndex;
-    }
-
-    protected void refreshBinlogIndexUpdateTime() {
-        binlogIndex.setUpdateTime(LocalDateTime.now());
-    }
-
-    @Override
-    public String getFileName() {
-        return binlogIndex.getFileName();
-    }
-
-    @Override
-    public boolean isRunning() {
-        return status == BinlogStatusEnum.RUNNING;
-    }
-
-    @Override
-    public void stop() {
-        this.status = BinlogStatusEnum.STOP;
-    }
-
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/binlog/BinlogActuator.java

@@ -1,30 +0,0 @@
-package org.dbsyncer.storage.binlog;
-
-import java.io.Closeable;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/26 23:23
- */
-public interface BinlogActuator extends Closeable {
-
-    /**
-     * 获取索引文件名
-     *
-     * @return
-     */
-    String getFileName();
-
-    /**
-     * 状态是否为运行中
-     *
-     * @return
-     */
-    boolean isRunning();
-
-    /**
-     * 标记为停止状态
-     */
-    void stop();
-}

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

@@ -125,8 +125,13 @@ public class Shard {
         final IndexSearcher searcher = getSearcher();
         final TopDocs topDocs = getTopDocs(searcher, option.getQuery(), MAX_SIZE, sort);
         Paging paging = new Paging(pageNum, pageSize);
-        List<Map> data = search(searcher, topDocs, option, pageNum, pageSize);
         paging.setTotal(topDocs.totalHits);
+        if(option.isQueryTotal()){
+            paging.setData(Collections.EMPTY_LIST);
+            return paging;
+        }
+
+        List<Map> data = search(searcher, topDocs, option, pageNum, pageSize);
         paging.setData(data);
         return paging;
     }

dbsyncer-storage/src/main/java/org/dbsyncer/storage/model/BinlogConfig.java

@@ -1,29 +0,0 @@
-package org.dbsyncer.storage.model;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/19 23:03
- */
-public final class BinlogConfig {
-    private String fileName;
-    private long position = 0;
-
-    public String getFileName() {
-        return fileName;
-    }
-
-    public BinlogConfig setFileName(String fileName) {
-        this.fileName = fileName;
-        return this;
-    }
-
-    public long getPosition() {
-        return position;
-    }
-
-    public BinlogConfig setPosition(long position) {
-        this.position = position;
-        return this;
-    }
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/model/BinlogIndex.java

@@ -1,70 +0,0 @@
-package org.dbsyncer.storage.model;
-
-import org.dbsyncer.storage.binlog.BinlogActuator;
-
-import java.io.IOException;
-import java.time.LocalDateTime;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Set;
-
-/**
- * @author AE86
- * @version 1.0.0
- * @date 2022/6/26 22:48
- */
-public class BinlogIndex {
-    private String fileName;
-    private Set<BinlogActuator> lock;
-    private LocalDateTime createTime;
-    private LocalDateTime updateTime;
-
-    public BinlogIndex(String fileName, LocalDateTime createTime) {
-        this.fileName = fileName;
-        this.lock = new HashSet<>();
-        this.createTime = createTime;
-        this.updateTime = LocalDateTime.now();
-    }
-
-    public void addLock(BinlogActuator binlogActuator) {
-        this.lock.add(binlogActuator);
-    }
-
-    public void removeAllLock() throws IOException {
-        Iterator<BinlogActuator> iterator = lock.iterator();
-        while (iterator.hasNext()){
-            BinlogActuator next = iterator.next();
-            next.close();
-            iterator.remove();
-        }
-    }
-
-    public boolean isRunning() {
-        for (BinlogActuator actuator : lock){
-            if(actuator.isRunning()){
-                return true;
-            }
-        }
-        return false;
-    }
-
-    public boolean isFreeLock() {
-        return lock.isEmpty();
-    }
-
-    public String getFileName() {
-        return fileName;
-    }
-
-    public LocalDateTime getCreateTime() {
-        return createTime;
-    }
-
-    public LocalDateTime getUpdateTime() {
-        return updateTime;
-    }
-
-    public void setUpdateTime(LocalDateTime updateTime) {
-        this.updateTime = updateTime;
-    }
-}

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/Option.java

@@ -29,14 +29,20 @@ public class Option {
 
     private Highlighter highlighter = null;
 
-    private Map<String, IndexFieldResolverEnum> fieldResolvers = new LinkedHashMap<>();
+    /**
+     * 只查总数
+     */
+    private boolean queryTotal;
+
+    private final Map<String, IndexFieldResolverEnum> fieldResolvers = new LinkedHashMap<>();
 
     public Option(Query query) {
         this.query = query;
     }
 
-    public Option(Query query, List<Param> params) {
+    public Option(Query query, boolean queryTotal, List<Param> params) {
         this.query = query;
+        this.queryTotal = queryTotal;
         if (!CollectionUtils.isEmpty(params)) {
             this.highLightKeys = params.stream()
                     .filter(p -> p.isHighlighter())
@@ -65,6 +71,14 @@ public class Option {
         return query;
     }
 
+    public boolean isQueryTotal() {
+        return queryTotal;
+    }
+
+    public void setQueryTotal(boolean queryTotal) {
+        this.queryTotal = queryTotal;
+    }
+
     public Set<String> getHighLightKeys() {
         return highLightKeys;
     }

dbsyncer-storage/src/main/java/org/dbsyncer/storage/query/Query.java

@@ -27,6 +27,11 @@ public class Query {
 
     private boolean enableHighLightSearch;
 
+    /**
+     * 查询应用性能，不用排序查询，只用查询总量即可
+     */
+    private boolean queryTotal;
+
     public Query() {
         this.params = new ArrayList<>();
     }
@@ -95,4 +100,12 @@ public class Query {
     public boolean isEnableHighLightSearch() {
         return enableHighLightSearch;
     }
+
+    public boolean isQueryTotal() {
+        return queryTotal;
+    }
+
+    public void setQueryTotal(boolean queryTotal) {
+        this.queryTotal = queryTotal;
+    }
 }

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

@@ -73,6 +73,7 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
         if (null != shard) {
             int pageNum = query.getPageNum() <= 0 ? 1 : query.getPageNum();
             int pageSize = query.getPageSize() <= 0 ? 20 : query.getPageSize();
+            boolean queryTotal = query.isQueryTotal();
             // 根据修改时间 > 创建时间排序
             Sort sort = new Sort(new SortField(ConfigConstant.CONFIG_MODEL_UPDATE_TIME, SortField.Type.LONG, true),
                     new SortField(ConfigConstant.CONFIG_MODEL_CREATE_TIME, SortField.Type.LONG, true));
@@ -88,10 +89,10 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
                     }
                 });
                 BooleanQuery q = builder.build();
-                return shard.query(new Option(q, params), pageNum, pageSize, sort);
+                return shard.query(new Option(q, queryTotal, params), pageNum, pageSize, sort);
             }
 
-            return shard.query(new Option(new MatchAllDocsQuery()), pageNum, pageSize, sort);
+            return shard.query(new Option(new MatchAllDocsQuery(), queryTotal, null), pageNum, pageSize, sort);
         }
         return new Paging(query.getPageNum(), query.getPageSize());
     }

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

@@ -6,7 +6,6 @@ import org.dbsyncer.common.util.StringUtil;
 import org.dbsyncer.connector.ConnectorFactory;
 import org.dbsyncer.connector.ConnectorMapper;
 import org.dbsyncer.connector.config.DatabaseConfig;
-import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.config.SqlBuilderConfig;
 import org.dbsyncer.connector.config.WriterBatchConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
@@ -16,6 +15,7 @@ import org.dbsyncer.connector.database.DatabaseConnectorMapper;
 import org.dbsyncer.connector.enums.ConnectorEnum;
 import org.dbsyncer.connector.enums.SetterEnum;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
+import org.dbsyncer.connector.model.Field;
 import org.dbsyncer.connector.util.DatabaseUtil;
 import org.dbsyncer.storage.AbstractStorageService;
 import org.dbsyncer.storage.StorageException;
@@ -34,10 +34,7 @@ import org.springframework.util.Assert;
 
 import javax.annotation.PostConstruct;
 import java.io.*;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -95,16 +92,20 @@ public class MysqlStorageServiceImpl extends AbstractStorageService {
     public Paging select(Query query) {
         synchronized (LOCK){
             Executor executor = getExecutor(query.getType(), query.getCollection());
-            List<Object> queryArgs = new ArrayList<>();
+            Paging paging = new Paging(query.getPageNum(), query.getPageSize());
             List<Object> queryCountArgs = new ArrayList<>();
-            String querySql = buildQuerySql(query, executor, queryArgs);
             String queryCountSql = buildQueryCountSql(query, executor, queryCountArgs);
+            Long total = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(queryCountSql, queryCountArgs.toArray(), Long.class));
+            if (query.isQueryTotal()) {
+                paging.setTotal(total);
+                paging.setData(Collections.EMPTY_LIST);
+                return paging;
+            }
 
+            List<Object> queryArgs = new ArrayList<>();
+            String querySql = buildQuerySql(query, executor, queryArgs);
             List<Map<String, Object>> data = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForList(querySql, queryArgs.toArray()));
             replaceHighLight(query, data);
-            Long total = connectorMapper.execute(databaseTemplate -> databaseTemplate.queryForObject(queryCountSql, queryCountArgs.toArray(), Long.class));
-
-            Paging paging = new Paging(query.getPageNum(), query.getPageSize());
             paging.setData(data);
             paging.setTotal(total);
             return paging;