write data and log

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

@@ -1,7 +1,6 @@
 package org.dbsyncer.biz;
 
 import org.dbsyncer.biz.vo.MappingVo;
-import org.dbsyncer.biz.vo.MetaVo;
 
 import java.util.List;
 import java.util.Map;
@@ -63,11 +62,4 @@ public interface MappingService {
      */
     String stop(String id);
 
-    /**
-     * 获取运行的驱动列表
-     *
-     * @return
-     */
-    List<MetaVo> getMetaAll();
-
 }

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

@@ -1,5 +1,10 @@
 package org.dbsyncer.biz;
 
+import org.dbsyncer.biz.vo.DataVo;
+import org.dbsyncer.biz.vo.LogVo;
+import org.dbsyncer.biz.vo.MetaVo;
+
+import java.util.List;
 import java.util.Map;
 
 /**
@@ -16,4 +21,46 @@ public interface MonitorService {
      */
     Map getThreadInfo();
 
+    /**
+     * 获取运行的驱动列表
+     *
+     * @return
+     */
+    List<MetaVo> getMetaAll();
+
+    /**
+     * 查询驱动同步数据
+     *
+     * @return
+     * @param id
+     * @param pageNum
+     * @param pageSize
+     */
+    List<DataVo> queryData(String id, int pageNum, int pageSize);
+
+    /**
+     * 清空驱动同步数据
+     *
+     * @param id
+     * @return
+     */
+    String clearData(String id);
+
+    /**
+     * 查询操作日志
+     *
+     * @param type
+     * @param pageNum
+     * @param pageSize
+     * @return
+     */
+    List<LogVo> queryLog(String type, int pageNum, int pageSize);
+
+    /**
+     * 清空操作日志
+     *
+     * @param id
+     * @return
+     */
+    String clearLog(String type);
 }

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

@@ -121,16 +121,6 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
         return "驱动停止成功";
     }
 
-    @Override
-    public List<MetaVo> getMetaAll() {
-        List<MetaVo> list = manager.getMetaAll()
-                .stream()
-                .map(m -> convertMeta2Vo(m))
-                .sorted(Comparator.comparing(MetaVo::getUpdateTime).reversed())
-                .collect(Collectors.toList());
-        return list;
-    }
-
     private MappingVo convertMapping2Vo(Mapping mapping) {
         String model = mapping.getModel();
         Assert.notNull(mapping, "Mapping can not be null.");
@@ -144,7 +134,7 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
         // 元信息
         Meta meta = manager.getMeta(mapping.getMetaId());
         Assert.notNull(meta, "Meta can not be null.");
-        MetaVo metaVo = new MetaVo(ModelEnum.getModelEnum(model).getName());
+        MetaVo metaVo = new MetaVo(ModelEnum.getModelEnum(model).getName(), mapping.getName());
         BeanUtils.copyProperties(meta, metaVo);
 
         MappingVo vo = new MappingVo(sConn, tConn, metaVo);
@@ -152,15 +142,6 @@ public class MappingServiceImpl extends BaseServiceImpl implements MappingServic
         return vo;
     }
 
