add pck

dbsyncer-listener/pom.xml

@@ -19,17 +19,17 @@
             <version>${project.parent.version}</version>
         </dependency>
 
-        <!-- Message 消息服务 -->
+        <!-- Cache 缓存服务 -->
         <dependency>
             <groupId>org.ghi</groupId>
-            <artifactId>dbsyncer-message</artifactId>
+            <artifactId>dbsyncer-cache</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
 
-        <!-- Schedule 定时服务 -->
+        <!-- Message 消息服务 -->
         <dependency>
             <groupId>org.ghi</groupId>
-            <artifactId>dbsyncer-schedule</artifactId>
+            <artifactId>dbsyncer-message</artifactId>
             <version>${project.parent.version}</version>
         </dependency>
 

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

@@ -1,6 +1,6 @@
 package org.dbsyncer.listener;
 
 
-public abstract class AbstractListener implements LifecycleListener {
+public abstract class AbstractListener implements Listener {
 
 }

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

@@ -1,25 +0,0 @@
-package org.dbsyncer.listener;
-
-/**
- * <h3>监听驱动停止</h3>
- * <ol type="1">
- * <li><dt>onClose</dt></li>
- * <dd>当驱动停止时会触发该事件</dd>
- * </ol>
- *
- * @ClassName: Listener
- * @author: AE86
- * @date: 2018年7月30日 上午10:58:54
- */
-public interface LifecycleListener extends Runnable {
-
-    /**
-     * 驱动停止
-     *
-     * @Title: onClose
-     * @Description: 监听驱动停止事件
-     * @return: void
-     */
-    void onClose();
-
-}

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

@@ -1,10 +1,28 @@
 package org.dbsyncer.listener;
 
 
+import org.dbsyncer.common.task.Task;
+import org.dbsyncer.connector.config.ConnectorConfig;
+import org.dbsyncer.listener.config.ListenerConfig;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Component;
 
+import java.util.Map;
+
 @Component
-public class ListenerFactory implements Listener {
+public class ListenerFactory implements ApplicationContextAware {
+
+    private Map<String, Listener> pull;
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        pull = applicationContext.getBeansOfType(Listener.class);
+    }
+
+    public void execute(Task task, ListenerConfig listenerConfig, ConnectorConfig connector) {
 
 
+    }
 }

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

@@ -0,0 +1,15 @@
+package org.dbsyncer.listener;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * 监听日志同步
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2020/05/05 15:28
+ */
+@Component
+public class LogListener extends AbstractListener {
+
+}

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

@@ -0,0 +1,15 @@
+package org.dbsyncer.listener;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * 定时同步
+ *
+ * @author AE86
+ * @version 1.0.0
+ * @date 2020/05/05 15:28
+ */
+@Component
+public class TimingListener extends AbstractListener {
+
+}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/Increment.java

@@ -1,6 +0,0 @@
-package org.dbsyncer.manager.extractor;
-
-public interface Increment {
-
-    void close(String metaId);
-}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/impl/IncrementExtractor.java

@@ -1,15 +1,13 @@
 package org.dbsyncer.manager.extractor.impl;
 
 import org.dbsyncer.common.task.Task;
-import org.dbsyncer.common.util.StringUtil;
-import org.dbsyncer.listener.Listener;
+import org.dbsyncer.connector.config.ConnectorConfig;
+import org.dbsyncer.listener.ListenerFactory;
 import org.dbsyncer.listener.config.ListenerConfig;
 import org.dbsyncer.manager.Manager;
 import org.dbsyncer.manager.extractor.AbstractExtractor;
-import org.dbsyncer.manager.extractor.Increment;
 import org.dbsyncer.parser.model.Connector;
 import org.dbsyncer.parser.model.Mapping;
-import org.dbsyncer.parser.model.TableGroup;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.BeansException;
@@ -19,7 +17,6 @@ import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
 
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -31,43 +28,32 @@ import java.util.concurrent.ConcurrentHashMap;
  * @date 2020/04/26 15:28
  */
 @Component
-public class IncrementExtractor extends AbstractExtractor implements ApplicationContextAware {
+public class IncrementExtractor extends AbstractExtractor {
 
     private final Logger logger = LoggerFactory.getLogger(getClass());
 
     @Autowired
-    private Listener listener;
+    private ListenerFactory listenerFactory;
 
     @Autowired
     private Manager manager;
 
-    private Map<String, Increment> pull;
-
     private Map<String, Task> map = new ConcurrentHashMap<>();
 
-    @Override
-    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        pull = applicationContext.getBeansOfType(Increment.class);
-    }
-
     @Override
     public void asyncStart(Mapping mapping) {
         final String metaId = mapping.getMetaId();
         map.putIfAbsent(metaId, new Task(metaId));
 
         try {
-            ListenerConfig listener = mapping.getListener();
-            // log / timing
-            String type = listener.getListenerType() + "Increment";
-            Increment increment = this.pull.get(type);
-            Assert.notNull(increment, "未知的同步方式.");
-
+            ListenerConfig listenerConfig = mapping.getListener();
             Connector connector = manager.getConnector(mapping.getSourceConnectorId());
             Assert.notNull(connector, "连接器不能为空.");
 
             // 执行任务
             logger.info("启动任务:{}", metaId);
             Task task = map.get(metaId);
+            listenerFactory.execute(task, listenerConfig, connector.getConfig());
 
         } catch (Exception e) {
             // TODO 记录错误日志

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/increment/LogIncrement.java

@@ -1,20 +0,0 @@
-package org.dbsyncer.manager.extractor.increment;
-
-import org.dbsyncer.manager.extractor.Increment;
-import org.springframework.stereotype.Component;
-
-/**
- * 监听日志同步
- *
- * @author AE86
- * @version 1.0.0
- * @date 2020/05/05 15:28
- */
-@Component
-public class LogIncrement implements Increment {
-
-    @Override
-    public void close(String metaId) {
-
-    }
-}

dbsyncer-manager/src/main/java/org/dbsyncer/manager/extractor/increment/TimingIncrement.java

@@ -1,20 +0,0 @@
-package org.dbsyncer.manager.extractor.increment;
-
-import org.dbsyncer.manager.extractor.Increment;
-import org.springframework.stereotype.Component;
-
-/**
- * 定时同步
- *
- * @author AE86
- * @version 1.0.0
- * @date 2020/05/05 15:28
- */
-@Component
-public class TimingIncrement implements Increment {
-
-    @Override
-    public void close(String metaId) {
-
-    }
-}

dbsyncer-monitor/pom.xml

@@ -26,12 +26,5 @@
             <version>${project.parent.version}</version>
         </dependency>
 
-        <!-- Schedule 定时服务 -->
-        <dependency>
-            <groupId>org.ghi</groupId>
-            <artifactId>dbsyncer-schedule</artifactId>
-            <version>${project.parent.version}</version>
-        </dependency>
-
     </dependencies>
 </project>

dbsyncer-parser/src/main/test/org.dbsyncer.parser/ConnectorParserTest.java → dbsyncer-parser/src/main/test/ConnectorParserTest.java

@@ -4,8 +4,6 @@ import org.apache.commons.io.FileUtils;
 import org.dbsyncer.common.util.JsonUtil;
 import org.dbsyncer.connector.config.ConnectorConfig;
 import org.dbsyncer.connector.enums.ConnectorEnum;
-import org.dbsyncer.listener.config.ListenerConfig;
-import org.dbsyncer.listener.enums.ListenerEnum;
 import org.dbsyncer.parser.model.Connector;
 import org.dbsyncer.parser.model.Mapping;
 import org.dbsyncer.parser.model.TableGroup;
@@ -49,15 +47,7 @@ public class ConnectorParserTest {
 
         // 解析基本信息
         JSONObject map = new JSONObject(json);
-        JSONObject listener = (JSONObject) map.remove("listener");
         Mapping mapping = JsonUtil.jsonToObj(map.toString(), Mapping.class);
-
-        // 解析监听器
-        String listenerType = listener.getString("listenerType");
-        Class<?> configClass = ListenerEnum.getConfigClass(listenerType);
-        Object obj = JsonUtil.jsonToObj(listener.toString(), configClass);
-        mapping.setListener((ListenerConfig) obj);
-
         System.out.println(mapping);
     }
 

dbsyncer-schedule/pom.xml

@@ -1,48 +0,0 @@
-<?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0"
-         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <parent>
-        <artifactId>dbsyncer</artifactId>
-        <groupId>org.ghi</groupId>
-        <version>1.0.0-Alpha</version>
-    </parent>
-    <modelVersion>4.0.0</modelVersion>
-
-    <artifactId>dbsyncer-schedule</artifactId>
-
-    <dependencies>
-        <!-- Common 公共层 -->
-        <dependency>
-            <groupId>org.ghi</groupId>
-            <artifactId>dbsyncer-common</artifactId>
-            <version>${project.parent.version}</version>
-        </dependency>
-
-        <!-- 定时器 -->
-        <dependency>
-            <groupId>org.quartz-scheduler</groupId>
-            <artifactId>quartz</artifactId>
-            <exclusions>
-                <exclusion>
-                    <groupId>com.zaxxer</groupId>
-                    <artifactId>HikariCP-java6</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.mchange</groupId>
-                    <artifactId>mchange-commons-java</artifactId>
-                </exclusion>
-                <exclusion>
-                    <groupId>com.mchange</groupId>
-                    <artifactId>c3p0</artifactId>
-                </exclusion>
-            </exclusions>
-        </dependency>
-        <dependency>
-            <groupId>org.quartz-scheduler</groupId>
-            <artifactId>quartz-jobs</artifactId>
-        </dependency>
-
-    </dependencies>
-
-</project>

dbsyncer-schedule/src/main/java/org/dbsyncer/schedule/SchedulerFactory.java

@@ -1,5 +0,0 @@
-package org.dbsyncer.schedule;
-
-public final class SchedulerFactory {
-
-}

dbsyncer-schedule/src/main/java/org/dbsyncer/schedule/quartz/TaskQuartzJob.java

@@ -1,34 +0,0 @@
-package org.dbsyncer.schedule.quartz;
-
-import org.quartz.Job;
-import org.quartz.JobDetail;
-import org.quartz.JobExecutionContext;
-import org.quartz.JobExecutionException;
-
-/**
- * 驱动定时任务执行工厂
- *
- * @author AE86
- * @ClassName: TaskQuartzJob
- * @Description: 触发驱动定时抽取数据, 暂时只支持Mysql、Oracle、Ldap
- * @date: 2017年11月8日 下午4:17:43
- */
-public class TaskQuartzJob implements Job {
-
-    @Override
-    public void execute(JobExecutionContext context) throws JobExecutionException {
-        JobDetail jobDetail = context.getJobDetail();
-        // 获取任务ID
-        String taskId = jobDetail.getKey().getName();
-        // 即将面临的问题如下：
-        // 1.如果上一次任务未处理完成,而当前又产生了新的任务,此时多个任务可能会在同一时间执行相同工作,会造成重复数据.
-        // 2.如果上一次任务未处理完成,而当前又产生了新的任务,长时间下去,可能会造成任务堆积,影响系统性能.
-
-        // 相应的解决方法：
-        // 1.创建集合,存放执行的任务
-        // 2.从集合里检查上一次任务存在,继续执行.
-        // 3.从集合里检查上一次任务不存在,拒绝当前任务.
-        //TaskQuartzHandle.getInstance().handle(taskId);
-    }
-
-}

pom.xml

@@ -11,7 +11,7 @@
     <name>dbsyncer</name>
     <url>https://gitee.com/ghi/dbsyncer</url>
 
-    <!-- 管理的子模块(14个) -->
+    <!-- 管理的子模块(13个) -->
     <modules>
         <module>dbsyncer-biz</module>
         <module>dbsyncer-cache</module>
@@ -24,7 +24,6 @@
         <module>dbsyncer-monitor</module>
         <module>dbsyncer-parser</module>
         <module>dbsyncer-plugin</module>
-        <module>dbsyncer-schedule</module>
         <module>dbsyncer-storage</module>
         <module>dbsyncer-web</module>
     </modules>