!9 merge

Merge pull request !9 from AE86/V_1.0.0

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

@@ -132,6 +132,7 @@ public class MappingServiceImpl implements MappingService {
     }
 
     private MappingVo convertMapping2Vo(Mapping mapping) {
+        String model = mapping.getModel();
         Assert.notNull(mapping, "Mapping can not be null.");
         Connector s = manager.getConnector(mapping.getSourceConnectorId());
         Connector t = manager.getConnector(mapping.getTargetConnectorId());
@@ -139,9 +140,14 @@ public class MappingServiceImpl implements MappingService {
         BeanUtils.copyProperties(s, sConn);
         ConnectorVo tConn = new ConnectorVo(monitor.alive(t.getId()));
         BeanUtils.copyProperties(t, tConn);
-        boolean isRunning = isRunning(mapping.getMetaId());
 
-        MappingVo vo = new MappingVo(isRunning, sConn, tConn);
+        // 元信息
+        Meta meta = manager.getMeta(mapping.getMetaId());
+        Assert.notNull(meta, "Meta can not be null.");
+        MetaVo metaVo = new MetaVo(ModelEnum.getModelEnum(model).getName());
+        BeanUtils.copyProperties(meta, metaVo);
+
+        MappingVo vo = new MappingVo(sConn, tConn, metaVo);
         BeanUtils.copyProperties(mapping, vo);
         return vo;
     }
@@ -150,7 +156,7 @@ public class MappingServiceImpl implements MappingService {
         Mapping mapping = manager.getMapping(meta.getMappingId());
         Assert.notNull(mapping, "驱动不存在.");
         ModelEnum modelEnum = ModelEnum.getModelEnum(mapping.getModel());
-        MetaVo metaVo = new MetaVo(mapping.getName(), modelEnum.getMessage());
+        MetaVo metaVo = new MetaVo(modelEnum.getName());
         BeanUtils.copyProperties(meta, metaVo);
         return metaVo;
     }

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

@@ -9,40 +9,28 @@ import org.dbsyncer.parser.model.Mapping;
  */
 public class MappingVo extends Mapping {
 
-    // 是否运行
-    private boolean running;
-
     // 连接器
     private ConnectorVo sourceConnector;
     private ConnectorVo targetConnector;
 
-    public MappingVo(boolean running, ConnectorVo sourceConnector, ConnectorVo targetConnector) {
-        this.running = running;
+    // 元信息
+    private MetaVo meta;
+
+    public MappingVo(ConnectorVo sourceConnector, ConnectorVo targetConnector, MetaVo meta) {
         this.sourceConnector = sourceConnector;
         this.targetConnector = targetConnector;
-    }
-
-    public boolean isRunning() {
-        return running;
-    }
-
-    public void setRunning(boolean running) {
-        this.running = running;
+        this.meta = meta;
     }
 
     public ConnectorVo getSourceConnector() {
         return sourceConnector;
     }
 
-    public void setSourceConnector(ConnectorVo sourceConnector) {
-        this.sourceConnector = sourceConnector;
-    }
-
     public ConnectorVo getTargetConnector() {
         return targetConnector;
     }
 
-    public void setTargetConnector(ConnectorVo targetConnector) {
-        this.targetConnector = targetConnector;
+    public MetaVo getMeta() {
+        return meta;
     }
 }

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

@@ -9,24 +9,13 @@ import org.dbsyncer.parser.model.Meta;
  */
 public class MetaVo extends Meta {
 
-    // 驱动名称
-    private String mappingName;
     // 同步方式
     private String model;
 
-    public MetaVo(String mappingName, String model) {
-        this.mappingName = mappingName;
+    public MetaVo(String model) {
         this.model = model;
     }
 
-    public String getMappingName() {
-        return mappingName;
-    }
-
-    public void setMappingName(String mappingName) {
-        this.mappingName = mappingName;
-    }
-
     public String getModel() {
         return model;
     }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/Connector.java

@@ -71,7 +71,7 @@ public interface Connector {
     /**
      * 分页获取数据源数据
      *
-     * @param config    数据源配置
+     * @param config    连接器配置
      * @param command   执行命令
      * @param pageIndex 页数
      * @param pageSize  页大小
@@ -82,11 +82,23 @@ public interface Connector {
     /**
      * 批量写入目标源数据
      *
-     * @param config  数据源配置
+     * @param config  连接器配置
      * @param command 执行命令
+     * @param fields  字段信息
      * @param data    数据
      * @return
      */
     Result writer(ConnectorConfig config, Map<String, String> command, List<Field> fields, List<Map<String, Object>> data);
 
+    /**
+     * 写入目标源数据
+     *
+     * @param config  连接器配置
+     * @param fields  字段信息
+     * @param command 执行命令
+     * @param event   事件
+     * @param data    数据
+     * @return
+     */
+    Result writer(ConnectorConfig config, List<Field> fields, Map<String, String> command, String event, Map<String, Object> data);
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/ConnectorFactory.java

@@ -100,6 +100,13 @@ public class ConnectorFactory {
         return result;
     }
 
+    public Result writer(ConnectorConfig config, List<Field> fields, Map<String, String> command, String event, Map<String, Object> data) {
+        Connector connector = getConnector(config.getConnectorType());
+        Result result = connector.writer(config, fields, command, event, data);
+        Assert.notNull(result, "Connector writer result can not null");
+        return result;
+    }
+
     /**
      * 获取连接器
      *

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

@@ -5,6 +5,7 @@ import org.dbsyncer.common.model.Result;
 import org.dbsyncer.common.util.CollectionUtils;
 import org.dbsyncer.connector.ConnectorException;
 import org.dbsyncer.connector.config.*;
+import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.connector.enums.OperationEnum;
 import org.dbsyncer.connector.enums.SetterEnum;
 import org.dbsyncer.connector.enums.SqlBuilderEnum;
@@ -14,6 +15,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.jdbc.core.BatchPreparedStatementSetter;
 import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.PreparedStatementSetter;
 import org.springframework.util.Assert;
 
 import java.sql.Connection;
@@ -221,6 +223,54 @@ public abstract class AbstractDatabaseConnector implements Database {
         return result;
     }
 
+    @Override
+    public Result writer(ConnectorConfig config, List<Field> fields, Map<String, String> command, String event, Map<String, Object> data) {
+        // 1、获取 SQL
+        String sql = command.get(event);
+        Assert.hasText(sql, "执行语句不能为空.");
+        if (CollectionUtils.isEmpty(data) || CollectionUtils.isEmpty(fields)) {
+            logger.error("writer data can not be empty.");
+            throw new ConnectorException("writer data can not be empty.");
+        }
+        List<Object> args = new ArrayList<>();
+        fields.forEach(f -> args.add(data.get(f.getName())));
+        if (!StringUtils.equals(ConnectorConstant.OPERTION_INSERT, event)) {
+            List<Field> pkList = fields.stream().filter(f -> f.isPk()).collect(Collectors.toList());
+            fields.add(pkList.get(0));
+        }
+        int size = fields.size();
+
+        DatabaseConfig cfg = (DatabaseConfig) config;
+        JdbcTemplate jdbcTemplate = null;
+        Result result = new Result();
+        try {
+            // 2、获取连接
+            jdbcTemplate = getJdbcTemplate(cfg);
+
+            // 3、设置参数
+            int update = jdbcTemplate.update(sql, (ps)-> {
+                Field f = null;
+                for (int i = 0; i < size; i++) {
+                    f = fields.get(i);
+                    SetterEnum.getSetter(f.getType()).set(ps, i + 1, f.getType(), data.get(f.getName()));
+                }
+            });
+            if (0 == update) {
+                throw new ConnectorException("写入失败");
+            }
+        } catch (Exception e) {
+            // 记录错误数据
+            result.getFailData().add(data);
+            result.getFail().set(1);
+            result.getError().append(e.getMessage()).append("\r\n");
+            logger.error(e.getMessage());
+        } finally {
+            // 释放连接
+            this.close(jdbcTemplate);
+        }
+        return result;
+    }
+
     @Override
     public JdbcTemplate getJdbcTemplate(DatabaseConfig config) {
         return DatabaseUtil.getJdbcTemplate(config);
@@ -395,7 +445,7 @@ public abstract class AbstractDatabaseConnector implements Database {
             f = fields.get(i);
             type = f.getType();
             val = row.get(f.getName());
-            SetterEnum.getSetter(type).preparedStatementSetter(ps, i + 1, type, val);
+            SetterEnum.getSetter(type).set(ps, i + 1, type, val);
         }
     }
 

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

@@ -15,7 +15,7 @@ public abstract class AbstractSetter implements Setter {
     protected abstract void set(PreparedStatement ps, int i, Object val) throws SQLException;
 
     @Override
-    public void preparedStatementSetter(PreparedStatement ps, int i, int type, Object val) {
+    public void set(PreparedStatement ps, int i, int type, Object val) {
         try {
             if (null == val) {
                 ps.setNull(i, type);

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

@@ -4,6 +4,6 @@ import java.sql.PreparedStatement;
 
 public interface Setter {
 
-    void preparedStatementSetter(PreparedStatement ps, int i, int type, Object val);
+    void set(PreparedStatement ps, int i, int type, Object val);
     
 }

dbsyncer-connector/src/main/java/org/dbsyncer/connector/ldap/LdapConnector.java

@@ -68,6 +68,11 @@ public final class LdapConnector implements Ldap {
 		return null;
 	}
 
+	@Override
+	public Result writer(ConnectorConfig config, List<Field> fields, Map<String, String> command, String event, Map<String, Object> data) {
+		return null;
+	}
+
 	@Override
 	public LdapTemplate getLdapTemplate(LdapConfig config) throws AuthenticationException, CommunicationException, javax.naming.NamingException {
 		return LdapUtil.getLdapTemplate(config);

dbsyncer-connector/src/main/java/org/dbsyncer/connector/redis/RedisConnector.java

@@ -72,6 +72,11 @@ public final class RedisConnector implements Redis {
         return null;
     }
 
+    @Override
+    public Result writer(ConnectorConfig config, List<Field> fields, Map<String, String> command, String event, Map<String, Object> data) {
+        return null;
+    }
+
     @Override
     public RedisTemplate getRedisTemplate(RedisConfig config) {
         return this.getRedisTemplate(config, null, null, null);

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

@@ -36,6 +36,13 @@ public abstract class DefaultExtractor implements Extractor {
         }
     }
 
+    public void clearAllListener() {
+        if (null != watcher) {
+            watcher.clear();
+            watcher = null;
+        }
+    }
+
     public void changedEvent(String tableName, String event, List<Object> before, List<Object> after) {
         if (!CollectionUtils.isEmpty(watcher)) {
             watcher.forEach(w -> w.changedEvent(tableName, event, before, after));

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

@@ -14,6 +14,9 @@ public interface Listener {
      * @param listenerConfig  监听器配置
      * @param map             增量参数
      * @return
+     * @throws IllegalAccessException
+     * @throws InstantiationException
      */
-    DefaultExtractor createExtractor(ConnectorConfig connectorConfig, ListenerConfig listenerConfig, Map<String, String> map);
+    DefaultExtractor createExtractor(ConnectorConfig connectorConfig, ListenerConfig listenerConfig, Map<String, String> map)
+            throws IllegalAccessException, InstantiationException;
 }

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

@@ -1,6 +1,5 @@
 package org.dbsyncer.listener;
 
-
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.listener.enums.ListenerEnum;
@@ -13,8 +12,10 @@ import java.util.Map;
 public class ListenerFactory implements Listener {
 
     @Override
-    public DefaultExtractor createExtractor(ConnectorConfig config, ListenerConfig listenerConfig, Map<String, String> map) {
-        DefaultExtractor extractor = ListenerEnum.getExtractor(config.getConnectorType());
+    public DefaultExtractor createExtractor(ConnectorConfig config, ListenerConfig listenerConfig, Map<String, String> map)
+            throws IllegalAccessException, InstantiationException {
+        Class<DefaultExtractor> clazz = (Class<DefaultExtractor>) ListenerEnum.getExtractor(config.getConnectorType());
+        DefaultExtractor extractor = clazz.newInstance();
         // log/timing
         extractor.setAction(ListenerTypeEnum.getAction(listenerConfig.getListenerType()));
         extractor.setConnectorConfig(config);

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

@@ -2,7 +2,6 @@ package org.dbsyncer.listener.enums;
 
 import org.apache.commons.lang.StringUtils;
 import org.dbsyncer.connector.enums.ConnectorEnum;
-import org.dbsyncer.listener.DefaultExtractor;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.extractor.MysqlExtractor;
 
@@ -16,15 +15,15 @@ public enum ListenerEnum {
     /**
      * Mysql
      */
-    MYSQL(ConnectorEnum.MYSQL.getType(), new MysqlExtractor()),
+    MYSQL(ConnectorEnum.MYSQL.getType(), MysqlExtractor.class),
     ;
 
     private String type;
-    private DefaultExtractor extractor;
+    private Class<?> clazz;
 
-    ListenerEnum(String type, DefaultExtractor extractor) {
+    ListenerEnum(String type, Class<?> clazz) {
         this.type = type;
-        this.extractor = extractor;
+        this.clazz = clazz;
     }
 
     /**
@@ -34,10 +33,10 @@ public enum ListenerEnum {
      * @return
      * @throws ListenerException
      */
-    public static DefaultExtractor getExtractor(String type) throws ListenerException {
+    public static Class<?> getExtractor(String type) throws ListenerException {
         for (ListenerEnum e : ListenerEnum.values()) {
             if (StringUtils.equals(type, e.getType())) {
-                return e.getExtractor();
+                return e.getClazz();
             }
         }
         throw new ListenerException(String.format("Extractor type \"%s\" does not exist.", type));
@@ -47,7 +46,7 @@ public enum ListenerEnum {
         return type;
     }
 
-    public DefaultExtractor getExtractor() {
-        return extractor;
+    public Class<?> getClazz() {
+        return clazz;
     }
-}
+}

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

@@ -1,14 +1,13 @@
 package org.dbsyncer.listener.extractor;
 
+import org.apache.commons.lang.RandomStringUtils;
 import org.apache.commons.lang.StringUtils;
 import org.dbsyncer.connector.config.DatabaseConfig;
 import org.dbsyncer.connector.constant.ConnectorConstant;
 import org.dbsyncer.listener.DefaultExtractor;
 import org.dbsyncer.listener.ListenerException;
 import org.dbsyncer.listener.config.Host;
-import org.dbsyncer.listener.mysql.binlog.BinlogEventListener;
-import org.dbsyncer.listener.mysql.binlog.BinlogEventV4;
-import org.dbsyncer.listener.mysql.binlog.BinlogRemoteClient;
+import org.dbsyncer.listener.mysql.binlog.*;
 import org.dbsyncer.listener.mysql.binlog.impl.event.*;
 import org.dbsyncer.listener.mysql.common.glossary.Column;
 import org.dbsyncer.listener.mysql.common.glossary.Pair;
@@ -48,7 +47,11 @@ public class MysqlExtractor extends DefaultExtractor {
             final Host host = cluster.get(master);
             final String username = config.getUsername();
             final String password = config.getPassword();
-            final String threadSuffixName = "mysql-binlog";
+            // mysql-binlog-127.0.0.1:3306-654321
+            final String threadSuffixName = new StringBuilder("mysql-binlog-")
+                    .append(host.getIp()).append(":").append(host.getPort()).append("-")
+                    .append(RandomStringUtils.randomNumeric(6))
+                    .toString();
 
             client = new BinlogRemoteClient(host.getIp(), host.getPort(), username, password, threadSuffixName);
             client.setBinlogFileName(map.get(BINLOG_FILENAME));
@@ -107,14 +110,18 @@ public class MysqlExtractor extends DefaultExtractor {
     private void refresh(AbstractBinlogEventV4 event) {
         String binlogFilename = event.getBinlogFilename();
         long nextPosition = event.getHeader().getNextPosition();
-        if (!StringUtils.equals(binlogFilename, client.getBinlogFileName()) || 0 != Long.compare(nextPosition,
-                client.getBinlogPosition())) {
+
+        // binlogFileName
+        if (StringUtils.isNotBlank(binlogFilename) && !StringUtils.equals(binlogFilename, client.getBinlogFileName())) {
             client.setBinlogFileName(binlogFilename);
-            client.setBinlogPosition(nextPosition);
-            map.put(BINLOG_FILENAME, client.getBinlogFileName());
-            map.put(BINLOG_POSITION, String.valueOf(nextPosition));
-            flushEvent();
         }
+        client.setBinlogPosition(nextPosition);
+
+        // nextPosition
+        logger.info("{}:{}", client.getBinlogFileName(), client.getBinlogPosition());
+        map.put(BINLOG_FILENAME, client.getBinlogFileName());
+        map.put(BINLOG_POSITION, String.valueOf(client.getBinlogPosition()));
+        flushEvent();
     }
 
     final class MysqlEventListener implements BinlogEventListener {

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

@@ -0,0 +1,79 @@
+package org.dbsyncer.manager.config;
+
+import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.connector.config.Field;
+import org.dbsyncer.parser.model.FieldMapping;
+import org.dbsyncer.parser.model.TableGroup;
+import org.springframework.util.Assert;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+public class FieldPicker {
+
+    private TableGroup tableGroup;
+    private List<Field> column;
+    private List<FieldMapping> fieldMapping;
+    private List<Node> index;
+    private int indexSize;
+
+    public FieldPicker(TableGroup tableGroup, List<Field> column, List<FieldMapping> fieldMapping) {
+        this.tableGroup = tableGroup;
+        this.column = column;
+        this.fieldMapping = fieldMapping;
+        init();
+    }
+
+    public Map<String, Object> getColumns(List<Object> list) {
+        if (!CollectionUtils.isEmpty(list)) {
+            Map<String, Object> data = new HashMap<>(indexSize);
+            int size = list.size();
+            index.parallelStream().forEach(node -> {
+                if (node.i <= size) {
+                    data.put(node.name, list.get(node.i));
+                }
+            });
+            return data;
+        }
+        return Collections.EMPTY_MAP;
+    }
+
+    public TableGroup getTableGroup() {
+        return tableGroup;
+    }
+
+    private void init() {
+        // column  => [1, 86, 0, 中文, 2020-05-15T12:17:22.000+0800, 备注信息]
+        Assert.notEmpty(column, "读取字段不能为空.");
+        Assert.notEmpty(fieldMapping, "映射关系不能为空.");
+
+        // 找到同步字段 => [{source.name}]
+        Set<String> key = fieldMapping.stream().map(m -> m.getSource().getName()).collect(Collectors.toSet());
+
+        // 记录字段索引 [{"ID":0},{"NAME":1}]
+        index = new LinkedList<>();
+        int size = column.size();
+        String k = null;
+        for (int i = 0; i < size; i++) {
+            k = column.get(i).getName();
+            if (key.contains(k)) {
+                index.add(new Node(k, i));
+            }
+        }
+        Assert.notEmpty(index, "同步映射关系不能为空.");
+        this.indexSize = index.size();
+    }
+
+    final class Node {
+        // 属性
+        String name;
+        // 索引
+        int i;
+
+        public Node(String name, int i) {
+            this.name = name;
+            this.i = i;
+        }
+    }
+
+}

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

@@ -1,24 +1,23 @@
 package org.dbsyncer.manager.puller.impl;
 
 import org.dbsyncer.common.event.Event;
+import org.dbsyncer.common.util.CollectionUtils;
+import org.dbsyncer.connector.config.Table;
 import org.dbsyncer.listener.DefaultExtractor;
-import org.dbsyncer.listener.Extractor;
 import org.dbsyncer.listener.Listener;
-import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.manager.Manager;
+import org.dbsyncer.manager.config.FieldPicker;
 import org.dbsyncer.manager.puller.AbstractPuller;
-import org.dbsyncer.manager.puller.Increment;
 import org.dbsyncer.parser.Parser;
-import org.dbsyncer.parser.model.Connector;
-import org.dbsyncer.parser.model.Mapping;
-import org.dbsyncer.parser.model.Meta;
-import org.dbsyncer.parser.model.TableGroup;
+import org.dbsyncer.parser.model.*;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -54,11 +53,15 @@ public class IncrementPuller extends AbstractPuller {
             Connector connector = manager.getConnector(mapping.getSourceConnectorId());
             Assert.notNull(connector, "连接器不能为空.");
             List<TableGroup> list = manager.getTableGroupAll(mappingId);
-            Assert.notEmpty(list, "映射关系不能为空");
+            Assert.notEmpty(list, "映射关系不能为空.");
             Meta meta = manager.getMeta(metaId);
             Assert.notNull(meta, "Meta不能为空.");
             DefaultExtractor extractor = listener.createExtractor(connector.getConfig(), mapping.getListener(), meta.getMap());
             Assert.notNull(extractor, "未知的监听配置.");
+            long now = System.currentTimeMillis();
+            meta.setBeginTime(now);
+            meta.setEndTime(now);
+            manager.editMeta(meta);
 
             // 监听数据变更事件
             extractor.addListener(new DefaultListener(mapping, list));
@@ -75,23 +78,12 @@ public class IncrementPuller extends AbstractPuller {
 
     @Override
     public void close(String metaId) {
-        Extractor extractor = map.get(metaId);
+        DefaultExtractor extractor = map.get(metaId);
         if (null != extractor) {
+            extractor.clearAllListener();
             extractor.close();
-        }
-    }
-
-    /**
-     * TODO 更新待优化，存在性能问题
-     *
-     * @param metaId
-     */
-    private void flush(String metaId) {
-        Meta meta = manager.getMeta(metaId);
-        DefaultExtractor extractor = map.get(metaId);
-        if (null != meta && null != extractor) {
-            meta.setMap(extractor.getMap());
-            manager.editMeta(meta);
+            finished(metaId);
+            logger.info("关闭成功:{}", metaId);
         }
     }
 
@@ -104,23 +96,49 @@ public class IncrementPuller extends AbstractPuller {
 
         private Mapping mapping;
         private List<TableGroup> list;
+        private String metaId;
+        private Map<String, List<FieldPicker>> tablePicker;
 
         public DefaultListener(Mapping mapping, List<TableGroup> list) {
             this.mapping = mapping;
             this.list = list;
+            this.metaId = mapping.getMetaId();
+            this.tablePicker = new LinkedHashMap<>();
+            list.forEach(t -> {
+                final Table table = t.getSourceTable();
+                final String tableName = table.getName();
+                tablePicker.putIfAbsent(tableName, new ArrayList<>());
+                tablePicker.get(tableName).add(new FieldPicker(t, table.getColumn(), t.getFieldMapping()));
+            });
         }
 
         @Override
         public void changedEvent(String tableName, String event, List<Object> before, List<Object> after) {
-            logger.info("监听数据>tableName:{},event:{},after:{}, after:{}", tableName, event, before, after);
+            logger.info("监听数据=> tableName:{}, event:{}, before:{}, after:{}", tableName, event, before, after);
+
             // 处理过程有异常向上抛
-            list.forEach(tableGroup -> parser.execute(mapping, tableGroup));
+            List<FieldPicker> pickers = tablePicker.get(tableName);
+            if (!CollectionUtils.isEmpty(pickers)) {
+                pickers.parallelStream().forEach(p -> {
+                    DataEvent data = new DataEvent(event, p.getColumns(before), p.getColumns(after));
+                    parser.execute(mapping, p.getTableGroup(), data);
+                });
+            }
+
         }
 
         @Override
         public void flushEvent() {
-            logger.info("flushEvent");
-            flush(mapping.getMetaId());
+            // TODO 更新待优化，存在性能问题
+            DefaultExtractor extractor = map.get(metaId);
+            if (null != extractor) {
+                logger.info("flushEvent map:{}", extractor.getMap());
+                Meta meta = manager.getMeta(metaId);
+                if (null != meta) {
+                    meta.setMap(extractor.getMap());
+                    manager.editMeta(meta);
+                }
+            }
         }
 
     }

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

@@ -8,6 +8,7 @@ import org.dbsyncer.connector.enums.FilterEnum;
 import org.dbsyncer.connector.enums.OperationEnum;
 import org.dbsyncer.parser.enums.ConvertEnum;
 import org.dbsyncer.parser.model.Connector;
+import org.dbsyncer.parser.model.DataEvent;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.TableGroup;
 
@@ -124,6 +125,7 @@ public interface Parser {
      *
      * @param mapping
      * @param tableGroup
+     * @param dataEvent
      */
-    void execute(Mapping mapping, TableGroup tableGroup);
+    void execute(Mapping mapping, TableGroup tableGroup, DataEvent dataEvent);
 }

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

@@ -171,14 +171,13 @@ public class ParserFactory implements Parser {
         String sTableName = tableGroup.getSourceTable().getName();
         String tTableName = tableGroup.getTargetTable().getName();
         Assert.notEmpty(fieldMapping, String.format("数据源表[%s]同步到目标源表[%s], 映射关系不能为空.", sTableName, tTableName));
-        // 获取同步字段
-        Picker picker = new Picker();
-        PickerUtil.pickFields(picker, fieldMapping);
-
         // 转换配置(默认使用全局)
         List<Convert> convert = CollectionUtils.isEmpty(tableGroup.getConvert()) ? mapping.getConvert() : tableGroup.getConvert();
         // 插件配置(默认使用全局)
         Plugin plugin = null == tableGroup.getPlugin() ? mapping.getPlugin() : tableGroup.getPlugin();
+        // 获取同步字段
+        Picker picker = new Picker();
+        PickerUtil.pickFields(picker, fieldMapping);
 
         // 检查分页参数
         Map<String, String> params = getMeta(metaId).getMap();
@@ -207,14 +206,14 @@ public class ParserFactory implements Parser {
             PickerUtil.pickData(picker, data);
 
             // 3、参数转换
-            List<Map<String, Object>> target = picker.getTarget();
+            List<Map<String, Object>> target = picker.getTargetList();
             ConvertUtil.convert(convert, target);
 
             // 4、插件转换
             pluginFactory.convert(plugin, data, target);
 
             // 5、写入目标源
-            Result writer = executeBatch(tConfig, command, picker.getTargetFields(), target, threadSize, batchSize);
+            Result writer = writeBatch(tConfig, command, picker.getTargetFields(), target, threadSize, batchSize);
 
             // 6、更新结果
             flush(task, writer, target.size());
@@ -225,8 +224,38 @@ public class ParserFactory implements Parser {
     }
 
     @Override
-    public void execute(Mapping mapping, TableGroup tableGroup) {
+    public void execute(Mapping mapping, TableGroup tableGroup, DataEvent dataEvent) {
+        logger.info("同步数据=> dataEvent:{}", dataEvent);
+        final String metaId = mapping.getMetaId();
 
+        ConnectorConfig tConfig = getConnectorConfig(mapping.getTargetConnectorId());
+        Map<String, String> command = tableGroup.getCommand();
+        List<FieldMapping> fieldMapping = tableGroup.getFieldMapping();
+        // 转换配置(默认使用全局)
+        List<Convert> convert = CollectionUtils.isEmpty(tableGroup.getConvert()) ? mapping.getConvert() : tableGroup.getConvert();
+        // 插件配置(默认使用全局)
+        Plugin plugin = null == tableGroup.getPlugin() ? mapping.getPlugin() : tableGroup.getPlugin();
+        // 获取同步字段
+        Picker picker = new Picker();
+        PickerUtil.pickFields(picker, fieldMapping);
+
+        // 1、映射字段
+        String event = dataEvent.getEvent();
+        Map<String, Object> data = dataEvent.getData();
+        PickerUtil.pickData(picker, data);
+
+        // 2、参数转换
+        Map<String, Object> target = picker.getTarget();
+        ConvertUtil.convert(convert, target);
+
+        // 3、插件转换
+        pluginFactory.convert(plugin, event, data, target);
+
+        // 4、写入目标源
+        Result writer = connectorFactory.writer(tConfig, picker.getTargetFields(), command, event, target);
+
+        // 5、更新结果
+        flush(metaId, writer, 1);
     }
 
     /**
@@ -237,20 +266,23 @@ public class ParserFactory implements Parser {
      * @param total
      */
     private void flush(Task task, Result writer, long total) {
+        flush(task.getId(), writer, total);
+
+        // 发布刷新事件给FullExtractor
+        task.setEndTime(System.currentTimeMillis());
+        applicationContext.publishEvent(new FullRefreshEvent(applicationContext, task));
+    }
+
+    private void flush(String metaId, Result writer, long total) {
         // 引用传递
         long fail = writer.getFail().get();
-        long success = total - fail;
-        Meta meta = getMeta(task.getId());
+        Meta meta = getMeta(metaId);
         meta.getFail().getAndAdd(fail);
-        meta.getSuccess().getAndAdd(success);
+        meta.getSuccess().getAndAdd(total - fail);
         // print process
-        logger.info("任务:{}, 成功:{}, 失败:{}", task.getId(), meta.getSuccess(), meta.getFail());
+        logger.info("任务:{}, 成功:{}, 失败:{}", metaId, meta.getSuccess(), meta.getFail());
 
         // TODO 记录错误日志
-
-        // 发布刷新事件给FullExtractor
-        task.setEndTime(System.currentTimeMillis());
-        applicationContext.publishEvent(new FullRefreshEvent(applicationContext, task));
     }
 
     /**
@@ -292,7 +324,7 @@ public class ParserFactory implements Parser {
      * @param batchSize
      * @return
      */
-    private Result executeBatch(ConnectorConfig config, Map<String, String> command, List<Field> fields, List<Map<String, Object>> target,
+    private Result writeBatch(ConnectorConfig config, Map<String, String> command, List<Field> fields, List<Map<String, Object>> target,
                                 int threadSize, int batchSize) {
         // 总数
         int total = target.size();

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

@@ -22,11 +22,11 @@ public enum ModelEnum {
     INCREMENT("increment", "增量");
 
     private String code;
-    private String message;
+    private String name;
 
-    ModelEnum(String code, String message) {
+    ModelEnum(String code, String name) {
         this.code = code;
-        this.message = message;
+        this.name = name;
     }
 
     public static ModelEnum getModelEnum(String code) throws ParserException {
@@ -50,11 +50,11 @@ public enum ModelEnum {
         this.code = code;
     }
 
-    public String getMessage() {
-        return message;
+    public String getName() {
+        return name;
     }
 
-    public void setMessage(String message) {
-        this.message = message;
+    public void setName(String name) {
+        this.name = name;
     }
 }

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/DataEvent.java

@@ -0,0 +1,54 @@
+package org.dbsyncer.parser.model;
+
+import org.apache.commons.lang.StringUtils;
+import org.dbsyncer.common.util.JsonUtil;
+import org.dbsyncer.connector.constant.ConnectorConstant;
+
+import java.util.Map;
+
+public final class DataEvent {
+
+    private String event;
+    private Map<String, Object> before;
+    private Map<String, Object> after;
+
+    public DataEvent(String event, Map<String, Object> before, Map<String, Object> after) {
+        this.event = event;
+        this.before = before;
+        this.after = after;
+    }
+
+    public Map<String, Object> getData() {
+        return StringUtils.equals(ConnectorConstant.OPERTION_DELETE, event) ? before : after;
+    }
+
+    public String getEvent() {
+        return event;
+    }
+
+    public void setEvent(String event) {
+        this.event = event;
+    }
+
+    public Map<String, Object> getBefore() {
+        return before;
+    }
+
+    public void setBefore(Map<String, Object> before) {
+        this.before = before;
+    }
+
+    public Map<String, Object> getAfter() {
+        return after;
+    }
+
+    public void setAfter(Map<String, Object> after) {
+        this.after = after;
+    }
+
+    @Override
+    public String toString() {
+        return JsonUtil.objToJson(this);
+    }
+
+}

dbsyncer-parser/src/main/java/org/dbsyncer/parser/model/Picker.java

@@ -9,7 +9,8 @@ public class Picker {
 
     private List<Field> sourceFields;
     private List<Field> targetFields;
-    private List<Map<String, Object>> target;
+    private List<Map<String, Object>> targetList;
+    private Map<String, Object> target;
 
     public List<Field> getSourceFields() {
         return sourceFields;
@@ -27,11 +28,19 @@ public class Picker {
         this.targetFields = targetFields;
     }
 
-    public List<Map<String, Object>> getTarget() {
+    public List<Map<String, Object>> getTargetList() {
+        return targetList;
+    }
+
+    public void setTargetList(List<Map<String, Object>> targetList) {
+        this.targetList = targetList;
+    }
+
+    public Map<String, Object> getTarget() {
         return target;
     }
 
-    public void setTarget(List<Map<String, Object>> target) {
+    public void setTarget(Map<String, Object> target) {
         this.target = target;
     }
 }

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

@@ -21,25 +21,37 @@ public abstract class ConvertUtil {
     public static void convert(List<Convert> convert, List<Map<String, Object>> data) {
         if (!CollectionUtils.isEmpty(convert) && !CollectionUtils.isEmpty(data)) {
             // 并行流计算
-            final int size = convert.size();
             data.parallelStream().forEach(row -> {
-                // 替换row值, 复用堆栈地址，减少开销
-                Convert c = null;
-                String name = null;
-                String code = null;
-                String args = null;
-                Object value = null;
-                for (int i = 0; i < size; i++) {
-                    c = convert.get(i);
-                    name = c.getName();
-                    code = c.getConvertCode();
-                    args = c.getArgs();
-                    value = ConvertEnum.getHandler(code).handle(args, row.get(name));
-
-                    row.put(name, value);
-                }
+                convert(convert, row);
             });
         }
     }
 
+    /**
+     * 转换参数
+     *
+     * @param convert
+     * @param row
+     */
+    public static void convert(List<Convert> convert, Map<String, Object> row) {
+        if (!CollectionUtils.isEmpty(convert) && !CollectionUtils.isEmpty(row)) {
+            // 替换row值, 复用堆栈地址，减少开销
+            final int size = convert.size();
+            Convert c = null;
+            String name = null;
+            String code = null;
+            String args = null;
+            Object value = null;
+            for (int i = 0; i < size; i++) {
+                c = convert.get(i);
+                name = c.getName();
+                code = c.getConvertCode();
+                args = c.getArgs();
+                value = ConvertEnum.getHandler(code).handle(args, row.get(name));
+
+                row.put(name, value);
+            }
+        }
+    }
+
 }

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

@@ -55,6 +55,28 @@ public abstract class PickerUtil {
                 target.add(r);
             }
 
+            picker.setTargetList(target);
+        }
+    }
+
+    public static void pickData(Picker picker, Map<String, Object> row) {
+        if(!CollectionUtils.isEmpty(row)){
+            Map<String, Object> target = new HashMap<>();
+            List<Field> sFields = picker.getSourceFields();
+            List<Field> tFields = picker.getTargetFields();
+
+            final int kSize = sFields.size();
+            String sName = null;
+            String tName = null;
+            Object v = null;
+            for (int k = 0; k < kSize; k++) {
+                sName = sFields.get(k).getName();
+                v = row.get(sName);
+
+                tName = tFields.get(k).getName();
+                target.put(tName, v);
+            }
+
             picker.setTarget(target);
         }
     }

dbsyncer-plugin/src/main/java/org/dbsyncer/plugin/PluginFactory.java

@@ -29,4 +29,9 @@ public class PluginFactory {
             // TODO 插件转换
         }
     }
+
+    public void convert(Plugin plugin, String event, Map<String, Object> source, Map<String, Object> target) {
+        if (null != plugin) {
+        }
+    }
 }

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

@@ -24,7 +24,6 @@ public class IndexController {
     public String index(HttpServletRequest request, ModelMap model) {
         model.put("connectors", connectorService.getConnectorAll());
         model.put("mappings", mappingService.getMappingAll());
-        model.put("metas", mappingService.getMetaAll());
         return "index/index.html";
     }
 

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

@@ -4,7 +4,7 @@
 .connectorList .well-sign-operation{color: #999999; border-radius: 50%; }
 .connectorList .well-sign-operation:hover {color: #333333;}
 .connectorList .dropdown {position: absolute; right:16px; top: 0px;}
-.connectorList .dropdown-menu {right: -86px;left: auto; top: 15px;}
+.connectorList .dropdown-menu {left: auto; top: 15px;}
 .connectorList .dropdown-menu i{font-size: 20px;}
 
 .mappingList .jumbotron .line{margin: 43px -30px;width: 188px;border-top: 1px solid #8c7c7c;}
@@ -20,5 +20,5 @@
 .mappingList .well-sign-operation{position: absolute; right:12px; color: #999999; border-radius: 50%;}
 .mappingList .well-sign-operation:hover { color: #333333;}
 .mappingList .dropdown {position: absolute; right:0px; top: 0px;}
-.mappingList .dropdown-menu {right: -70px;left: auto; top: 28px;}
+.mappingList .dropdown-menu {right: 12px;left: auto; top: 28px;}
 .mappingList .dropdown-menu i{font-size: 20px;}

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

@@ -5,7 +5,9 @@
 <!-- Connector Mapping -->
 <div class="container-fluid">
     <div class="row">
-        <div class="col-md-4">
+
+        <!-- 连接器管理 -->
+        <div class="col-md-12">
             <form class="form-horizontal" role="form" method="post">
                 <!-- 连接器开始位置 -->
                 <div class="form-group">
@@ -23,7 +25,7 @@
                             <div class="panel-body text-center">
                                 <div class="row connectorList">
                                     <!-- 连接器__开始 -->
-                                    <div class="col-md-3" th:each="c,state : ${connectors}">
+                                    <div class="col-md-1" th:each="c,state : ${connectors}">
                                         <div th:id="${c?.id}" class="jumbotron dbsyncer_block">
                                             <div class="row">
                                                 <img th:src="@{'/images/'+ ${c?.config?.connectorType} + '.png'}">
@@ -49,7 +51,12 @@
                     </div>
                 </div>
                 <!-- 连接器开结束位置 -->
+            </form>
+        </div>
 
+        <!-- 驱动管理 -->
+        <div class="col-md-12">
+            <form class="form-horizontal" role="form" method="post">
                 <!-- 驱动开始位置 -->
                 <div class="form-group" th:if="${connectors?.size() gt 0}">
                     <div class="col-md-12">
@@ -66,7 +73,7 @@
                             <div class="panel-body">
                                 <div class="row mappingList">
                                     <!-- 驱动__开始 -->
-                                    <div class="col-md-12" th:each="m,state : ${mappings}">
+                                    <div class="col-md-4" th:each="m,state : ${mappings}">
                                         <div th:id="${m?.id}" class="jumbotron dbsyncer_block">
                                             <!--驱动标题信息 -->
                                             <div class="row text-center" th:text="${m?.name}" th:title="${m?.name}"></div>
@@ -83,15 +90,17 @@
                                                         </div>
                                                         <div class="col-md-1"></div>
                                                     </div>
-                                                    <span th:if="${m?.sourceConnector?.running}" class="well-sign-left"><i class="fa fa-2x fa-circle well-sign-green"></i></span>
+                                                    <span th:if="${m?.sourceConnector?.running}" th:title="连接器正常" class="well-sign-left"><i class="fa fa-2x fa-circle well-sign-green"></i></span>
                                                     <span th:unless="${m?.sourceConnector?.running}" th:title="连接器异常" class="well-sign-left"><i class="fa fa-2x fa-times-circle-o well-sign-red"></i></span>
                                                 </div>
 
                                                 <!--中间图标 -->
                                                 <div class="col-md-2">
                                                     <div class="line">
-                                                        <span th:if="${m?.running}" class="running-through-rate well-sign-green">✔</span>
-                                                        <span class="running-state">[[${m?.running}?'运行中':'未运行']]</span>
+                                                        <span th:if="${m?.meta?.state eq 1}" class="running-through-rate well-sign-green">✔</span>
+                                                        <span th:if="${m?.meta?.state eq 0}" class="running-state label label-info">未运行</span>
+                                                        <span th:if="${m?.meta?.state eq 1}" class="running-state label label-success">运行中</span>
+                                                        <span th:if="${m?.meta?.state eq 2}" class="running-state label label-warning">停止中</span>
                                                     </div>
                                                 </div>
 
@@ -105,12 +114,40 @@
                                                             <span th:text="${m?.targetConnector?.name}" th:title="${m?.targetConnector?.name}"></span>
                                                         </div>
                                                         <div class="col-md-1"></div>
-                                                        <span th:if="${m?.targetConnector?.running}" class="well-sign-right"><i class="fa fa-2x fa-circle well-sign-green"></i></span>
+                                                        <span th:if="${m?.targetConnector?.running}" th:title="连接器正常" class="well-sign-right"><i class="fa fa-2x fa-circle well-sign-green"></i></span>
                                                         <span th:unless="${m?.targetConnector?.running}" th:title="连接器异常" class="well-sign-right"><i class="fa fa-2x fa-times-circle-o well-sign-red"></i></span>
                                                     </div>
                                                 </div>
                                             </div>
 
+                                            <div class="row">
+                                                <table class="table table-hover">
+                                                    <tbody>
+                                                    <tr>
+                                                        <td>同步方式: <span th:text="${m?.meta?.model}"></span></td>
+                                                    </tr>
+                                                    <tr>
+                                                        <td class="text-left">
+                                                            同步结果:
+                                                            总数:[[${m?.meta?.total}]]
+                                                            <span th:if="${m?.meta?.model eq '全量' and (m?.meta?.success + m?.meta?.fail) gt 0}">
+                                                                        ,进度:[[${#numbers.formatDecimal(((m?.meta?.success + m?.meta?.fail) / m?.meta?.total * 100.00),0 ,2)}]]%
+                                                                        ,耗时:[[${(m?.meta?.endTime - m?.meta?.beginTime) / 1000}]]秒
+                                                                    </span>
+                                                            <span th:if="${m?.meta?.success gt 0}">,成功:[[${m?.meta?.success}]]</span>
+                                                            <span th:if="${m?.meta?.fail gt 0}">,失败:[[${m?.meta?.fail}]] <a href="javascript:;" class="label label-danger">日志</a></span>
+                                                        </td>
+                                                    </tr>
+                                                    <tr>
+                                                        <td class="text-left">
+                                                            启动时间:
+                                                            <span th:if="${m?.meta?.state != 0 and m?.meta?.beginTime gt 0}">[[${#dates.format(m?.meta?.beginTime, 'yyyy-MM-dd HH:mm:ss')}]]</span>
+                                                        </td>
+                                                    </tr>
+                                                    </tbody>
+                                                </table>
+                                            </div>
+
                                             <div class="row text-right text-muted" th:text="${#dates.format(m?.updateTime, 'yyyy-MM-dd HH:mm:ss')}"></div>
                                         </div>
 
@@ -119,11 +156,11 @@
                                             <a data-toggle="dropdown" href="javascript:;"><span class="well-sign-operation"><i class="fa fa-gears fa-2x"></i></span></a>
                                             <ul class="dropdown-menu" role="menu" aria-labelledby="dLabel">
                                                 <!-- 未运行 -->
-                                                <li th:if="${not m?.running}" th:url="'/mapping/start?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-check-circle-o well-sign-green"></i>&nbsp;&nbsp;启动</a></li>
+                                                <li th:if="${m?.meta?.state ne 1}" th:url="'/mapping/start?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-check-circle-o well-sign-green"></i>&nbsp;&nbsp;启动</a></li>
                                                 <!-- 运行中 -->
-                                                <li th:if="${m?.running}" th:url="'/mapping/stop?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-times-circle-o well-sign-red"></i>&nbsp;&nbsp;停止</a></li>
+                                                <li th:if="${m?.meta?.state eq 1}" th:url="'/mapping/stop?id='+${m?.id}"><a href="javascript:;"><i class="fa fa-times-circle-o well-sign-red"></i>&nbsp;&nbsp;停止</a></li>
                                                 <!-- 未运行 -->
-                                                <li th:if="${not m?.running}" th:url="'/mapping/remove?id='+${m?.id}" confirm="true" confirmMessage="确认删除?"><a href="javascript:;"><i class="fa fa-trash well-sign-red"></i>&nbsp;&nbsp;删除</a></li>
+                                                <li th:if="${m?.meta?.state ne 1}" th:url="'/mapping/remove?id='+${m?.id}" confirm="true" confirmMessage="确认删除?"><a href="javascript:;"><i class="fa fa-trash well-sign-red"></i>&nbsp;&nbsp;删除</a></li>
                                             </ul>
                                         </div>
                                     </div>
@@ -138,45 +175,6 @@
             </form>
         </div>
 
-        <!-- 实时数据 -->
-        <div class="col-md-8">
-            <table th:if="${metas?.size() gt 0}" class="table table-hover">
-                <caption>驱动运行监控</caption>
-                <thead>
-                <tr>
-                    <th>序号</th>
-                    <th>任务名称</th>
-                    <th>同步方式</th>
-                    <th>结果</th>
-                    <th>状态</th>
-                    <th>启动时间</th>
-                </tr>
-                </thead>
-                <tbody>
-                <tr th:each="m,state : ${metas}">
-                    <td th:text="${state?.index + 1}"></td>
-                    <td th:text="${m?.mappingName}"></td>
-                    <td th:text="${m?.model}"></td>
-                    <td>
-                        总数:[[${m?.total}]]
-                        <span th:if="${m?.model eq '全量' and (m?.success + m?.fail) gt 0}">
-                            ,进度:[[${#numbers.formatDecimal(((m?.success + m?.fail) / m?.total * 100.00),0 ,2)}]]%
-                            ,耗时:[[${(m?.endTime - m?.beginTime) / 1000}]]秒
-                        </span>
-                        <span th:if="${m?.success gt 0}">,成功:[[${m?.success}]]</span>
-                        <span th:if="${m?.fail gt 0}">,失败:[[${m?.fail}]] <a href="javascript:;" class="label label-danger">日志</a></span>
-                    </td>
-                    <td>
-                        <span th:if="${m?.state eq 0}" class="label label-info">未运行</span>
-                        <span th:if="${m?.state eq 1}" class="label label-success">运行中</span>
-                        <span th:if="${m?.state eq 2}" class="label label-warning">停止中</span>
-                    </td>
-                    <td><span th:if="${m?.state != 0 and m?.beginTime gt 0}">[[${#dates.format(m?.beginTime, 'yyyy-MM-dd HH:mm:ss')}]]</span></td>
-                </tr>
-                </tbody>
-            </table>
-        </div>
-
     </div>
 </div>