-    private MetaVo convertMeta2Vo(Meta meta) {
-        Mapping mapping = manager.getMapping(meta.getMappingId());
-        Assert.notNull(mapping, "驱动不存在.");
-        ModelEnum modelEnum = ModelEnum.getModelEnum(mapping.getModel());
-        MetaVo metaVo = new MetaVo(modelEnum.getName());
-        BeanUtils.copyProperties(meta, metaVo);
-        return metaVo;
-    }
-
     /**
      * 检查是否存在驱动
      *

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

@@ -1,11 +1,26 @@
 package org.dbsyncer.biz.impl;
 
 import org.dbsyncer.biz.MonitorService;
+import org.dbsyncer.biz.vo.DataVo;
+import org.dbsyncer.biz.vo.LogVo;
+import org.dbsyncer.biz.vo.MetaVo;
+import org.dbsyncer.common.util.JsonUtil;
+import org.dbsyncer.manager.Manager;
 import org.dbsyncer.monitor.Monitor;
+import org.dbsyncer.parser.enums.ModelEnum;
+import org.dbsyncer.parser.model.Mapping;
+import org.dbsyncer.parser.model.Meta;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.util.Assert;
 
+import java.util.Comparator;
+import java.util.List;
 import java.util.Map;
+import java.util.stream.Collectors;
 
 /**
  * @author AE86
@@ -15,11 +30,81 @@ import java.util.Map;
 @Service
 public class MonitorServiceImpl implements MonitorService {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
     @Autowired
     private Monitor monitor;
 
+    @Autowired
+    private Manager manager;
+
     @Override
     public Map getThreadInfo() {
         return monitor.getThreadInfo();
     }
+
+    @Override
+    public List<MetaVo> getMetaAll() {
+        List<MetaVo> list = manager.getMetaAll()
+                .stream()
+                .map(m -> convertMeta2Vo(m))
+                .sorted(Comparator.comparing(MetaVo::getUpdateTime).reversed())
+                .collect(Collectors.toList());
+        return list;
+    }
+
+    @Override
+    public List<DataVo> queryData(String id, int pageNum, int pageSize) {
+        Assert.hasText(id, "id不能为空.");
+        List<DataVo> list = manager.queryData(id, pageNum, pageSize)
+                .stream()
+                .map(m -> convertData2Vo(m))
+                .sorted(Comparator.comparing(DataVo::getCreateTime).reversed())
+                .collect(Collectors.toList());
+        return list;
+    }
+
+    @Override
+    public String clearData(String id) {
+        manager.clearData(id);
+        return "清空同步数据成功";
+    }
+
+    @Override
+    public List<LogVo> queryLog(String type, int pageNum, int pageSize) {
+        List<LogVo> list = manager.queryLog(type, pageNum, pageSize)
+                .stream()
+                .map(m -> convertLog2Vo(m))
+                .sorted(Comparator.comparing(LogVo::getCreateTime).reversed())
+                .collect(Collectors.toList());
+        return list;
+    }
+
+    @Override
+    public String clearLog(String type) {
+        manager.clearLog(type);
+        return "清空日志成功";
+    }
+
+    private MetaVo convertMeta2Vo(Meta meta) {
+        Mapping mapping = manager.getMapping(meta.getMappingId());
+        Assert.notNull(mapping, "驱动不存在.");
+        ModelEnum modelEnum = ModelEnum.getModelEnum(mapping.getModel());
+        MetaVo metaVo = new MetaVo(modelEnum.getName(), mapping.getName());
+        metaVo.setMappingName(mapping.getName());
+        BeanUtils.copyProperties(meta, metaVo);
+        return metaVo;
+    }
+
+    private DataVo convertData2Vo(Map m) {
+        // TODO VO转换
+        logger.info(m.toString());
+        DataVo dataVo = new DataVo();
+        return dataVo;
+    }
+
+    private LogVo convertLog2Vo(Map m) {
+        LogVo logVo = new LogVo();
+        return logVo;
+    }
+
 }

dbsyncer-biz/src/main/java/org/dbsyncer/biz/vo/DataVo.java

@@ -0,0 +1,59 @@
+package org.dbsyncer.biz.vo;
+
+public class DataVo {
+
+    private String id;
+    private boolean success;
+    private String event;
+    private String error;
+    private String json;
+    private long createTime;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public boolean isSuccess() {
+        return success;
+    }
+
+    public void setSuccess(boolean success) {
+        this.success = success;
+    }
+
+    public String getEvent() {
+        return event;
+    }
+
+    public void setEvent(String event) {
+        this.event = event;
+    }
+
+    public String getError() {
+        return error;
+    }
+
+    public void setError(String error) {
+        this.error = error;
+    }
+
+    public String getJson() {
+        return json;
+    }
+
+    public void setJson(String json) {
+        this.json = json;
+    }
+
+    public long getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(long createTime) {
+        this.createTime = createTime;
+    }
+}

dbsyncer-biz/src/main/java/org/dbsyncer/biz/vo/LogVo.java

@@ -0,0 +1,41 @@
+package org.dbsyncer.biz.vo;
+
+public class LogVo {
+
+    private String id;
+    private String type;
+    private String json;
+    private long createTime;
+
+    public String getId() {
+        return id;
+    }
+
+    public void setId(String id) {
+        this.id = id;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getJson() {
+        return json;
+    }
+
+    public void setJson(String json) {
+        this.json = json;
+    }
+
+    public long getCreateTime() {
+        return createTime;
+    }
+
+    public void setCreateTime(long createTime) {
+        this.createTime = createTime;
+    }
+}

dbsyncer-biz/src/main/java/org/dbsyncer/biz/vo/MetaVo.java

@@ -11,9 +11,12 @@ public class MetaVo extends Meta {
 
     // 同步方式
     private String model;
+    // 驱动名称
+    private String mappingName;
 
-    public MetaVo(String model) {
+    public MetaVo(String model, String mappingName) {
         this.model = model;
+        this.mappingName = mappingName;
     }
 
     public String getModel() {
@@ -23,4 +26,12 @@ public class MetaVo extends Meta {
     public void setModel(String model) {
         this.model = model;
     }
+
+    public String getMappingName() {
+        return mappingName;
+    }
+
+    public void setMappingName(String mappingName) {
+        this.mappingName = mappingName;
+    }
 }

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

@@ -75,6 +75,16 @@ public interface Manager extends Executor {
 
     List<Meta> getMetaAll();
 
+    // Data
+    List<Map> queryData(String id, int pageNum, int pageSize);
+
+    void clearData(String id);
+
+    // Log
+    List<Map> queryLog(String type, int pageNum, int pageSize);
+
+    void clearLog(String type);
+
     // ConnectorEnum
     List<ConnectorEnum> getConnectorEnumAll();
 

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

@@ -12,6 +12,7 @@ import org.dbsyncer.manager.enums.GroupStrategyEnum;
 import org.dbsyncer.manager.enums.HandlerEnum;
 import org.dbsyncer.manager.puller.Puller;
 import org.dbsyncer.manager.template.impl.OperationTemplate;
+import org.dbsyncer.manager.template.impl.DataTemplate;
 import org.dbsyncer.parser.Parser;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.enums.MetaEnum;
@@ -19,6 +20,7 @@ import org.dbsyncer.parser.model.*;
 import org.dbsyncer.plugin.PluginFactory;
 import org.dbsyncer.plugin.config.Plugin;
 import org.dbsyncer.storage.constant.ConfigConstant;
+import org.dbsyncer.storage.enums.StorageEnum;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -51,6 +53,9 @@ public class ManagerFactory implements Manager, ApplicationContextAware, Applica
     @Autowired
     private OperationTemplate operationTemplate;
 
+    @Autowired
+    private DataTemplate dataTemplate;
+
     private Map<String, Puller> map;
 
     @Override
@@ -196,8 +201,27 @@ public class ManagerFactory implements Manager, ApplicationContextAware, Applica
         Meta meta = new Meta();
         meta.setType(ConfigConstant.META);
         QueryConfig<Meta> queryConfig = new QueryConfig<>(meta);
-        List<Meta> metas = operationTemplate.queryAll(queryConfig);
-        return metas;
+        return operationTemplate.queryAll(queryConfig);
+    }
+
+    @Override
+    public List<Map> queryData(String id, int pageNum, int pageSize) {
+        return dataTemplate.query(StorageEnum.DATA, id, pageNum, pageSize);
+    }
+
+    @Override
+    public void clearData(String id) {
+        dataTemplate.clear(StorageEnum.DATA, id);
+    }
+
+    @Override
+    public List<Map> queryLog(String type, int pageNum, int pageSize) {
+        return dataTemplate.query(StorageEnum.LOG, type, pageNum, pageSize);
+    }
+
+    @Override
+    public void clearLog(String type) {
+        dataTemplate.clear(StorageEnum.LOG, type);
     }
 
     @Override

dbsyncer-manager/src/main/java/org/dbsyncer/manager/config/PreloadConfig.java

@@ -1,6 +1,7 @@
 package org.dbsyncer.manager.config;
 
 import org.dbsyncer.manager.enums.GroupStrategyEnum;
+import org.dbsyncer.manager.enums.HandlerEnum;
 import org.dbsyncer.manager.template.Handler;
 
 public class PreloadConfig {
@@ -9,17 +10,17 @@ public class PreloadConfig {
 
     private GroupStrategyEnum groupStrategyEnum;
 
-    private Handler handler;
+    private HandlerEnum handlerEnum;
 
-    public PreloadConfig(String filterType, Handler handler) {
+    public PreloadConfig(String filterType, HandlerEnum handlerEnum) {
         this.filterType = filterType;
-        this.handler = handler;
+        this.handlerEnum = handlerEnum;
     }
 
-    public PreloadConfig(String filterType, GroupStrategyEnum groupStrategyEnum, Handler handler) {
+    public PreloadConfig(String filterType, GroupStrategyEnum groupStrategyEnum, HandlerEnum handlerEnum) {
         this.filterType = filterType;
         this.groupStrategyEnum = groupStrategyEnum;
-        this.handler = handler;
+        this.handlerEnum = handlerEnum;
     }
 
     public String getFilterType() {
@@ -30,7 +31,7 @@ public class PreloadConfig {
         return groupStrategyEnum;
     }
 
-    public Handler getHandler() {
-        return handler;
+    public HandlerEnum getHandlerEnum() {
+        return handlerEnum;
     }
 }

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

@@ -0,0 +1,38 @@
+package org.dbsyncer.manager.template.impl;
+
+import org.dbsyncer.storage.StorageService;
+import org.dbsyncer.storage.constant.ConfigConstant;
+import org.dbsyncer.storage.enums.StorageEnum;
+import org.dbsyncer.storage.query.Query;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 同步数据和日志模板
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2020/5/20 18:59
+ */
+@Component
+public final class DataTemplate {
+
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
+    @Autowired
+    private StorageService storageService;
+
+    public List<Map> query(StorageEnum type, String collectionId, int pageNum, int pageSize) {
+        Query query = new Query(pageNum, pageSize);
+        return storageService.query(type, query, collectionId);
+    }
+
+    public void clear(StorageEnum type, String collectionId) {
+        storageService.clear(type, collectionId);
+    }
+}

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

@@ -60,7 +60,7 @@ public final class PreloadTemplate extends AbstractTemplate implements Applicati
         boolean empty = CollectionUtils.isEmpty(list);
         logger.info("PreLoad {}:{}", filterType, empty ? 0 : list.size());
         if (!empty) {
-            Handler handler = config.getHandler();
+            Handler handler = config.getHandlerEnum().getHandler();
             GroupStrategyEnum strategy = getDefaultStrategy(config);
             list.forEach(map -> {
                 String json = (String) map.get(ConfigConstant.CONFIG_MODEL_JSON);
@@ -75,13 +75,13 @@ public final class PreloadTemplate extends AbstractTemplate implements Applicati
     @Override
     public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
         // Load connectors
-        execute(new PreloadConfig(ConfigConstant.CONNECTOR, HandlerEnum.PRELOAD_CONNECTOR.getHandler()));
+        execute(new PreloadConfig(ConfigConstant.CONNECTOR, HandlerEnum.PRELOAD_CONNECTOR));
         // Load mappings
-        execute(new PreloadConfig(ConfigConstant.MAPPING, HandlerEnum.PRELOAD_MAPPING.getHandler()));
+        execute(new PreloadConfig(ConfigConstant.MAPPING, HandlerEnum.PRELOAD_MAPPING));
         // Load tableGroups
-        execute(new PreloadConfig(ConfigConstant.TABLE_GROUP, GroupStrategyEnum.TABLE, HandlerEnum.PRELOAD_TABLE_GROUP.getHandler()));
+        execute(new PreloadConfig(ConfigConstant.TABLE_GROUP, GroupStrategyEnum.TABLE, HandlerEnum.PRELOAD_TABLE_GROUP));
         // Load metas
-        execute(new PreloadConfig(ConfigConstant.META, HandlerEnum.PRELOAD_META.getHandler()));
+        execute(new PreloadConfig(ConfigConstant.META, HandlerEnum.PRELOAD_META));
 
         // 启动驱动
         Meta meta = new Meta();

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

@@ -48,7 +48,7 @@ public class FlushServiceImpl implements FlushService {
         long now = System.currentTimeMillis();
         List<Map> list = data.parallelStream().map(r -> {
             Map<String, Object> params = new HashMap();
-            params.put(ConfigConstant.CONFIG_MODEL_ID, snowflakeIdWorker.nextId());
+            params.put(ConfigConstant.CONFIG_MODEL_ID, String.valueOf(snowflakeIdWorker.nextId()));
             params.put(ConfigConstant.DATA_SUCCESS, success);
             params.put(ConfigConstant.DATA_EVENT, event);
             params.put(ConfigConstant.DATA_ERROR, error);

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

@@ -33,6 +33,8 @@ public abstract class AbstractStorageService implements StorageService, Applicat
 
     public abstract void delete(String collectionId, String id) throws IOException;
 
+    public abstract void deleteAll(String collectionId) throws IOException;
+
     /**
      * 记录日志
      *
@@ -136,6 +138,16 @@ public abstract class AbstractStorageService implements StorageService, Applicat
         }
     }
 
+    @Override
+    public void clear(StorageEnum type, String collectionId) {
+        try {
+            deleteAll(getCollectionId(type, collectionId));
+        } catch (IOException e) {
+            logger.error("clear collectionId:{}, failed:{}", collectionId, e.getMessage());
+            throw new StorageException(e);
+        }
+    }
+
     private String getCollectionId(StorageEnum type, String collectionId) {
         Assert.notNull(type, "StorageEnum can not be null.");
         Strategy strategy = map.get(type.getType().concat("Strategy"));

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

@@ -45,4 +45,12 @@ public interface StorageService {
      * @param list
      */
     void addData(StorageEnum data, String collectionId, List<Map> list);
+
+    /**
+     * 清空数据/日志
+     *
+     * @param type
+     * @param collectionId
+     */
+    void clear(StorageEnum type, String collectionId);
 }

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

@@ -35,6 +35,8 @@ public class Shard {
 
     private String path;
 
+    private static final int MAX_SIZE = 10000;
+
     public Shard(String path) throws IOException {
         this.path = path;
         init();
@@ -66,8 +68,8 @@ public class Shard {
         }
     }
 
-    public List<Map> prefixQuery(Query query) throws IOException {
-        return executeQuery(query);
+    public List<Map> query(Query query, int pageNum, int pageSize) throws IOException {
+        return executeQuery(query, pageNum, pageSize);
     }
 
     public void insert(Document doc) throws IOException {
@@ -107,18 +109,28 @@ public class Shard {
      * 执行查询，并打印查询到的记录数
      *
      * @param query
+     * @param pageNum
+     * @param pageSize
      * @throws IOException
      */
-    private List<Map> executeQuery(Query query) throws IOException {
+    private List<Map> executeQuery(Query query, int pageNum, int pageSize) throws IOException {
+        TopDocs topDocs = indexSearcher.search(query, pageSize > MAX_SIZE ? MAX_SIZE : pageSize);
+
+        ScoreDoc[] docs = topDocs.scoreDocs;
+        int total = docs.length;
+        int begin = (pageNum - 1) * pageSize;
+        int end = pageNum * pageSize;
 
-        TopDocs topDocs = indexSearcher.search(query, 10000);
+        // 判断边界
+        begin = begin > total ? total : begin;
+        end = end > total ? total : end;
 
         List<Map> list = new ArrayList<>();
         Map r = null;
         IndexableField f = null;
-        for (ScoreDoc scoreDoc : topDocs.scoreDocs) {
+        while (begin < end) {
             //取得对应的文档对象
-            Document doc = indexSearcher.doc(scoreDoc.doc);
+            Document doc = indexSearcher.doc(docs[begin].doc);
             r = new LinkedHashMap<>();
             Iterator<IndexableField> iterator = doc.iterator();
             while (iterator.hasNext()) {
@@ -126,6 +138,7 @@ public class Shard {
                 r.put(f.name(), f.stringValue());
             }
             list.add(r);
+            begin++;
         }
         return list;
     }

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

@@ -7,9 +7,9 @@ package org.dbsyncer.storage.query;
  */
 public class Param {
     private String key;
-    private Object value;
+    private String value;
 
-    public Param(String key, Object value) {
+    public Param(String key, String value) {
         this.key = key;
         this.value = value;
     }
@@ -18,7 +18,7 @@ public class Param {
         return key;
     }
 
-    public Object getValue() {
+    public String getValue() {
         return value;
     }
-}
+}

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

@@ -12,8 +12,22 @@ public class Query {
 
     private List<Param> params;
 
+    private int pageNum = 1;
+
+    private int pageSize = 20;
+
     public Query() {
-        params = new ArrayList<>();
+        this.params = new ArrayList<>();
+    }
+
+    public Query(int pageNum, int pageSize) {
+        this.pageNum = pageNum;
+        this.pageSize = pageSize;
+        this.params = new ArrayList<>();
+    }
+
+    public void put(String key, String value) {
+        params.add(new Param(key, value));
     }
 
     public List<Param> getParams() {
@@ -24,9 +38,19 @@ public class Query {
         this.params = params;
     }
 
-    public void put(String key, String value) {
-        params.add(new Param(key, value));
+    public int getPageNum() {
+        return pageNum;
     }
-}
 
+    public void setPageNum(int pageNum) {
+        this.pageNum = pageNum;
+    }
 
+    public int getPageSize() {
+        return pageSize;
+    }
+
+    public void setPageSize(int pageSize) {
+        this.pageSize = pageSize;
+    }
+}

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

@@ -3,7 +3,10 @@ package org.dbsyncer.storage.support;
 import org.apache.lucene.document.Document;
 import org.apache.lucene.index.IndexableField;
 import org.apache.lucene.index.Term;
-import org.apache.lucene.search.PrefixQuery;
+import org.apache.lucene.search.BooleanClause;
+import org.apache.lucene.search.BooleanQuery;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.TermQuery;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.storage.AbstractStorageService;
 import org.dbsyncer.storage.StorageException;
@@ -60,15 +63,28 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
     @Override
     public List<Map> select(String collectionId, Query query) {
         Shard shard = map.get(collectionId);
-        if(null != shard){
+
+        // 检查是否存在历史
+        if (null == shard) {
+            shard = cacheShardIfExist(collectionId);
+        }
+
+        if (null != shard) {
             try {
+                int pageNum = query.getPageNum() <= 0 ? 1 : query.getPageNum();
+                int pageSize = query.getPageSize() <= 0 ? 20 : query.getPageSize();
+                // 设置参数
                 List<Param> params = query.getParams();
                 if (!CollectionUtils.isEmpty(params)) {
-                    Param p = params.get(0);
-                    Term term = new Term(p.getKey(), (String) p.getValue());
-                    PrefixQuery q = new PrefixQuery(term);
-                    return shard.prefixQuery(q);
+                    BooleanQuery.Builder builder = new BooleanQuery.Builder();
+                    params.forEach(p ->
+                            builder.add(new TermQuery(new Term(p.getKey(), p.getValue())), BooleanClause.Occur.MUST)
+                    );
+                    BooleanQuery q = builder.build();
+                    return shard.query(q, pageNum, pageSize);
                 }
+
+                return shard.query(new MatchAllDocsQuery(), pageNum, pageSize);
             } catch (IOException e) {
                 logger.error(e.getMessage());
             }
@@ -97,27 +113,34 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
         map.get(collectionId).delete(new Term(ConfigConstant.CONFIG_MODEL_ID, id));
     }
 
+    @Override
+    public void deleteAll(String collectionId) throws IOException {
+        Shard shard = map.get(collectionId);
+        if (null != shard) {
+            shard.deleteAll();
+            map.remove(collectionId);
+        }
+    }
+
     @Override
     public void insertLog(String collectionId, Map<String, Object> params) throws IOException {
-//        createShardIfNotExist(collectionId);
-        // TODO 实现日志写入
-        logger.info(params.toString());
+        createShardIfNotExist(collectionId);
+        Document doc = ParamsUtil.convertLog2Doc(params);
+        map.get(collectionId).insert(doc);
     }
 
     @Override
     public void insertData(String collectionId, List<Map> list) throws IOException {
-//        createShardIfNotExist(collectionId);
-        // TODO 实现数据写入
-        logger.info(list.toString());
-//        List<Document> docs = list.parallelStream().map(r -> ParamsUtil.convertData2Doc(r)).collect(Collectors.toList());
-//        map.get(collectionId).insertBatch(docs);
+        createShardIfNotExist(collectionId);
+        List<Document> docs = list.parallelStream().map(r -> ParamsUtil.convertData2Doc(r)).collect(Collectors.toList());
+        map.get(collectionId).insertBatch(docs);
     }
 
     /**
      * 如果不存在分片则创建(线程安全)
-     *<p>/data/config</p>
-     *<p>/data/log</p>
-     *<p>/data/data/123</p>
+     * <p>/data/config</p>
+     * <p>/data/log</p>
+     * <p>/data/data/123</p>
      *
      * @param collectionId
      * @throws IOException
@@ -128,4 +151,16 @@ public class DiskStorageServiceImpl extends AbstractStorageService {
         }
     }
 
+    private Shard cacheShardIfExist(String collectionId) {
+        String path = PATH + collectionId;
+        if (new File(path).exists()) {
+            try {
+                map.putIfAbsent(collectionId, new Shard(path));
+            } catch (IOException e) {
+                logger.error(e.getMessage());
+            }
+        }
+        return map.get(collectionId);
+    }
+
 }

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

@@ -38,6 +38,11 @@ public class MysqlStorageServiceImpl extends AbstractStorageService {
 
     }
 
+    @Override
+    public void deleteAll(String collectionId) throws IOException {
+
+    }
+
     @Override
     public void insertLog(String collectionId, Map<String, Object> params) throws IOException {
 

dbsyncer-storage/src/main/java/org/dbsyncer/storage/util/ParamsUtil.java

@@ -33,6 +33,21 @@ public abstract class ParamsUtil {
         return doc;
     }
 
+    public static Document convertLog2Doc(Map params) {
+        Assert.notNull(params, "Params can not be null.");
+        Document doc = new Document();
+        String id = (String) params.get(ConfigConstant.CONFIG_MODEL_ID);
+        String type = (String) params.get(ConfigConstant.CONFIG_MODEL_TYPE);
+        String json = (String) params.get(ConfigConstant.CONFIG_MODEL_JSON);
+        Long createTime = (Long) params.get(ConfigConstant.CONFIG_MODEL_CREATE_TIME);
+
+        doc.add(new StringField(ConfigConstant.CONFIG_MODEL_ID, id, Field.Store.YES));
+        doc.add(new StringField(ConfigConstant.CONFIG_MODEL_TYPE, type, Field.Store.YES));
+        doc.add(new StoredField(ConfigConstant.CONFIG_MODEL_JSON, json));
+        doc.add(new LongPoint(ConfigConstant.CONFIG_MODEL_CREATE_TIME, createTime));
+        return doc;
+    }
+
     public static Document convertData2Doc(Map params) {
         Assert.notNull(params, "Params can not be null.");
         Document doc = new Document();

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

@@ -1,8 +1,7 @@
 package org.dbsyncer.web.controller.index;
 
-import org.dbsyncer.biz.MappingService;
+import org.dbsyncer.biz.MonitorService;
 import org.dbsyncer.biz.vo.RestResult;
-import org.dbsyncer.web.controller.BaseController;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -13,18 +12,18 @@ import org.springframework.web.bind.annotation.ResponseBody;
 
 @Controller
 @RequestMapping("/meta")
-public class MetaController extends BaseController {
+public class MetaController {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     @Autowired
-    private MappingService mappingService;
+    private MonitorService monitorService;
 
     @GetMapping("/getAll")
     @ResponseBody
     public RestResult getAll() {
         try {
-            return RestResult.restSuccess(mappingService.getMetaAll());
+            return RestResult.restSuccess(monitorService.getMetaAll());
         } catch (Exception e) {
             logger.error(e.getLocalizedMessage(), e.getClass());
             return RestResult.restFail(e.getMessage());

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

@@ -1,10 +1,16 @@
 package org.dbsyncer.web.controller.monitor;
 
 import org.dbsyncer.biz.MonitorService;
+import org.dbsyncer.biz.vo.RestResult;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Controller;
 import org.springframework.ui.ModelMap;
+import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
 
 import javax.servlet.http.HttpServletRequest;
 
@@ -12,6 +18,8 @@ import javax.servlet.http.HttpServletRequest;
 @RequestMapping("/monitor")
 public class MonitorController {
 
+    private final Logger logger = LoggerFactory.getLogger(getClass());
+
     @Autowired
     private MonitorService monitorService;
 
@@ -21,4 +29,48 @@ public class MonitorController {
         return "monitor/monitor.html";
     }
 
+    @GetMapping("/queryData")
+    @ResponseBody
+    public RestResult queryData(@RequestParam(value = "id") String id, @RequestParam(value = "pageNum") int pageNum, @RequestParam(value = "pageSize") int pageSize) {
+        try {
+            return RestResult.restSuccess(monitorService.queryData(id, pageNum, pageSize));
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e.getClass());
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
+    @GetMapping("/queryLog")
+    @ResponseBody
+    public RestResult queryLog(@RequestParam(value = "type") String type, @RequestParam(value = "pageNum") int pageNum, @RequestParam(value = "pageSize") int pageSize) {
+        try {
+            return RestResult.restSuccess(monitorService.queryLog(type, pageNum, pageSize));
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e.getClass());
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
+    @GetMapping("/clearData")
+    @ResponseBody
+    public RestResult clearData(@RequestParam(value = "id") String id) {
+        try {
+            return RestResult.restSuccess(monitorService.clearData(id));
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e.getClass());
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
+    @GetMapping("/clearLog")
+    @ResponseBody
+    public RestResult clearLog(@RequestParam(value = "type") String type) {
+        try {
+            return RestResult.restSuccess(monitorService.clearLog(type));
+        } catch (Exception e) {
+            logger.error(e.getLocalizedMessage(), e.getClass());
+            return RestResult.restFail(e.getMessage());
+        }
+    }
+
 }