提供Hibernate模板，spark local任务，sftp和ftp相关

async-task/AsyncTaskQueue.java

@@ -0,0 +1,247 @@
+package com.yiidata.intergration.web.task;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+/**
+ *
+ * 异步任务队列
+ *
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2020/11/16
+ * Time: 15:36
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+@Slf4j
+public class AsyncTaskQueue {
+
+
+    /**
+     * 任务缓存
+     */
+    private final static Map<String, Object> TASK_QUEUE = new ConcurrentHashMap<>();
+
+
+    /**
+     * 任务错误队列
+     */
+    private final static Map<String, Object> ERROR_TASK_QUEUE = new ConcurrentHashMap<>();
+
+
+    /**
+     * 正在运行的队列
+     */
+    private final static Set<String> RUNNING_TASK_QUEUE = new HashSet<>();
+
+
+    /**
+     * 控制 Spark 并发的信号量
+     */
+    private final Semaphore semaphore;
+
+
+    /**
+     * 公平锁
+     */
+    private final static Lock LOCK = new ReentrantLock();
+
+
+
+    private static AsyncTaskQueue SPARK_QUEUE;
+
+    private AsyncTaskQueue(int permits) {
+        semaphore = new Semaphore(permits);
+    }
+
+
+    /**
+     * 初次调用有效，
+     * @return
+     */
+    public static AsyncTaskQueue getInstance() {
+        return getInstance(3);
+    }
+
+
+    /**
+     * 按照配置，设置并发量。 第一次调用有效
+     * @param permits
+     * @return
+     */
+    public static synchronized AsyncTaskQueue getInstance(int permits) {
+        if(SPARK_QUEUE == null) {
+            SPARK_QUEUE = new AsyncTaskQueue(permits);
+        }
+        return SPARK_QUEUE;
+    }
+
+
+
+
+    /**
+     * 添加任务
+     * @param taskId
+     * @param taskInfo
+     */
+    public static boolean addTask(String taskId, Map taskInfo) {
+        LOCK.lock();
+        try {
+            if(!TASK_QUEUE.containsKey(taskId)) {
+                TASK_QUEUE.put(taskId, taskInfo);
+                log.info("add task: {} , params: {}", taskId, String.valueOf(taskInfo));
+                return true;
+            }
+        } finally {
+            LOCK.unlock();
+        }
+        return false;
+    }
+
+
+    /**
+     * 获取当前需要执行队列的长度
+     * @return
+     */
+    public static int getPendingTaskSize() {
+        LOCK.lock();
+        try {
+            HashMap<String, Object> tmpMap = new HashMap<>(TASK_QUEUE);
+            for (String s : RUNNING_TASK_QUEUE) {
+                tmpMap.remove(s);
+            }
+            return tmpMap.size();
+        } finally {
+            LOCK.unlock();
+        }
+    }
+
+
+    /**
+     * 获取当前需要执行队列
+     * @return
+     */
+    public static Set<String> getPendingTasks() {
+        LOCK.lock();
+        try {
+            HashMap<String, Object> tmpMap = new HashMap<>(TASK_QUEUE);
+            for (String s : RUNNING_TASK_QUEUE) {
+                tmpMap.remove(s);
+            }
+            return tmpMap.keySet();
+        } finally {
+            LOCK.unlock();
+        }
+    }
+
+
+    /**
+     * 获取当前正在执行任务的长度
+     * @return
+     */
+    public static int getRunningTaskSize() {
+        return RUNNING_TASK_QUEUE.size();
+    }
+
+
+
+    public static Object getTaskInfo(String taskId) {
+        return TASK_QUEUE.get(taskId);
+    }
+
+
+    /**
+     * 移除任务
+     * @param taskId
+     */
+    public static void removeTask(String taskId) {
+        LOCK.lock();
+        try {
+            TASK_QUEUE.remove(taskId);
+            RUNNING_TASK_QUEUE.remove(taskId);
+            log.info("remove task: {}", taskId);
+        } finally {
+            LOCK.unlock();
+        }
+    }
+
+
+    /**
+     * 错误的任务报告
+     * @param taskId
+     */
+    public static void reportError(String taskId) {
+        LOCK.lock();
+        try {
+            Object errorTaskInfo = TASK_QUEUE.remove(taskId);
+            ERROR_TASK_QUEUE.put(taskId, errorTaskInfo);
+            RUNNING_TASK_QUEUE.remove(taskId);
+        } finally {
+            LOCK.unlock();
+        }
+    }
+
+
+    /**
+     * 判断任务是否正在运行
+     * @param taskId
+     * @return
+     */
+    public static boolean runningTask(String taskId) {
+        return RUNNING_TASK_QUEUE.contains(taskId);
+    }
+
+
+    /**
+     * 执行该函数
+     * @param executor
+     * @param task
+     */
+    public void execute(Executor executor, final SuperTask task) {
+        executor.execute(()->{
+            final String runningTaskId = task.getTaskId();
+            // 有任务需要运行
+            if(AsyncTaskQueue.runningTask(runningTaskId)) {
+                // 取得的待运行的task，不能是正在运行的列表中的
+                log.info("task {} running.", runningTaskId);
+                return;
+            }
+            // 获得一个许可
+            try {
+                semaphore.acquire();
+            } catch (InterruptedException e) {
+                return;
+            }
+            try {
+                // 运行任务
+                RUNNING_TASK_QUEUE.add(runningTaskId);
+                log.info("running task: {}", runningTaskId);
+                task.run();
+                log.info("finished task: {}", runningTaskId);
+                // 执行成功，移除
+                removeTask(runningTaskId);
+            } catch (Exception e) {
+                log.info("执行任务异常。error task: " + runningTaskId, e);
+                // 运行错误
+                reportError(runningTaskId);
+            } finally {
+                // 释放许可
+                semaphore.release();
+            }
+        });
+    }
+}

async-task/DefaultTaskFactory.java

@@ -0,0 +1,80 @@
+package com.yiidata.intergration.web.task;
+
+import com.yiidata.intergration.web.config.IntenetProperties;
+import com.yiidata.intergration.web.config.TaskProperties;
+import com.yiidata.intergration.web.modules.dingtalk.service.DingTalkMessageService;
+import com.yiidata.intergration.web.modules.dingtalk.task.CzscAnalysisTask;
+import com.yiidata.intergration.web.modules.dingtalk.task.XGAnalysisTask;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2020/11/17
+ * Time: 12:54
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+@Component
+@Slf4j
+public class DefaultTaskFactory {
+
+
+
+    @Autowired
+    TaskProperties taskProperties;
+
+
+    @Autowired
+    IntenetProperties intenetProperties;
+
+
+    @Autowired
+    DingTalkMessageService dingTalkMessageService;
+
+
+    @Autowired
+    JdbcTemplate jdbcTemplate;
+
+    /**
+     * 根据 TaskInfo 生成 Task，不同 Task 类型
+     * @param taskId TaskID
+     * @param taskInfo
+     * @return
+     */
+    public SuperTask newTask(String taskId, Map<String, Object> taskInfo) {
+        String type = Optional.ofNullable((String)taskInfo.get("taskType")).orElse("TEST");
+        switch (type) {
+            case "CZSC_ANALYSIS":
+                //缠中说禅 分析任务
+                CzscAnalysisTask task = new CzscAnalysisTask(
+                        taskId, taskInfo);
+                task.setDingTalkMessageService(dingTalkMessageService);
+                task.setIntenetProperties(intenetProperties);
+                task.setJdbcTemplate(jdbcTemplate);
+                return task;
+
+            case "XG_ANALYSIS":
+                // 选股任务
+                XGAnalysisTask xgAnalysisTask = new XGAnalysisTask(taskId, taskInfo);
+                xgAnalysisTask.setDingTalkMessageService(dingTalkMessageService);
+                xgAnalysisTask.setIntenetProperties(intenetProperties);
+                xgAnalysisTask.setJdbcTemplate(jdbcTemplate);
+                return xgAnalysisTask;
+            default:
+                // 测试任务，未知任务
+                return new TestTask(taskId);
+        }
+    }
+}

async-task/SuperTask.java

@@ -0,0 +1,56 @@
+package com.yiidata.intergration.web.task;
+
+/**
+ *
+ * Spark 任务并发控制，控制总超类
+ *
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2020/11/17
+ * Time: 11:11
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+public abstract class SuperTask implements Runnable {
+
+
+    final String taskId;
+
+
+    public SuperTask(String taskId) {
+        this.taskId = taskId;
+    }
+
+
+    /**
+     * 获取前端 TASKID
+     *
+     * @return
+     */
+    public final String getTaskId() {
+        return taskId;
+    }
+
+    /**
+     * 执行任务
+     */
+    @Override
+    public final void run() {
+        try {
+            doExecute();
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+    }
+
+
+    /**
+     * 执行任务
+     * @throws Exception
+     */
+    public abstract void doExecute() throws Exception;
+}

async-task/TaskActivation.java

@@ -0,0 +1,26 @@
+package com.yiidata.intergration.web.task;
+
+/**
+ *
+ *
+ * 任务消息激活
+ *
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2020/11/17
+ * Time: 13:54
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+public interface TaskActivation {
+
+
+    /**
+     * 触发 Task 执行
+     */
+    public void active();
+}

async-task/TaskActivationTask.java

@@ -0,0 +1,134 @@
+package com.yiidata.intergration.web.task;
+
+import com.yiidata.intergration.web.config.TaskProperties;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+/**
+ *
+ * 控制 Spark Task 队列
+ *
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2020/11/17
+ * Time: 18:04
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+@Component
+@Slf4j
+public class TaskActivationTask implements Callable<Map<String, Object>>, TaskActivation,
+        InitializingBean, DisposableBean {
+
+
+
+    @Autowired
+    DefaultTaskFactory defaultTaskFactory;
+
+
+    @Autowired
+    ThreadPoolTaskExecutor taskExecutor;
+
+
+    @Autowired
+    TaskProperties taskProperties;
+
+
+    /**
+     * 线程完成的句柄
+     */
+    private Future<Map<String, Object>> taskFuture;
+
+
+    /**
+     * 当前是否正在运行
+     */
+    private boolean running = true;
+
+
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        AsyncTaskQueue.getInstance(taskProperties.getParallel());
+        taskFuture = taskExecutor.submit(this);
+        log.info("start scan TaskActivationTask thread.");
+    }
+
+
+    @Override
+    public void active() {
+        try {
+            synchronized (this) {
+                this.notify();
+            }
+        } catch (Exception e) {
+            log.error("notify TaskActivationTask thread error.", e);
+        }
+    }
+
+    @Override
+    public Map<String, Object> call() throws Exception {
+        while (running) {
+            // 开始提交任务
+            try {
+                // 如果待运行的队列非空，且当前运行的并发小于要求的值，则开启任务
+                if(AsyncTaskQueue.getPendingTaskSize() > 0 && AsyncTaskQueue.getRunningTaskSize() < taskProperties.getParallel()) {
+                    Set<String> pendingTasks = AsyncTaskQueue.getPendingTasks();
+                    int count = taskProperties.getParallel() - AsyncTaskQueue.getRunningTaskSize(); // 当前可提交的任务数量
+
+                    // 开始提交 count 个任务
+                    int i = 0;
+                    for (String taskId : pendingTasks) {
+                        i++;
+                        AsyncTaskQueue.getInstance().execute(taskExecutor,
+                                defaultTaskFactory.newTask(taskId, (Map) AsyncTaskQueue.getTaskInfo(taskId)));
+                        if(i >= count) {
+                            break;
+                        }
+                    }
+                }
+            } catch (Exception e) {
+                log.error("Submit InnerTaskExecution Error!", e);
+            }
+
+            try {
+                synchronized (this) {
+                    // 停止 10s，采用 wait 比 sleep 好, 让出 CPU 时间，使执行其他线程
+                    this.wait(TimeUnit.SECONDS.toMillis(30));
+                }
+            } catch (InterruptedException e) {
+                log.error("Interrupted TaskActivationTask Thread!");
+                // 系统中断
+                running = false;
+            }
+        }
+
+        return new HashMap<>();
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        log.warn("destroy spring context, will stop TaskActivationTask thread!");
+        running = false;
+        active();
+        if(taskFuture != null) {
+            taskFuture.cancel(true);
+        }
+    }
+
+}

java-sftp-ftp/FtpClientHandler.java

@@ -0,0 +1,269 @@
+package ftp;
+
+import org.apache.commons.net.ftp.FTPClient;
+import org.apache.commons.net.ftp.FTPFile;
+import org.apache.commons.net.ftp.FTPReply;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.stream.Collectors;
+
+
+/**
+ * Created by hadoop on 2018/8/9.
+ */
+public class FtpClientHandler implements P11ClientHandler {
+
+    protected static Logger logger = LoggerFactory.getLogger("FtpClientHandler");
+
+
+    private FTPClient ftpClient;
+    private int reply;
+    private final String ftpHost;
+    private final int ftpPort;
+    private final String ftpUsername;
+    private final String ftpPassword;
+
+    public FtpClientHandler(String host,
+                            int port,
+                            String username,
+                            String password,
+                            Properties params) {
+        ftpHost = host;
+        ftpPort = port;
+        ftpUsername = username;
+        ftpPassword = password;
+        this.reconnect(params);
+    }
+
+
+    /**
+     * 以账号和密码重新连接
+     * @param params
+     */
+    public void reconnect(Properties params) {
+        try {
+            ftpClient = new FTPClient();
+            ftpClient.connect(ftpHost, ftpPort); //连接Ftp服务器
+            ftpClient.login(ftpUsername, ftpPassword);
+            ftpClient.setControlEncoding("UTF-8");
+            ftpClient.setFileType(FTPClient.BINARY_FILE_TYPE);
+            ftpClient.enterLocalPassiveMode(); // 被动模式
+            reply = ftpClient.getReplyCode();
+            if (!FTPReply.isPositiveCompletion(reply)) {
+                logger.info("Donot connect FTP Server, username or password is error.");
+                ftpClient.disconnect();
+            } else {
+                logger.info("The FTP connect success.");
+            }
+        } catch (IOException e) {
+            logger.error("ftp connect is error.", e);
+            throw new IllegalStateException("can not connect sftp: " + ftpHost+":"+ ftpPort);
+        }
+    }
+
+
+    /**
+     * 读取文件流
+     * @param file
+     * @return
+     * @throws IOException
+     */
+    public InputStream readFile(String file) throws IOException {
+        return ftpClient.retrieveFileStream(file);
+    }
+
+
+
+    /**
+     * 读取文件流
+     * @param file
+     * @return
+     * @throws IOException
+     */
+    @Override
+    public OutputStream writeFile(String file, boolean overwrite) throws IOException {
+        if(overwrite) {
+            return ftpClient.storeFileStream(file);
+        } else {
+            return ftpClient.appendFileStream(file);
+        }
+    }
+
+    public boolean renameFile(String file, String newFileName) {
+        try {
+            return ftpClient.rename(file, newFileName);
+        } catch (IOException e) {
+            logger.error("Error in rename ftp file.", e);
+            return false;
+        }
+    }
+
+    /**
+     * 删除文件, 返回 true 则删除成功
+     * @param fileName
+     */
+    public boolean deleteFile(String fileName) {
+        try {
+            return ftpClient.deleteFile(fileName);
+        } catch (IOException e) {
+            logger.info("Delete file fail.", e);
+        }
+        return false;
+    }
+
+
+
+    /**
+     * 判断目录是否存在
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean existsDir(String path) {
+        try {
+            FTPFile mdtmFile = ftpClient.mlistFile(path);
+            return mdtmFile != null && mdtmFile.isDirectory();
+        } catch (IOException e) {
+            logger.error(e.getMessage());
+        }
+        return false;
+    }
+
+
+
+    /**
+     * 判断目录是否存在
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean existsFile(String path) {
+        try {
+            FTPFile mdtmFile = ftpClient.mlistFile(path);
+            return mdtmFile != null && mdtmFile.isFile();
+        } catch (IOException e) {
+        	logger.error(e.getMessage());
+        }
+        return false;
+    }
+
+
+    /**
+     * 判断目录或者文件是否存在
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean exists(String path) {
+        try {
+            FTPFile mdtmFile = ftpClient.mlistFile(path);
+            return mdtmFile != null;
+        } catch (IOException e) {
+        	logger.error(e.getMessage());
+        }
+        return false;
+    }
+
+
+    /**
+     * 创建多层目录
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean mkdirs(String path) {
+        Path path1 = Paths.get(path);
+        Path parentPath = path1.getParent();
+        if(existsDir(parentPath.toString())){
+            return mkdir(path1.toString()); // 创建当前文件夹
+        } else {
+            if(mkdirs(parentPath.toString())) { // 创建父级目录
+                return mkdir(path1.toString()); // 创建当前文件夹
+            }
+        }
+        // 最后创建文件夹肯定是不成功的。
+        return existsDir(path);
+    }
+
+
+    /**
+     * 创建目录, 返回 true 则创建目录成功
+     * @param dirName 一级或者多级目录
+     */
+    @Override
+    public boolean mkdir(String dirName) {
+        try {
+            return ftpClient.makeDirectory(dirName);
+        } catch (IOException e) {
+            logger.info("mkdir " + dirName + " fail.", e);
+        }
+        return false;
+    }
+
+    /**
+     * 获取一个目录下或者文件的子文件
+     * @param ftpPath
+     * @return
+     */
+    @Override
+    public List<Path> getChildren(String ftpPath) {
+        try {
+            FTPFile[] ftpFiles = ftpClient.listFiles(ftpPath);
+            List<Path> files = Arrays.stream(ftpFiles).map(it->Paths.get(ftpPath, it.getName())).collect(Collectors.toList());
+            return files;
+        } catch (IOException e) {
+            logger.info("Error in scan ftpfile.");
+        }
+        return Collections.emptyList();
+    }
+
+
+
+    @Override
+    public void close() throws IOException {
+        ftpClient.disconnect();
+    }
+
+    //test
+    public static void main(String[] args) throws IOException {
+        Properties map = new Properties();
+        String host = "localhost";
+        int port = 2121;
+        String username = "admin";
+        String password = "admin";
+        FtpClientHandler handler = new FtpClientHandler(host, port, username, password, map);
+//        handler.changeWorkingDirectory("/sms");
+        List<Path> result = handler.getChildren("/ta-lib");
+        for (Path ftpFile : result) {
+            System.out.println(ftpFile.toString());
+        }
+
+        /*
+        OutputStream out = handler.writeFile("/hello.txt", false);
+        IOUtils.copy(new FileInputStream("pom.xml"), out);
+        out.flush();
+        out.close();
+        */
+
+        System.out.println(handler.exists("/hello.txt"));
+        System.out.println(handler.exists("/example_data_1225"));
+        System.out.println(handler.exists("/world/example_data_1225"));
+
+        if(!handler.existsDir("/world/example_data_1225")) {
+            System.out.println(handler.mkdirs("/world/example_data_1225"));
+        }
+        System.out.println(handler.exists("/world/example_data_1225"));
+
+    }
+
+}

java-sftp-ftp/P11ClientHandler.java

@@ -0,0 +1,85 @@
+package ftp;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.Properties;
+
+/**
+ *
+ *
+ * 太保家园，P11 连接。 可选的子实现：FTP， SFTP 协议。使用完成请关闭 close
+ *
+ *
+ * Created by 59850 on 2019/3/28.
+ */
+public interface P11ClientHandler extends Closeable {
+
+
+    /**
+     * 根据 配置参数，重新连接 ftp
+     * @param params
+     * @throws IOException
+     */
+    public void reconnect(Properties params) throws IOException;
+
+
+    /**
+     * 写入文件，输出流
+     * @param file 文件地址，绝对路径
+     * @param overwrite 是否覆盖写入
+     * @return
+     * @throws IOException
+     */
+    public OutputStream writeFile(String file, boolean overwrite) throws IOException;
+
+
+    /**
+     * 创建目录，， 如 Linux: mkdir /dir/path, 如 /dir 不存在则报错
+     * @param path
+     * @return
+     */
+    public boolean mkdir(String path);
+
+
+    /**
+     * 创建目录，多层级创建， 如 Linux: mkdir -p /dir/path
+     * @param path
+     * @return
+     */
+    public boolean mkdirs(String path);
+
+
+
+    /**
+     * 获取子目录或子文件
+     * @param ftpPath
+     * @return
+     */
+    public List<Path> getChildren(String ftpPath);
+
+    /**
+     * 判断是否存在该路径，无论文件夹或者文件
+     * @param path
+     * @return 返回 true 则存在
+     */
+    public boolean exists(String path);
+
+
+    /**
+     * 判断是否存在该文件夹
+     * @param path 文件夹地址
+     * @return  返回 true 则存在
+     */
+    public boolean existsDir(String path);
+
+
+    /**
+     * 判断是否存在该文件
+     * @param path 文件地址
+     * @return 返回 true 则存在
+     */
+    public boolean existsFile(String path);
+}

java-sftp-ftp/SftpClientHandler.java

@@ -0,0 +1,262 @@
+package ftp;
+
+import com.google.common.collect.Lists;
+import org.apache.sshd.client.subsystem.sftp.fs.SftpFileSystemProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.nio.file.*;
+import java.util.*;
+
+import static java.nio.file.StandardOpenOption.APPEND;
+import static java.nio.file.StandardOpenOption.CREATE;
+
+
+/**
+ * Created by hadoop on 2018/8/9.
+ */
+public class SftpClientHandler implements P11ClientHandler {
+
+    protected static Logger logger = LoggerFactory.getLogger("SftpClientHandler");
+
+
+    private final String sftpHost;
+    private final int sftpPort;
+    private final String ftpUsername;
+    private final String ftpPassword;
+
+
+    /**
+     * sftp 文件系统
+     */
+    FileSystem fs;
+
+
+
+    public SftpClientHandler(String host,
+                             int port,
+                             String username,
+                             String password,
+                             Properties params) {
+        sftpHost = host;
+        sftpPort = port;
+        ftpUsername = username;
+        ftpPassword = password;
+        this.reconnect(params);
+    }
+
+
+    /**
+     * 以账号和密码重新连接
+     * @param params
+     */
+    @Override
+    public void reconnect(Properties params) {
+        java.net.URI uri = SftpFileSystemProvider.createFileSystemURI(this.sftpHost, sftpPort, ftpUsername, ftpPassword);
+        try {
+            fs = FileSystems.newFileSystem(uri, (Map) params);
+        } catch (FileSystemAlreadyExistsException e){
+            logger.warn("exists file system sftp...");
+            fs = FileSystems.getFileSystem(uri);
+        } catch (IOException e) {
+            logger.error("ftp connect is error.", e);
+            throw new IllegalStateException("can not connect sftp: " + sftpHost+":"+ sftpPort);
+        }
+    }
+
+
+
+    /**
+     * 读取文件流
+     * @param file
+     * @return
+     * @throws IOException
+     */
+    @Override
+    public OutputStream writeFile(String file, boolean overwrite) throws IOException {
+        Path path1 = fs.getPath(file);
+        if(overwrite) {
+            return Files.newOutputStream(path1, CREATE);
+        } else {
+            return Files.newOutputStream(path1, APPEND);
+        }
+    }
+
+
+
+    public boolean renameFile(String file, String newFileName) {
+        try {
+            Path path1 = fs.getPath(file);
+            Path path = Paths.get(path1.getParent().toString(), newFileName);
+            Files.move(path, path1);
+            return true;
+        } catch (Exception e) {
+            logger.error("Error in rename ftp file.", e);
+            return false;
+        }
+    }
+
+
+    /**
+     * 删除文件, 返回 true 则删除成功
+     * @param path
+     */
+    public boolean deleteFile(String path) {
+        try {
+            Path path1 = fs.getPath(path);
+            Files.delete(path1);
+        } catch (Exception e) {
+            logger.info("Delete file fail.", e);
+        }
+        return false;
+    }
+
+
+
+    /**
+     * 判断目录是否存在
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean existsDir(String path) {
+        try {
+            Path path1 = fs.getPath(path);
+            return Files.exists(path1) && Files.isDirectory(path1);
+        } catch (Exception e) {
+        }
+        return false;
+    }
+
+
+
+    /**
+     * 判断目录是否存在
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean existsFile(String path) {
+        try {
+            Path path1 = fs.getPath(path);
+            return Files.exists(path1) && Files.isRegularFile(path1);
+        } catch (Exception e) {
+        }
+        return false;
+    }
+
+
+    /**
+     * 判断目录或者文件是否存在
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean exists(String path) {
+        try {
+            Path path1 = fs.getPath(path);
+            return Files.exists(path1);
+        } catch (Exception e) {
+        }
+        return false;
+    }
+
+
+    /**
+     * 创建多层目录
+     * @param path
+     * @return
+     */
+    @Override
+    public boolean mkdirs(String path) {
+        Path path1 = fs.getPath(path);
+        Path parentPath = path1.getParent();
+        if(parentPath == null) {
+            // parentPath 等于 null 则说明是文件系统的根目录了。
+            return true;
+        }
+        // 父级目录已经存在，只创建当前目录
+        if(existsDir(parentPath.toString())){
+            return mkdir(path1.toString()); // 创建当前文件夹
+        } else {
+            // 父级目录不存在，则先创建父级目录，再创建当前目录
+            if(mkdirs(parentPath.toString())) { // 创建父级目录
+                return mkdir(path1.toString()); // 创建当前文件夹
+            }
+        }
+        // 最后创建文件夹肯定是不成功的。
+        return existsDir(path);
+    }
+
+
+    /**
+     * 创建目录, 返回 true 则创建目录成功
+     * @param dirName 一级或者多级目录
+     */
+    @Override
+    public boolean mkdir(String dirName) {
+        try {
+            Path path1 = fs.getPath(dirName);
+            Files.createDirectory(path1);
+            return true;
+        } catch (IOException e) {
+            logger.info("mkdir " + dirName + " fail.", e);
+        }
+        return false;
+    }
+
+
+
+    /**
+     * 获取一个目录下或者文件的子文件
+     * @param ftpPath
+     * @return
+     */
+    @Override
+    public List<Path> getChildren(String ftpPath) {
+        Path remotePath = fs.getPath(ftpPath);
+        try {
+            DirectoryStream<Path> paths = Files.newDirectoryStream(remotePath);
+            return Lists.newArrayList(paths.iterator());
+        } catch (IOException e) {
+            logger.info("Error in scan sftpfile.");
+        }
+        return Collections.emptyList();
+    }
+
+
+
+    @Override
+    public void close() throws IOException {
+        logger.warn("close the file system");
+        fs.close();
+    }
+
+
+    public static void main(String[] args) {
+        Properties map = new Properties();
+        String host = "localhost";
+        int port = 2121;
+        String username = "admin";
+        String password = "admin";
+        SftpClientHandler handler = new SftpClientHandler(host, port, username, password, map);
+//        handler.changeWorkingDirectory("/sms");
+        List<Path> result = handler.getChildren("/");
+
+        /*
+        OutputStream out = handler.writeFile("/hello.txt", false);
+        IOUtils.copy(new FileInputStream("pom.xml"), out);
+        out.flush();
+        out.close();
+        */
+
+        handler.exists("/hello.txt");
+        handler.exists("/example_data_1225");
+        handler.exists("/world/example_data_1225");
+
+        handler.mkdirs("/world/example_data_1225");
+        handler.exists("/world/example_data_1225");
+    }
+}

java-sftp-ftp/package-info.java

@@ -0,0 +1,18 @@
+/**
+ *
+ * P11
+ *
+ * <pre>
+ *
+ * Created by zhenqin.
+ * User: zhenqin
+ * Date: 2019/3/15
+ * Time: 18:17
+ * Vendor: yiidata.com
+ * To change this template use File | Settings | File Templates.
+ *
+ * </pre>
+ *
+ * @author zhenqin
+ */
+package ftp;

local-spark/JobExecuteResultHandler.java

@@ -0,0 +1,156 @@
+package com.primeton.datainsight.task;
+
+import com.primeton.datainsight.bo.WarnMessage;
+import com.primeton.datainsight.service.WarnService;
+import com.primeton.datainsight.utils.Constance;
+import com.primeton.datainsight.utils.TimeUtils;
+import org.apache.commons.exec.DefaultExecuteResultHandler;
+import org.apache.commons.exec.ExecuteException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Date;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ *
+ * 任务超时告警监控
+ *
+ *
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2019/4/11
+ * Time: 11:04
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+public class JobExecuteResultHandler extends DefaultExecuteResultHandler {
+
+    private static Logger LOG = LoggerFactory.getLogger(JobExecuteResultHandler.class);
+
+
+    /**
+     * 开始时间
+     */
+    public final long startTime = System.currentTimeMillis();
+
+
+    /**
+     * 预警服务
+     */
+    final WarnService warnService;
+
+
+    /**
+     * 是否开启预警
+     */
+    final boolean enable;
+
+    /**
+     * 任务一小时预警
+     */
+    final long warnTimeout;
+
+
+    /**
+     * 1 小时候后每10分钟预警一次
+     */
+    final long warnTimeoutInterval;
+
+
+    /**
+     * 预警次数
+     */
+    final AtomicInteger warnCount = new AtomicInteger(0);
+
+
+    /**
+     * 任务名称，作为任务日志显示用
+     */
+    String taskName;
+
+    public JobExecuteResultHandler(WarnService warnService) {
+        this(warnService, warnService.isWarnEnable(), warnService.getWarnJobTimeout(), warnService.getWarnJobPeriod());
+    }
+
+
+    /**
+     * 任务执行多长时间后未完成预警，如果预警后每隔多长时间继续预警
+     * @param warnTimeout
+     * @param warnTimeoutInterval
+     */
+    public JobExecuteResultHandler(WarnService warnService,
+                                   boolean enable,
+                                   long warnTimeout,
+                                   long warnTimeoutInterval) {
+        this.warnService = warnService;
+        this.enable = enable;
+        this.warnTimeout = warnTimeout > 0 ? warnTimeout : -1L;
+        this.warnTimeoutInterval = warnTimeoutInterval > 0 ? warnTimeoutInterval : -1L;
+    }
+
+
+    @Override
+    public void onProcessComplete(int exitValue) {
+        // 退出码非 0 则认为任务不成功
+        if(exitValue != 0){
+            throw new IllegalStateException("进程非正常退出。exitValue: " + exitValue);
+        }
+        super.onProcessComplete(exitValue);
+    }
+
+    @Override
+    public void onProcessFailed(ExecuteException e) {
+        super.onProcessFailed(e);
+        if(warnService.isWarnEnable()) {
+            WarnMessage warnMessage = new WarnMessage();
+            warnMessage.setWarnCatalog(Constance.WARN_CATALOG_TASK_EXE_FAILED);
+            warnMessage.setWarnType("task");
+            warnMessage.setWarnLevel("HIGH");
+            warnMessage.setWarnStart(new Date());
+            warnMessage.setWarnSubtext("任务执行失败");
+            warnMessage.setWarnContent(taskName + "执行失败。REASON: " + e.getMessage());
+            warnService.warn(warnMessage);
+        }
+    }
+
+
+    @Override
+    public void waitFor() throws InterruptedException {
+        while (!hasResult()) {
+            super.waitFor(1000);
+            long l = System.currentTimeMillis();
+            // 任务如果是开启的，就用任务的，否则使用全局的
+            boolean enable = this.enable;
+            // 如果没有开始告警，则不会查询一次 getTaskRunningTimeout（），提高性能
+            long warnTimeout = enable && this.warnTimeout > 0 ? this.warnTimeout : 3600000;
+            if(enable && (l - startTime) >= warnTimeout) {
+                // 超时了，开始预警
+                long warnInterval = this.warnTimeoutInterval > 0 ? this.warnTimeoutInterval : 600000;
+                long timeout = warnCount.get() * warnInterval + warnTimeout;
+                if(l - startTime >= timeout) {
+                    warnCount.incrementAndGet();
+                    LOG.warn("任务执行超时：" + TimeUtils.getTimeString(timeout));
+                    WarnMessage warnMessage = new WarnMessage();
+                    warnMessage.setWarnCatalog(Constance.WARN_CATALOG_TASK_TIMEOUT);
+                    warnMessage.setWarnType("task");
+                    warnMessage.setWarnLevel("HIGH");
+                    warnMessage.setWarnStart(new Date());
+                    warnMessage.setWarnSubtext("任务执行超时");
+                    warnMessage.setWarnContent(taskName + " 执行超时, 超时：" + TimeUtils.getTimeString(timeout));
+                    warnService.warn(warnMessage);
+
+                }
+            }
+        }
+    }
+
+
+    public void setTaskName(String taskName) {
+        this.taskName = taskName;
+    }
+}

local-spark/JobQueue.java

@@ -0,0 +1,407 @@
+package com.primeton.datainsight.task;
+
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.ImmutableSet;
+import com.primeton.datainsight.support.JobListener;
+import org.apache.commons.exec.ExecuteWatchdog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * <p>
+ * commons of this class
+ * </p>
+ *
+ * Created by zhaopx on 2018/1/24 0024-14:56
+ * Vendor: primeton.com
+ */
+public class JobQueue {
+
+
+    /**
+     * jobId -> Job Detail
+     */
+    protected final static Map<String, JobEntity> taskId2Job = new ConcurrentHashMap();
+
+
+    /**
+     *
+     * Yarn 上运行的 Application ID 和状态的缓存。appid -> {jobId, state}
+     *
+     */
+    protected final static Map<String, AppEntity> appId2Task = new ConcurrentHashMap();
+
+
+
+    /**
+     *
+     * Yarn 上运行的 Application ID 和状态的缓存。appid -> {jobId, state}
+     *
+     */
+    protected final static Map<String, ExecuteWatchdog> TASKID_TO_TASKREF_MAP = new ConcurrentHashMap();
+
+
+
+
+    protected static Logger LOG = LoggerFactory.getLogger(JobQueue.class);
+
+
+    /**
+     * 添加一个Task的 Observer 监听者
+     * @param jobId Job ID
+     * @param taskId TaskID
+     * @param watchdog 监听者
+     */
+    public static void addTaskObserver(String jobId, String taskId, ExecuteWatchdog watchdog) {
+        TASKID_TO_TASKREF_MAP.put(taskId, watchdog);
+    }
+
+
+
+    /**
+     * 添加一个Task的 Observer 监听者
+     * @param jobId Job ID
+     * @param taskId TaskID
+     */
+    public static void removeTaskObserver(String jobId, String taskId) {
+        TASKID_TO_TASKREF_MAP.remove(taskId);
+    }
+
+
+
+    /**
+     * 添加一个Task的 Observer 监听者
+     * @param jobId Job ID
+     * @param taskId TaskID
+     */
+    public static ExecuteWatchdog getTaskObserver(String jobId, String taskId) {
+        return TASKID_TO_TASKREF_MAP.get(taskId);
+    }
+
+    /**
+     * 返回尚未运行的 Application
+     * @return
+     */
+    public static Set<String> getBeforeRunning(){
+        ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+        for (Map.Entry<String, AppEntity> entry : appId2Task.entrySet()) {
+            if(entry.getValue().getState() == null) {
+                continue;
+            }
+
+            // Yarn 任务在运行前的几个状态
+            if("NEW|NEW_SAVING|SUBMITTED|ACCEPTED".contains(entry.getValue().getState())) {
+                builder.add(entry.getKey());
+            }
+        }
+        return builder.build();
+    }
+
+
+
+    /**
+     * 返回正在运行的 Application
+     * @return
+     */
+    public static Set<String> getRunningJob(){
+        ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+        for (Map.Entry<String, AppEntity> entry : appId2Task.entrySet()) {
+            if("RUNNING".equals(entry.getValue().getState())) {
+                builder.add(entry.getKey());
+            }
+        }
+        return builder.build();
+    }
+
+
+
+
+    /**
+     * 返回正在运行的 Quartz 任务
+     * @return
+     */
+    public static Set<JobEntity> getRunningJob0(){
+        ImmutableSet.Builder<JobEntity> builder = ImmutableSet.builder();
+        for (Map.Entry<String, JobEntity> entry : taskId2Job.entrySet()) {
+            builder.add(entry.getValue());
+        }
+        return builder.build();
+    }
+
+
+
+    /**
+     * 返回运行结束的 Application
+     * @return
+     */
+    public static Set<String> getFinishedJob() {
+        ImmutableSet.Builder<String> builder = ImmutableSet.builder();
+        for (Map.Entry<String, AppEntity> entry : appId2Task.entrySet()) {
+            if(entry.getValue().getState() == null) {
+                continue;
+            }
+            if("FINISHED|FAILED|KILLED".contains(entry.getValue().getState())) {
+                builder.add(entry.getKey());
+            }
+        }
+        return builder.build();
+    }
+
+
+    /**
+     * Spark Job 启动后推送到队列
+     * @param jobId 任务ID
+     * @param taskId taskID
+     * @param map 任务参数
+     * @param jobListener 任务回调
+     */
+    public static void pushNewTask(String jobId, String taskId, String uuid, JSONObject map, JobListener jobListener) {
+        JobEntity jobEntity = new JobEntity(jobId, taskId, uuid, map);
+        jobEntity.setJobListener(jobListener);
+        taskId2Job.put(taskId, jobEntity);
+    }
+
+
+    /**
+     * Yarn 反向推送到该接口，日志解析端推送
+     * @param appId AppID
+     * @param jobId JobID
+     * @param taskId TASK
+     * @param state 状态
+     */
+    public static void runningTask(String appId, String jobId, String taskId, String state) {
+        AppEntity appEntity = appId2Task.get(appId);
+        if(appEntity == null){
+            appEntity = new AppEntity(appId);
+        }
+
+        appEntity.setJobId(jobId);
+        appEntity.setTaskId(taskId);
+        appEntity.setState(state);
+        appId2Task.put(appId, appEntity);
+
+        // 运行前， 回调
+        JobEntity entity = taskId2Job.get(taskId);
+        if(entity != null && entity.getJobListener() != null){
+            JSONObject json = new JSONObject();
+            json.put("source", "console");
+            json.put("appId", appId);
+            json.put("uuid", entity.getUuid());
+
+            entity.getJobListener().call(jobId, taskId, state, json);
+        }
+
+        // Job 运行结束，一定会发送一次或多次，不用特别处理，运行前可能队列阻塞，启动迟缓，跟超时不同，需要回调
+    }
+
+
+    /**
+     * 移除 Job，移除后不再做监控
+     * @param appId Yarn AppId
+     */
+    public static void removeMonitorJob(String appId) {
+        AppEntity appEntity = appId2Task.get(appId);
+        if(appEntity == null){
+            return;
+        }
+
+        JobEntity jobEntity = taskId2Job.get(appEntity.getTaskId());
+        if(jobEntity == null){
+            LOG.warn("{} 已完成，移除该任务。", appEntity.getTaskId());
+            appId2Task.remove(appId);
+            return;
+        }
+
+        // 移除Job
+        appId2Task.remove(appId);
+        taskId2Job.remove(appEntity.getTaskId());
+        LOG.warn("{}/{} 已完成，移除该任务。", appEntity.getJobId(), appId);
+    }
+
+
+    /**
+     * 删除任务
+     * @param jobId 任务JobID
+     * @param taskId 任务TaskID
+     */
+    public static void removeMonitorTask(String jobId, String taskId) {
+        taskId2Job.remove(taskId);
+        LOG.warn("{}/{} 已完成，移除该任务。", jobId, taskId);
+    }
+
+    /**
+     * 移除任务
+     * @param jobId
+     * @param taskId
+     */
+    public static void removeErrorJob(String jobId, String taskId){
+        taskId2Job.remove(taskId);
+        LOG.warn("{}/{} 运行失败，从动态列表中移除。", jobId, taskId);
+    }
+
+
+    /**
+     * 获取 App Job 信息
+     * @param appId
+     * @return
+     */
+    public static AppEntity getApp(String appId) {
+        return appId2Task.get(appId);
+    }
+
+    public static class AppEntity {
+
+        final String appId;
+
+        String jobId;
+
+        String taskId;
+
+        String state;
+
+
+        public AppEntity(String appId) {
+            this.appId = appId;
+        }
+
+
+        public String getAppId() {
+            return appId;
+        }
+
+        public String getJobId() {
+            return jobId;
+        }
+
+        public void setJobId(String jobId) {
+            this.jobId = jobId;
+        }
+
+        public String getTaskId() {
+            return taskId;
+        }
+
+        public void setTaskId(String taskId) {
+            this.taskId = taskId;
+        }
+
+        public String getState() {
+            return state;
+        }
+
+        public void setState(String state) {
+            this.state = state;
+        }
+
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            AppEntity appEntity = (AppEntity) o;
+
+            return appId != null ? appId.equals(appEntity.appId) : appEntity.appId == null;
+
+        }
+
+        @Override
+        public int hashCode() {
+            return appId != null ? appId.hashCode() : 0;
+        }
+    }
+
+
+
+    public static class JobEntity {
+
+        /**
+         * Job ID
+         */
+        String jobId;
+
+        /**
+         * Task ID
+         */
+        String taskId;
+
+
+        /**
+         * UUID
+         */
+        String uuid;
+
+
+        /**
+         * Job 配置
+         */
+        JSONObject map;
+
+
+        /**
+         * 任务消息结果回调
+         */
+        JobListener jobListener;
+
+        /**
+         * 构造方法
+         * @param jobId Job ID
+         * @param uuid Task ID
+         * @param map Job 配置
+         */
+        public JobEntity(String jobId, String taskId, String uuid, JSONObject map) {
+            this.jobId = jobId;
+            this.taskId = taskId;
+            this.uuid = uuid;
+            this.map = map;
+        }
+
+
+        public String getJobId() {
+            return jobId;
+        }
+
+
+        public String getTaskId() {
+            return taskId;
+        }
+
+        public String getUuid() {
+            return uuid;
+        }
+
+        public JSONObject getMap() {
+            return map;
+        }
+
+
+        public JobListener getJobListener() {
+            return jobListener;
+        }
+
+        public void setJobListener(JobListener jobListener) {
+            this.jobListener = jobListener;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            JobEntity jobEntity = (JobEntity) o;
+
+            return jobId != null ? jobId.equals(jobEntity.jobId) : jobEntity.jobId == null;
+
+        }
+
+        @Override
+        public int hashCode() {
+            return jobId != null ? jobId.hashCode() : 0;
+        }
+    }
+
+}

local-spark/LogOutputStream.java

@@ -0,0 +1,147 @@
+package com.primeton.datainsight.task;
+
+import com.google.common.base.Charsets;
+import org.apache.commons.lang.SystemUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedWriter;
+import java.io.Closeable;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.OutputStreamWriter;
+import java.nio.charset.Charset;
+
+/**
+ * <p>
+ * </p>
+ * Created by zhaopx on 2018/1/12 0012-18:39
+ */
+public class LogOutputStream extends OutputStream implements Closeable {
+	private static Logger logger = LoggerFactory.getLogger(LogOutputStream.class);
+    /**
+     * Job Id
+     */
+    final String jobId;
+
+    /**
+     * Task Id
+     */
+    final String taskId;
+
+
+    /**
+     * console 输出日志的编码， Windows 使用GBK
+     */
+    final java.nio.charset.Charset charset;
+
+
+    /**
+     * 日志输出流
+     */
+    BufferedWriter logWriter;
+
+
+    /**
+     * 缓冲大小
+     */
+    volatile long bufferSiuze = 0;
+
+
+    /**
+     * 获取 Task 的目录
+     */
+    final String taskDir;
+
+
+    public LogOutputStream(String baseDir, String jobId, String taskId) {
+        this.jobId = jobId;
+        this.taskId = taskId;
+        if(SystemUtils.IS_OS_WINDOWS) {
+            charset = Charset.forName("GBK");
+        } else {
+            charset = Charsets.UTF_8;
+        }
+
+        // TODO Task dir
+        File tasks = new File(baseDir);
+        if(!tasks.exists() || tasks.isFile()){
+            tasks.mkdirs();
+        }
+
+        File jobDir = new File(tasks, jobId);
+        if(!jobDir.exists() || jobDir.isFile()){
+            jobDir.mkdirs();
+        }
+
+        // 临时Job
+        File taskDir = new File(jobDir, taskId);
+        // 肯定不存在的
+        if(!taskDir.exists()){
+            taskDir.mkdirs();
+        }
+        this.taskDir = taskDir.getAbsolutePath();
+        File logfile = new File(taskDir, "stdout");
+        try {
+            logWriter = new BufferedWriter(new OutputStreamWriter(new FileOutputStream(logfile), "UTF-8"));
+        } catch (Exception e) {
+            logger.error("无法打开文件。", e);
+        }
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+        if(b > 0) {
+            logger.info("unknown byte {}", b);
+        }
+    }
+
+
+    /**
+     * write one line
+     * @param line one line
+     * @return
+     * @throws IOException
+     */
+    public void write(String line) throws IOException {
+        if(logWriter != null) {
+            try {
+                logWriter.write(line);
+                logWriter.flush();
+            } catch (IOException e) {
+            	logger.error(e.getMessage());
+            }
+        }
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+        String line = new String(b, off, len, charset);
+        write(line);
+    }
+
+
+    @Override
+    public void flush() throws IOException {
+        if(logWriter != null) {
+            logWriter.flush();
+        }
+    }
+
+    @Override
+    public void close() throws IOException {
+        if(logWriter != null) {
+            logWriter.flush();
+            logWriter.close();
+        }
+
+        logWriter = null;
+    }
+
+
+    public String getTaskDir() {
+        return taskDir;
+    }
+}

local-spark/PidProcessDestroyer.java

@@ -0,0 +1,88 @@
+package com.primeton.datainsight.task;
+
+import org.apache.commons.exec.ProcessDestroyer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Field;
+
+/**
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2019/6/3
+ * Time: 18:40
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+public class PidProcessDestroyer implements ProcessDestroyer {
+    public static final String PID = "pid";
+
+
+    private static Logger logger = LoggerFactory.getLogger(PidProcessDestroyer.class);
+
+
+    final String taskId;
+
+
+
+
+    /**
+     * 进程 ID， 如果没有获取到取值 -1
+     */
+    private int processId = -1;
+
+
+    /**
+     * 根据 TaskId 更新 Task 状态
+     * @param taskId
+     */
+    public PidProcessDestroyer(String taskId) {
+        this.taskId = taskId;
+    }
+
+
+    /**
+     * 获取子进程内 Process PID
+     *
+     * @param process
+     * @return
+     */
+    public static int getProcessId(Process process) {
+        int processId = 0;
+        try {
+            Field f = process.getClass().getDeclaredField(PID);
+            f.setAccessible(true);
+            processId = f.getInt(process);
+        } catch (Throwable e) {
+            logger.error(e.getMessage(), e);
+        }
+        return processId;
+    }
+
+    @Override
+    public boolean add(Process process) {
+        processId = getProcessId(process);
+        logger.info("task {} state RUNNING pid {}", taskId, processId);
+        //taskHistService.updateTaskPid(taskId, "RUNNING", processId);
+        return true;
+    }
+
+    @Override
+    public boolean remove(Process process) {
+        return true;
+    }
+
+    @Override
+    public int size() {
+        return 1;
+    }
+
+
+    public int getPid() {
+        return processId;
+    }
+}

local-spark/SparkJobOutputStream.java

@@ -0,0 +1,78 @@
+package com.primeton.datainsight.task;
+
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * <p>
+ * </p>
+ * Created by zhaopx on 2018/1/12 0012-18:39
+ */
+public class SparkJobOutputStream extends LogOutputStream implements Closeable {
+
+
+
+    /**
+     * 从 console log 中获取 appid 的 正则
+     */
+    final String pa = "Client: Application report for (application_\\d+_\\d+) \\(state: (ACCEPTED|RUNNING|FINISHED|FAILED|KILLED)\\)";
+
+
+    /**
+     * Yarn App
+     */
+    final Pattern YARN_APPID_PATTERN = Pattern.compile(pa);
+
+
+
+    /**
+     * 提交的应用 Yarn 程序ID
+     */
+    String appId;
+
+
+    /**
+     * 程序的状态
+     */
+    String state = "UNKNOWN";
+
+
+    public SparkJobOutputStream(String baseDir, String jobId, String taskId) {
+        super(baseDir, jobId, taskId);
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+        String line = new String(b, off, len, charset);
+        super.write(line);
+        Matcher matcher = YARN_APPID_PATTERN.matcher(line);
+        if(matcher.find()){
+            String appId = matcher.group(1);
+            String state = matcher.group(2);
+            setAppId(appId);
+            setState(state);
+            return;
+        }
+    }
+
+    public String getAppId() {
+        return appId;
+    }
+
+    public String getState() {
+        return state;
+    }
+
+    public void setAppId(String appId) {
+        this.appId = appId;
+    }
+
+    public void setState(String state) {
+        this.state = state;
+        //JobQueue.runningTask(getAppId(), jobId, taskId, state);
+    }
+
+}

local-spark/SparkNativeLocalTask.java

@@ -0,0 +1,547 @@
+package com.primeton.datainsight.task;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Sets;
+import com.primeton.datainsight.configure.TaskProperties;
+import com.primeton.datainsight.exec.SparkTaskSubmits;
+import com.primeton.datainsight.job.IJob;
+import com.primeton.datainsight.service.WarnService;
+import com.primeton.datainsight.support.JobListener;
+import com.primeton.datainsight.support.ProcessPublisher;
+import com.primeton.datainsight.utils.ClasspathPackageScanner;
+import com.primeton.datainsight.utils.Constance;
+import com.primeton.datainsight.utils.NetworkInterfaceManager;
+import org.apache.commons.exec.CommandLine;
+import org.apache.commons.exec.DefaultExecutor;
+import org.apache.commons.exec.ExecuteWatchdog;
+import org.apache.commons.exec.PumpStreamHandler;
+import org.apache.commons.lang.ClassUtils;
+import org.apache.commons.lang.StringUtils;
+import org.joda.time.DateTime;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.UUID;
+
+
+/**
+ * <p>
+ * </p>
+ * Created by zhaopx on 2018/1/3 0003-18:06
+ */
+public class SparkNativeLocalTask extends SparkTask {
+
+
+    private static Logger LOG = LoggerFactory.getLogger(SparkNativeLocalTask.class);
+
+    /**
+     * 公共的add的类列表
+     */
+    private final List<String> commonDependList;
+
+
+    /**
+     * 任务的JobId
+     */
+    protected final String jobId;
+
+    /**
+     * 前端传输的taskId
+     */
+    protected final String originTaskId;
+
+
+    /**
+     * 入口类
+     */
+    protected final String jobClassStr;
+
+
+    /**
+     * Spark App Name
+     */
+    protected final String appName;
+
+
+    /**
+     * 传递给 Task 的参数
+     */
+    protected final Map<String, String> taskParams;
+
+    /**
+     * 实现接口的类
+     */
+    protected Class jobClass = null;
+
+
+    /**
+     * 告警服务
+     */
+    WarnService warnService;
+
+
+    JobListener jobListener;
+
+
+
+    ProcessPublisher processPublisher;
+
+
+    /**
+     * 任务参数配置
+     */
+    TaskProperties taskProperties;
+
+
+    /**
+     * dependency jar
+     */
+    protected java.net.URL[] dependJars = null;
+
+
+    /**
+     * 不关心 originTaskId，采用 UUID 生成
+     * @param jobId
+     * @param taskParams 提供给App的入口参数
+     */
+    public SparkNativeLocalTask(String jobId,
+                                String jobClass,
+                                String appName,
+                                Map<String, String> taskParams) {
+        this(jobId, jobClass, appName, UUID.randomUUID().toString().replaceAll("-", ""), taskParams);
+    }
+
+
+    /**
+     * 传入 dataFile 地址， file or hdfs
+     * @param originTaskId 这里的 originID 是指 taskInstanceId。 本质上Spark运行的ID和前面指定的ID是无法直接使用的，这里理解为做个关联
+     * @param taskParams 提供给App的入口参数
+     */
+    public SparkNativeLocalTask(String jobId,
+                                String jobClass,
+                                String appName,
+                                String originTaskId,
+                                Map<String, String> taskParams) {
+        super(StringUtils.isBlank(originTaskId) ? UUID.randomUUID().toString().replaceAll("-", "") : originTaskId);
+        this.jobId = jobId;
+        this.originTaskId = getTaskId();
+        this.jobClassStr = jobClass;
+        this.appName = appName;
+        this.taskParams = Optional.ofNullable(taskParams).orElse(new HashMap<>());
+        ImmutableList.Builder<String> builder = ImmutableList.<String>builder();
+        //String appLibHome = ServerConfig.getAppHome() + "/lib/";
+        try {
+            // 是绝对路径，无需加前缀
+            String fastJsonJar = ClasspathPackageScanner.findContainingJar(JSON.class);
+            builder.add(fastJsonJar);
+        } catch (Exception e) {
+            LOG.warn("find data3c commom jar, server jar, fastjson jar error.", e);
+        }
+
+        this.commonDependList = builder.build();
+    }
+
+
+
+    @Override
+    public void doExecute() throws Exception {
+        JSONObject params = new JSONObject();
+        params.put("id", jobId);
+        params.put("originTaskId", originTaskId);
+        params.put("jobClass", jobClassStr);
+        params.put("rpcPort", 8088);
+        params.put("jobName", appName);
+
+        params.put("master", taskProperties.getMaster());
+        params.put("deployMode", taskProperties.getDeployMode());
+        params.put("queue", taskProperties.getQueue());
+        params.put("appName", appName);
+        params.put("files", taskProperties.getFiles());
+
+        execute(params);
+    }
+
+    private void execute(Map<String, Object> params) throws Exception {
+        JSONObject map = (JSONObject)params;
+        final String jobId = map.getString("id");
+        // 前端的taskId，因为在任务中需要
+        final String originTaskId = map.getString("originTaskId");
+
+        // quartz job name, 跟 jobId 有关系
+        LOG.info("Start SparkJob {} at: {}", jobId, DateTime.now().toString("yyyy-MM-dd HH:mm:ss"));
+
+        final String jobClass = map.getString("jobClass");
+
+        // TASK UUID
+        String uuid = UUID.randomUUID().toString().replaceAll("-", "");
+        try {
+            Set<String> dependJars = new HashSet<>();
+            if(StringUtils.isNotBlank(taskProperties.getJars())) {
+                String[] split = taskProperties.getJars().split(",");
+                dependJars.addAll(Sets.newHashSet(split));
+            }
+            List<java.net.URL> jars = getJars(jobId, dependJars);
+            if(!dependJars.isEmpty()){
+                this.dependJars = jars.toArray(new java.net.URL[jars.size()]);
+                if(!jars.isEmpty()) {
+                    LOG.info("add jars: " + Arrays.toString(this.dependJars));
+                    // URLClassLoader， 加载外部的 lib 资源，用于动态加载，部分job 在系统上add 后不用重启
+                    java.net.URLClassLoader classLoader = new java.net.URLClassLoader(this.dependJars, this.getClass().getClassLoader());
+                    this.jobClass = (Class<? extends IJob>) ClassUtils.getClass(classLoader, jobClass, false);
+                }
+            }
+
+            if(this.jobClass == null) {
+                this.jobClass = (Class<? extends IJob>) ClassUtils.getClass(jobClass, false);
+            }
+        } catch (Exception e) {
+            throw new IllegalStateException(e);
+        }
+
+        String serverHost = NetworkInterfaceManager.INSTANCE.getLocalHostAddress();
+        if(StringUtils.isBlank(serverHost)){
+            throw new IllegalArgumentException("unknown data3c server. env[DATA3C_HOST] or sysprop[data3c.host] to set.");
+        }
+
+        // RPC 反向回调的地址
+        String endPoint = serverHost + ":" + map.getIntValue("rpcPort");
+        String[] args = this.parseArgs(jobId, originTaskId, map, endPoint);
+
+        try {
+            JSONObject object = new JSONObject();
+            object.put("finalStatus", "UNDEFINED");
+            object.put("appState", "NEW");
+            object.put("source", "data3c");
+            object.put("jobType", "SPARK");
+            object.put("taskName", map.getString("jobName"));
+            object.put("uuid", uuid);
+            object.put("__ORIGIN_TASK_ID", originTaskId);
+
+            jobListener.call(jobId, originTaskId, "NEW", object);
+            processPublisher.publish(Constance.JOB_PROCESS_TOPIC, object.toJSONString());
+            // 把正在运行的任务放置到队列
+            JobQueue.pushNewTask(jobId, originTaskId, uuid, map, jobListener);
+
+
+            int exitCode = runAsRunTimeProcess(jobId,
+                    originTaskId,
+                    map,
+                    args);
+            if(exitCode != 0) {
+                // 执行异常
+                throw new IllegalStateException("Spark Task Invalid exitCode: " + exitCode);
+            }
+            JSONObject result = new JSONObject();
+            result.put("source", "data3c");
+            result.put("jobType", "SPARK");
+            result.put("exitCode", exitCode);
+            object.put("appState", "SUCCEEDED");
+            result.put("finalStatus", "FINISHED");
+            result.put("taskName", map.getString("jobName"));
+            result.put("process", 1);
+            result.put("uuid", uuid);
+            result.put("__ORIGIN_TASK_ID", originTaskId);
+            jobListener.call(jobId, originTaskId, "SUCCEEDED", result);
+            processPublisher.publish(Constance.JOB_PROCESS_TOPIC, result.toJSONString());
+            LOG.info("{}/{} 运行结束。 STATE: {}", jobId, originTaskId, result.toJSONString());
+        } catch (Exception e) {
+            JSONObject error = new JSONObject();
+            error.put("source", "data3c");
+            error.put("jobType", "SPARK");
+            error.put("appState", "FAILED");
+            error.put("finalStatus", "FAILED");
+            error.put("taskName", map.getString("jobName"));
+            error.put("message", e.getMessage());
+            error.put("uuid", uuid);
+            error.put("__ORIGIN_TASK_ID", originTaskId);
+            //error.put("error", ExceptionUtils.getFullStackTrace(e));
+            jobListener.call(jobId, originTaskId, "FAILED", error);
+            processPublisher.publish(Constance.JOB_PROCESS_TOPIC, error.toJSONString());
+            JobQueue.removeErrorJob(jobId, originTaskId);
+            LOG.error("run job error: ", e);
+        }
+    }
+
+
+    /**
+     * 调用系统命令，启动Job
+     *
+     * @param args
+     * @return 0 正常运行结束，返回0； 不正常的结束，返回小于 0
+     */
+    private int runAsRunTimeProcess(final String jobId,
+                                    final String taskId,
+                                    final JSONObject taskConfig,
+                                    String[] args){
+        String execute = "spark-submit";
+
+        // 如果配置了SaprkHome，则从Spark Home 下读取spark-submit
+        if(System.getenv("SPARK_HOME") != null){
+            //execute = System.getenv("SPARK_HOME") + "bin" + execute;
+            String sparkHome = System.getenv("SPARK_HOME");
+            execute = StringUtils.join(Arrays.asList(sparkHome, "bin", execute).iterator(), File.separator);
+        }
+        String cmd = execute + " " + StringUtils.join(args, " ");
+        LOG.info("execute command: " + cmd);
+        try (SparkJobOutputStream out = new SparkJobOutputStream(taskProperties.getTaskLogDir(), jobId, taskId)){
+            PumpStreamHandler streamHandler = new PumpStreamHandler(out);
+
+            // 不限时间
+            final ExecuteWatchdog watchDog = new ExecuteWatchdog(ExecuteWatchdog.INFINITE_TIMEOUT);
+            DefaultExecutor executor = new DefaultExecutor();
+            executor.setExitValue(0);
+
+            PidProcessDestroyer pidProcessDestroyer = new PidProcessDestroyer(taskId);
+            executor.setStreamHandler(streamHandler);
+            executor.setWatchdog(watchDog);
+            executor.setProcessDestroyer(pidProcessDestroyer);
+            JobQueue.addTaskObserver(jobId, taskId, watchDog);
+
+            Map<String, String> environment = new HashMap<>(System.getenv());
+            environment.put("SPARK_SUBMIT_OPTS", taskProperties.getSparkSubmitOpts());
+            executor.setWorkingDirectory(new File(out.getTaskDir()));
+
+            // 是否开启任务
+            JobExecuteResultHandler handler = new JobExecuteResultHandler(
+                    warnService);
+            handler.setTaskName(taskConfig.getString("jobName"));
+            environment.put("TASK_DIR", out.getTaskDir());
+            executor.execute(CommandLine.parse(cmd), environment, handler);
+            handler.waitFor();
+            return handler.getExitValue();
+        } catch (Exception e) {
+            LOG.error(jobId + "/" + taskId + " 执行任务提交命令失败。");
+            throw new IllegalStateException(e.getMessage());
+        } finally {
+            JobQueue.removeTaskObserver(jobId, taskId);
+        }
+    }
+
+
+
+    /**
+     * 解析 Job 参数
+     * @param jobId 任务ID
+     * @param taskId 随机唯一ID
+     * @param taskConfig task 配置
+     * @param endPoint 反向推送RPC的端口号
+     * @return
+     */
+    protected String[] parseArgs(final String jobId, final String taskId, final JSONObject taskConfig, final String endPoint) {
+        String classInJar = null;
+        // 路径相同，不会出现多个
+        Set<String> jarPaths = new HashSet<>();
+
+        try {
+            // 添加第三方的依赖jar
+            jarPaths.addAll(commonDependList);
+            if(dependJars != null) {
+                for (java.net.URL jar : dependJars) {
+                    jarPaths.add(jar.getPath());
+                }
+            }
+
+            classInJar = ClasspathPackageScanner.findContainingJar(jobClass);
+            //jarPaths.add(classInJar);
+            jarPaths.remove(classInJar);
+        } catch (Exception e) {
+            throw new IllegalArgumentException(e);
+        }
+
+        List<String> list = new ArrayList<>();
+        list.add("--master");
+        list.add("'"+ taskConfig.getString("master") + "'");
+
+        // local 模式，用于测试，不需要提交 deployMode 参数
+        // 当为 yarn 是存在 yarn-client 和 yarn-cluster 两种模式
+        if(StringUtils.equals(taskConfig.getString("master"), "yarn")) {
+            list.add("--deploy-mode");
+            list.add(taskConfig.getString("deployMode"));
+        }
+
+        // json 參數
+        JSONObject parameters = (JSONObject) taskConfig.get("parameters");
+        if(parameters == null) {
+            parameters = new JSONObject();
+        }
+
+        list.add("--name");
+        String name = getTaskName(taskConfig);
+        list.add(name);
+
+        list.add("--class");
+        boolean amcImplJob = true;
+        if(IJob.class.isAssignableFrom(jobClass)) {
+            list.add(SparkTaskSubmits.class.getName());
+            //集成了 AMC 的任务
+            amcImplJob = true;
+        } else {
+            //list.add(taskConfig.getString("jobClass"));
+            list.add(jobClass.getName());
+            // 原生任务
+            amcImplJob = false;
+        }
+
+        if(StringUtils.isNotBlank(taskProperties.getJvmOpts())) {
+            list.add(taskProperties.getJvmOpts());
+        }
+
+        list.add("--queue");
+        list.add(Optional.ofNullable(taskConfig.getString("queue")).orElse("default"));
+
+        list.add("--conf");
+        list.add("spark.app.name=" + name);
+        Set<String> confArgs = new HashSet<>();
+        if(StringUtils.isNotBlank(taskProperties.getSparkParameters())) {
+            String[] confSplit = taskProperties.getSparkParameters().split(",");
+            confArgs.addAll(Sets.newHashSet(confSplit));
+        }
+        if(confArgs.size() > 0) {
+            for (String confArg : confArgs) {
+                list.add("--conf");
+                list.add(confArg);
+            }
+        }
+        // 添加文件
+        String addFiles = (String)taskConfig.get("files");
+        if(StringUtils.isNotBlank(addFiles)) {
+            String[] files = addFiles.split(",");
+            Set<String> fileSet = new HashSet<>(files.length);
+            if(files.length > 0) {
+                for (String file : files) {
+                    File file1 = new File(file);
+                    if(file1.exists() && file1.isFile()) {
+                        fileSet.add(file1.getAbsolutePath());
+                        LOG.info("add spark file: {}", file1.getAbsolutePath());
+                    }
+                }
+            }
+            if(fileSet.size() > 0) {
+                list.add("--files");
+                list.add(StringUtils.join(fileSet, ","));
+            }
+        }
+
+        // 解决了 在 lib 包下，又在 jobs 包下的任务包。优先使用 jobs 下的
+        String jobJarName = new File(classInJar).getName();
+        String jobJar = classInJar;
+        Set<String> dependJarsPath = new HashSet<>();
+        for (String jarPath : jarPaths) {
+            File jarFile = new File(jarPath);
+            if(jarFile.exists() && jarFile.getName().equals(jobJarName)) {
+                // 如果 jobs 下有和 lib 下的同名jar，则使用 jobs 下面的。
+                jobJar = jarPath;
+            } else {
+                dependJarsPath.add(jarPath);
+            }
+        }
+        list.add("--jars");
+        String dependJars = StringUtils.join(dependJarsPath, ",");
+        list.add(dependJars);
+        list.add(jobJar);
+
+        for (String jarPath : jarPaths) {
+            LOG.info("add jar {}", jarPath);
+        }
+
+
+        // dataFile 文件地址，作为参数传入
+        for (Map.Entry<String, String> taskParam : this.taskParams.entrySet()) {
+            parameters.put(taskParam.getKey(), taskParam.getValue());
+        }
+        parameters.put("__JOB_ID", jobId);
+        parameters.put("__TASK_ID", taskId);
+        if(StringUtils.isNotBlank(originTaskId)){
+            parameters.put("__ORIGIN_TASK_ID", originTaskId);
+        }
+
+        if(StringUtils.isNotBlank(taskProperties.getAppParameters())) {
+            list.add(taskProperties.getAppParameters());
+        } else {
+            // 这个之后的是程序参数，args，第0个 参数
+            list.add(jobClass.getName());
+
+            org.apache.commons.codec.binary.Base64 base64 = new org.apache.commons.codec.binary.Base64(1, ":".getBytes());
+            // 第1个 参数
+            list.add(base64.encodeToString(parameters.toJSONString().getBytes(StandardCharsets.UTF_8)));
+
+            // 第2个 参数
+            list.add(endPoint);
+        }
+
+        LOG.info("args: " + StringUtils.join(list.iterator(), " "));
+        return list.toArray(new String[list.size()]);
+    }
+
+    public void setWarnService(WarnService warnService) {
+        this.warnService = warnService;
+    }
+
+    public void setJobListener(JobListener jobListener) {
+        this.jobListener = jobListener;
+    }
+
+    public void setProcessPublisher(ProcessPublisher processPublisher) {
+        this.processPublisher = processPublisher;
+    }
+
+    public void setTaskProperties(TaskProperties taskProperties) {
+        this.taskProperties = taskProperties;
+    }
+
+    /**
+     * 获取 Task name
+     * @param taskConfig Task Config
+     * @return 返回任务名称
+     */
+    protected String getTaskName(final JSONObject taskConfig) {
+        String name = taskConfig.getString("jobName");
+        if(StringUtils.isBlank(name) || "null".equalsIgnoreCase(name)) {
+            name = jobClass.getSimpleName();
+        }
+
+        // 把前端的任务名称追加到yarn的name上，yarn上好识别
+        String forceTaskName = taskConfig.getString("taskName");
+        if(StringUtils.isNotBlank(forceTaskName)){
+            name = name + "：" + forceTaskName;
+        }
+        return name;
+    }
+
+
+
+    public List<java.net.URL> getJars(String jobId, Set<String> dependJars) throws Exception {
+        List<java.net.URL> jars = new ArrayList<>(dependJars.size());
+        for (String dependJar : dependJars) {
+            File file = new File(dependJar);
+            if (file.exists() && file.isFile()) {
+                jars.add(file.toURL());
+                continue;
+            } else {
+                String jobJarDir = taskProperties.getSparkLibDir();
+                // 单个文件, 找不到
+                file = new File(jobJarDir, dependJar);
+                if (file.exists() && file.isFile()) {
+                    jars.add(file.toURL());
+                    continue;
+                }
+            }
+
+            LOG.warn("{} not exists, skip file.", dependJar);
+        }
+        return jars;
+    }
+
+}

nacos-config/DataTagApplication.java

@@ -0,0 +1,55 @@
+package com.primeton.datatag;
+
+import com.alibaba.nacos.api.config.ConfigType;
+import com.alibaba.nacos.api.config.annotation.NacosValue;
+import com.alibaba.nacos.spring.context.annotation.config.NacosPropertySource;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.HashMap;
+import java.util.Map;
+
+@NacosPropertySource(dataId = "damp", type = ConfigType.YAML, autoRefreshed = true)
+@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
+@RestController
+public class DataTagApplication {
+
+
+    @Value(value = "${hadoop.nameNodeUrl}")
+    String hadoopUrl;
+
+
+    @Value(value = "${hbase.url}")
+    String hbaseUrl;
+
+    @RequestMapping(value = "/hi", method = {RequestMethod.POST, RequestMethod.GET})
+    @ResponseBody
+    public Map<String, Object> sayHello(@RequestParam("name") String name) {
+        final HashMap<String, Object> map = new HashMap<>();
+        map.put("name", name);
+        map.put("massage", "hello, " + name);
+        map.put("hadoopUrl", hadoopUrl);
+        map.put("hbaseUrl", hbaseUrl);
+        return map;
+    }
+
+
+    public static void main(String[] args) {
+        SpringApplication.run(DataTagApplication.class, args);
+        System.out.println("/***************************************************/");
+        System.out.println("/*                                                 */");
+        System.out.println("/*(♥◠‿◠)ﾉﾞ   cdhExecutor启动成功         ლ(´ڡ`ლ)ﾞ */");
+        System.out.println("/*                                                 */");
+        System.out.println("/***************************************************/");
+    }
+
+
+
+}

nacos-config/bootstrap.yml

@@ -0,0 +1,32 @@
+nacos:
+  discovery:
+    server-addr: localhost:8848
+  config:
+    prefix: damp
+    bootstrap:
+      # 开启预加载配置
+      enable: true
+    # 服务地址
+    server-addr: 127.0.0.1:8848
+    # 服务账号
+    username: nacos
+    # 服务密码
+    password: nacos
+    # data-id
+    data-id: damp
+    # group
+    group: DEFAULT_GROUP
+    # 命名空间
+    # namespace: '刚刚自己新建的命名空间ID，没有新建不需要配置namaspace'
+    # 配置文件类型
+    type: yaml
+    file-extension: yaml
+    # 最大重试次数
+    max-retry: 10
+    # 自动刷新
+    auto-refresh: true
+    # 重试时间
+    config-retry-time: 2000
+    # 监听长轮询超时时间
+    config-long-poll-timeout: 46000
+

nacos-config/pom.xml

@@ -0,0 +1,84 @@
+<?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>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.1.5.RELEASE</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>datatag</artifactId>
+    <packaging>jar</packaging>
+
+    <properties>
+        <lombok.version>1.18.4</lombok.version>
+        <guava.version>27.0.1-jre</guava.version>
+    </properties>
+    <dependencies>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>${guava.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba.boot</groupId>
+            <artifactId>nacos-config-spring-boot-starter</artifactId>
+            <version>0.2.7</version>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.google.guava</groupId>
+                    <artifactId>guava</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <!-- Spring Boot依赖 -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-configuration-processor</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <!-- 引入jdbc支持 -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+        </dependency>
+        <!-- 引入log4j2依赖 -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-log4j2</artifactId>
+        </dependency>
+
+        <!-- lombok插件 -->
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>${lombok.version}</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <!-- FastJson -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>1.2.73</version>
+        </dependency>
+    </dependencies>
+
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

sql-template-file/HibernateLocalSessionFactoryBean.java

@@ -0,0 +1,147 @@
+package com.primeton.dgs.kernel.core.dao.hibernate;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.SQLException;
+import java.util.Map;
+import java.util.Properties;
+
+import com.primeton.dgs.kernel.core.util.MetadataWebContext;
+import com.primeton.dgs.kernel.core.web.springframework.DatabaseRecognizer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.context.ApplicationContextException;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.orm.hibernate5.LocalSessionFactoryBean;
+
+import javax.sql.DataSource;
+
+
+/**
+ * Hibernate 配置。
+ *
+ * 加载不同的 Hibernate 配置，根据 DatabaseRecognizer 返回的数据库类别，初始化不同的 Hibernate 方言
+ *
+ * @author zhaopx
+ *
+ */
+public class HibernateLocalSessionFactoryBean extends LocalSessionFactoryBean {
+	private static final String PREFIX = "META-INF/hibernate/"; // 加载hibernate模型文件的路径前缀
+	private static final String ENDFIX = "*.hbm.xml"; // 文件名的扩展后缀名
+
+
+	private DatabaseRecognizer databaseRecognizer;
+
+	// 数据库的品牌和版本
+	private String databaseProductName;
+
+
+	private String databaseProductVersion;
+
+	private static final Logger logger = LoggerFactory.getLogger(HibernateLocalSessionFactoryBean.class);
+
+
+	@Override
+	public void setDataSource(DataSource dataSource) {
+		super.setDataSource(dataSource);
+		try(Connection conn = dataSource.getConnection()) {
+			DatabaseMetaData meta = conn.getMetaData();
+			String databaseName = meta.getDatabaseProductName();
+			String productVersion = meta.getDatabaseProductVersion();
+			if (logger.isInfoEnabled()) {
+				logger.info("Current Database is " + databaseName + ", Version is " + productVersion);
+			}
+			databaseName = recognizeDatabaseName(databaseName, productVersion);
+			databaseRecognizer.setCurrentDatabaseProductName(databaseName);
+			databaseRecognizer.setCurrentDatabaseProductVersion(productVersion);
+
+			// 将数据库名缓存起来
+			MetadataWebContext.getInstance().setDatabaseName(databaseName);
+		} catch (SQLException e) {
+			String s = "Cannot determine the database brand name for loading ApplicationContext";
+			logger.error(s, e);
+			throw new ApplicationContextException(s, e);
+		}
+	}
+
+	/**
+	 * 覆写该方法,将hibernate配置文件,通过读取特定类路径下的文件获取
+	 */
+	@Override
+	public void setMappingResources(String mappingResources[]) {
+		try {
+			ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
+			Resource[] resources = resolver.getResources("classpath*:".concat(PREFIX).concat(ENDFIX));
+			String[] resourceStrings = new String[resources.length];
+			for(int index = 0; index < resources.length; index++){
+				resourceStrings[index] = PREFIX.concat(resources[index].getFilename());
+			}
+			super.setMappingResources(resourceStrings);
+		} catch (IOException e) {
+			throw new IllegalStateException("无法加载 " + ENDFIX + " 文件。", e);
+		}
+    }
+
+
+	@Override
+	public void afterPropertiesSet() throws IOException {
+		// 首先根据数据库选择配置文件
+		String databaseName = MetadataWebContext.getInstance().getDatabaseName();
+		Properties props = new Properties();
+		String configFile = "classpath*:".concat("/spring/hibernate_").concat(databaseName.toLowerCase()).concat("_local.properties");
+		try {
+			ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
+			Resource[] resources = resolver.getResources(configFile);
+			for (Resource resource : resources) {
+				logger.info("found hibernate local file: {}", resource.getURL());
+				try(InputStream inputStream = resource.getInputStream()) {
+					Properties tmp = new Properties();
+					tmp.load(inputStream);
+					props.putAll(tmp);
+				}
+			}
+		} catch (Exception e) {
+			logger.error("无法读取配置文件: " + configFile, e);
+		}
+
+		if(!props.isEmpty()) {
+			for (Map.Entry<Object, Object> entry : props.entrySet()) {
+				getHibernateProperties().put(entry.getKey(), entry.getValue());
+			}
+		}
+
+		logger.info("hibernate properties print as ");
+		for (Map.Entry<Object, Object> entry : getHibernateProperties().entrySet()) {
+			logger.info("{}: {}", entry.getKey(), entry.getValue());
+		}
+		super.afterPropertiesSet();
+	}
+
+	/**
+	 * 识别数据库名称，已经配置在spring/corebean/context-base.xml文件中
+	 * @param databaseName 从数据库连接获取的数据库名称
+	 * @param productVersion 数据库版本
+	 * @return 转换成系统需要的名称
+	 */
+	private String recognizeDatabaseName(String databaseName, String productVersion) {
+		this.databaseProductName = databaseName;
+		this.databaseProductVersion = productVersion;
+		return databaseRecognizer.getDatabaseName(databaseName);
+	}
+
+	public void setDatabaseRecognizer(DatabaseRecognizer databaseRecognizer) {
+		this.databaseRecognizer = databaseRecognizer;
+	}
+
+	public String getDatabaseProductName() {
+		return databaseProductName;
+	}
+
+	public String getDatabaseProductVersion() {
+		return databaseProductVersion;
+	}
+}

sql-template-file/HibernateStatementCompareField.java

@@ -0,0 +1,135 @@
+package com.primeton.dgs.kernel.core.dao.hibernate;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class HibernateStatementCompareField extends HibernateStatementField
+{
+  private static final long serialVersionUID = 1L;
+  private String compare;
+  private String compareProperty;
+  private String compareValue;
+  public static final String AVAILABLE_COMPARE = "eq,ne,gt,ge,lt,le";
+
+  public String getCompare()
+  {
+    return this.compare;
+  }
+
+  public void setCompare(String compare) {
+    this.compare = compare;
+  }
+
+  public String getCompareProperty() {
+    return this.compareProperty;
+  }
+
+  public void setCompareProperty(String compareProperty) {
+    this.compareProperty = compareProperty;
+  }
+
+  public String getCompareValue() {
+    return this.compareValue;
+  }
+
+  public void setCompareValue(String compareValue) {
+    this.compareValue = compareValue;
+  }
+
+  public HibernateStatementCompareField()
+  {
+  }
+
+  public HibernateStatementCompareField(String compare, String compareProperty, String compareValue) {
+    this.compare = compare;
+    this.compareProperty = compareProperty;
+    this.compareValue = compareValue;
+  }
+
+  private static int getCompareOperateCode(String compare)
+  {
+    String[] ac = "eq,ne,gt,ge,lt,le".split(",");
+    for (int i = 0; i < ac.length; i++) {
+      if (ac[i].equalsIgnoreCase(compare)) {
+        return i;
+      }
+    }
+    return -1;
+  }
+
+  public static boolean isAvailableCompare(String compare)
+  {
+    return getCompareOperateCode(compare) >= 0;
+  }
+
+  private static boolean match(String compare, Object value, String compareValue)
+    throws NumberFormatException
+  {
+    if (value == null) {
+      return false;
+    }
+
+    int reslt = -1;
+    if ((value instanceof String)) {
+      reslt = ((String)value).compareTo(compareValue);
+    } else if ((value instanceof Integer)) {
+      reslt = ((Integer)value).compareTo(new Integer(compareValue));
+    } else if ((value instanceof Long)) {
+      reslt = ((Long)value).compareTo(new Long(compareValue));
+    } else if ((value instanceof Float)) {
+      reslt = ((Float)value).compareTo(new Float(compareValue));
+    } else if ((value instanceof Double)) {
+      reslt = ((Double)value).compareTo(new Double(compareValue));
+    } else if ((value instanceof Short)) {
+      reslt = ((Short)value).compareTo(new Short(compareValue));
+    } else if ((value instanceof BigInteger)) {
+      reslt = ((BigInteger)value).compareTo(new BigInteger(compareValue));
+    } else if ((value instanceof BigDecimal)) {
+      reslt = ((BigDecimal)value).compareTo(new BigDecimal(compareValue));
+    } else if ((value instanceof Boolean)) {
+      int intBool = ((Boolean)value).booleanValue() ? 1 : 0;
+      int intCompareBool = Boolean.valueOf(compareValue).booleanValue() ? 1 : 0;
+      reslt = intBool - intCompareBool;
+    }
+
+    int code = getCompareOperateCode(compare);
+    if (reslt == 0) {
+      if ((code == 0) || (code == 3) || (code == 5))
+        return true;
+    }
+    else if (reslt > 0) {
+      if ((code == 1) || (code == 2) || (code == 3))
+        return true;
+    }
+    else if ((reslt < 0) && (
+      (code == 1) || (code == 4) || (code == 5))) {
+      return true;
+    }
+
+    return false;
+  }
+
+  @Override
+  public StringBuffer getRawText(Object vo)
+  {
+    return getText(vo, true, new AtomicInteger(-1));
+  }
+
+
+  @Override
+  public StringBuffer getText(Object vo, AtomicInteger index)
+  {
+    return getText(vo, false, index);
+  }
+
+  private StringBuffer getText(Object vo, boolean isRaw, AtomicInteger index)
+  {
+    StringBuffer sb = null;
+    Object actualValue = getBeanProperty(vo, this.compareProperty);
+    if (match(this.compare, actualValue, this.compareValue)) {
+      sb = isRaw ? getChildrenRawText(vo) : getChildrenText(vo, index);
+    }
+    return sb;
+  }
+}

sql-template-file/HibernateStatementDynamicField.java

@@ -0,0 +1,67 @@
+package com.primeton.dgs.kernel.core.dao.hibernate;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class HibernateStatementDynamicField extends HibernateStatementField
+{
+  private static final long serialVersionUID = 1L;
+  private String property;
+  private String onEmpty;
+
+  public String getProperty()
+  {
+    return this.property;
+  }
+
+  public void setProperty(String property) {
+    this.property = property;
+  }
+
+  public String getOnEmpty() {
+    return this.onEmpty;
+  }
+
+  public void setOnEmpty(String onEmpty) {
+    this.onEmpty = onEmpty;
+  }
+
+  public boolean onEmpty()
+  {
+    return ("true".equalsIgnoreCase(this.onEmpty)) || ("yes".equalsIgnoreCase(this.onEmpty));
+  }
+
+  public HibernateStatementDynamicField()
+  {
+  }
+
+  public HibernateStatementDynamicField(String property, String onEmpty)
+  {
+    this.property = property;
+    this.onEmpty = onEmpty;
+  }
+
+  public StringBuffer getRawText(Object vo)
+  {
+    return getText(vo, true, new AtomicInteger(-1));
+  }
+
+
+  @Override
+  public StringBuffer getText(Object vo, AtomicInteger index)
+  {
+    return getText(vo, false, index);
+  }
+
+  private StringBuffer getText(Object vo, boolean isRaw, AtomicInteger index)
+  {
+    StringBuffer sb = null;
+    boolean isEmptyValue = isEmpty(getBeanProperty(vo, this.property));
+    if ((onEmpty()) && (isEmptyValue)) {
+      sb = isRaw ? getChildrenRawText(vo) : getChildrenText(vo, index);
+    }
+    if ((!onEmpty()) && (!isEmptyValue)) {
+      sb = isRaw ? getChildrenRawText(vo) : getChildrenText(vo, index);
+    }
+    return sb;
+  }
+}

sql-template-file/HibernateStatementField.java

@@ -0,0 +1,110 @@
+package com.primeton.dgs.kernel.core.dao.hibernate;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import org.apache.commons.beanutils.PropertyUtils;
+
+import com.primeton.dgs.kernel.core.ex.HibernateStatementException;
+
+public abstract class HibernateStatementField implements Serializable {
+	private static final long serialVersionUID = 1L;
+	public static final String SPACE = " ";
+	public static final Pattern PLACEHOLDER = Pattern.compile("#[\\w]+#");
+	public static final Pattern REPLACEMENT = Pattern.compile("\\$[\\w]+\\$");
+
+	protected List<HibernateStatementField> children = new ArrayList();
+
+	public List<HibernateStatementField> getChildren() {
+		return this.children;
+	}
+
+	protected StringBuffer getChildrenRawText(Object vo) {
+		StringBuffer sb = new StringBuffer();
+		for (Iterator it = this.children.iterator(); it.hasNext();) {
+			StringBuffer s = ((HibernateStatementField) it.next())
+					.getRawText(vo);
+			if ((s != null) && (s.length() > 0)) {
+				sb.append(s).append(" ");
+			}
+		}
+		return sb;
+	}
+
+	protected StringBuffer getChildrenText(Object vo, AtomicInteger index) {
+		StringBuffer sb = new StringBuffer();
+		for (Iterator it = this.children.iterator(); it.hasNext();) {
+			StringBuffer s = ((HibernateStatementField) it.next()).getText(vo, index);
+			if ((s != null) && (s.length() > 0)) {
+				sb.append(s).append(" ");
+			}
+		}
+		return sb;
+	}
+
+	public abstract StringBuffer getRawText(Object paramObject);
+
+	public abstract StringBuffer getText(Object paramObject, AtomicInteger index);
+
+	public List<Object> getParam(Object vo) {
+		StringBuffer sb = getRawText(vo);
+		if (sb == null) {
+			return null;
+		}
+		List result = new ArrayList();
+		Matcher m = PLACEHOLDER.matcher(sb);
+		while (m.find()) {
+			String placeholder = sb.substring(m.start(), m.end());
+			String property = placeholder
+					.substring(1, placeholder.length() - 1);
+			result.add(getBeanProperty(vo, property));
+		}
+		return result;
+	}
+
+	public Map<String, Object> getParamMap(Object vo) {
+		StringBuffer sb = getRawText(vo);
+		if (sb == null) {
+			return null;
+		}
+		Map result = new HashMap();
+		Matcher m = PLACEHOLDER.matcher(sb);
+		while (m.find()) {
+			String placeholder = sb.substring(m.start(), m.end());
+			String property = placeholder
+					.substring(1, placeholder.length() - 1);
+			result.put(property, getBeanProperty(vo, property));
+		}
+		return result;
+	}
+
+	protected Object getBeanProperty(Object vo, String property) {
+		if (vo == null)
+			return null;
+		try {
+			return PropertyUtils.getProperty(vo, property);
+		} catch (NoSuchMethodException e) {
+			return null;
+		} catch (Exception e) {
+			throw new HibernateStatementException("property \"" + property
+					+ "\" is missing!", e);
+		}
+
+	}
+
+	public static final boolean isEmpty(Object value) {
+		if (value == null) {
+			return true;
+		}
+		if ((value instanceof String)) {
+			return "".equals(((String) value).trim());
+		}
+		return false;
+	}
+}

sql-template-file/HibernateStatementSessionFactoryBean.java

@@ -0,0 +1,322 @@
+package com.primeton.dgs.kernel.core.dao.hibernate;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.jdom.Document;
+import org.jdom.Element;
+import org.jdom.JDOMException;
+import org.jdom.Text;
+import org.jdom.input.SAXBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+
+import com.primeton.dgs.kernel.core.ex.HibernateStatementException;
+import com.primeton.dgs.kernel.core.util.MetadataWebContext;
+
+
+public class HibernateStatementSessionFactoryBean implements Serializable {
+	private static final long serialVersionUID = 1L;
+	private static final String PREFIX = "META-INF/sqlxml/"; // 加载模块的SQL文件的路径前缀
+	private static final String ENDFIX = "*Sql.xml"; // 文件名的扩展后缀名
+
+	private HibernateStatementSessionFactoryBean parent;
+	private static HibernateStatementSessionFactoryBean instance;
+	private Map<String, List<HibernateStatementField>> statements;
+	private boolean ready = false;
+	private Resource[] statementLocations;
+	private boolean merge = false;
+	private static Logger log = LoggerFactory.getLogger(HibernateStatementSessionFactoryBean.class);
+	
+	public HibernateStatementSessionFactoryBean() {
+		instance = this;
+		this.statements = new HashMap();
+	}
+
+	public void setStatementResources(String[] statementResources) {
+		setStatementResources();
+	}
+
+	public void setStatementLocations(File[] statementLocations) {
+		setStatementResources();
+	}
+
+	public void setStatementLocations(Resource[] statementLocations) {
+		this.statementLocations = statementLocations;
+	}
+
+	public void setStatementResources(String statementResources) {
+		setStatementResources();
+	}
+
+	private void setStatementResources() {
+		try {
+			ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
+			statementLocations = resolver.getResources("classpath*:"
+					.concat(PREFIX)
+					.concat(MetadataWebContext.getInstance().getDatabaseName()
+							.toLowerCase()).concat("/").concat(ENDFIX));
+			System.setProperty("dataBaseName", MetadataWebContext.getInstance().getDatabaseName());
+		} catch (IOException e) {
+			log.error("加载 NativeSQL 异常。", e);
+		}
+	}
+
+	public Resource[] getStatementLocations() {
+		return this.statementLocations;
+	}
+
+	public boolean isReady() {
+		return this.ready;
+	}
+
+	public HibernateStatementSessionFactoryBean getParent() {
+		return this.parent;
+	}
+
+	public void setParent(HibernateStatementSessionFactoryBean parent) {
+		this.parent = parent;
+	}
+
+	public boolean isMerge() {
+		return this.merge;
+	}
+
+	public void setMerge(boolean merge) {
+		this.merge = merge;
+	}
+
+	public void init() throws IOException {
+		this.ready = false;
+		this.statements.clear();
+		for (int i = 0; i < this.statementLocations.length; i++) {
+			config(this.statementLocations[i].getInputStream());
+			log.info("read xml sql template file: {}", statementLocations[i].getURI());
+		}
+		if ((isMerge()) && (this.parent != null)) {
+			doMerge();
+		}
+		this.ready = true;
+	}
+
+	private void doMerge() {
+		if (this.parent != null)
+			for (Iterator it = this.statements.keySet().iterator(); it
+					.hasNext();) {
+				String key = (String) it.next();
+				if (this.parent.statements.containsKey(key)) {
+					throw new HibernateStatementException("statement of id=\""
+							+ key + "\" has bean defined!");
+				}
+				this.parent.statements.put(key, (List) this.statements.get(key));
+			}
+	}
+
+	public void config(InputStream in) throws IOException {
+		SAXBuilder sb = new SAXBuilder();
+		try {
+			Document doc = sb.build(in);
+			Element root = doc.getRootElement();
+			List statements = root.getChildren("statement");
+			for (int i = 0; i < statements.size(); i++) {
+				Element stmt = (Element) statements.get(i);
+				String id = stmt.getAttributeValue("id");
+				if ((id == null) || ("".equals(id.trim()))) {
+					throw new HibernateStatementException(
+							"statement''s \"id\" should not be empty!");
+				}
+				if (this.statements.containsKey(id)) {
+					throw new HibernateStatementException("statement of id=\""
+							+ id + "\" has bean defined!");
+				}
+
+				List fields = stmt.getContent();
+				this.statements.put(id, readFields(fields));
+			}
+		} catch (JDOMException e) {
+			throw new HibernateStatementException("statement file is error!", e);
+		} catch (IOException e) {
+			throw new HibernateStatementException("statement file not exist!",
+					e);
+		} finally {
+			if (in != null)
+				in.close();
+		}
+	}
+
+	private List<HibernateStatementField> readFields(List<?> fields) {
+		List result = new ArrayList();
+		for (int j = 0; j < fields.size(); j++) {
+			if ((fields.get(j) instanceof Text)) {
+				String text = ((Text) fields.get(j)).getTextTrim();
+				if (!"".equals(text)) {
+					result.add(new HibernateStatementTextField(text));
+				}
+			} else if ((fields.get(j) instanceof Element)) {
+				Element field = (Element) fields.get(j);
+				if ("dynamic-field".equals(field.getName())) {
+					result.add(readDynamicField(field));
+				} else if ("compare-field".equals(field.getName())) {
+					result.add(readCompareField(field));
+				} else {
+					throw new HibernateStatementException("\""
+							+ field.getName() + "\" is not supportable!");
+				}
+			}
+		}
+		return result;
+	}
+
+	private HibernateStatementDynamicField readDynamicField(Element field) {
+		String property = field.getAttributeValue("property");
+		String onEmpty = field.getAttributeValue("onEmpty");
+		if ((property == null) || ("".equals(property))) {
+			throw new HibernateStatementException(
+					"\"property\" of dynamic-field is required!");
+		}
+		HibernateStatementDynamicField f = new HibernateStatementDynamicField(
+				property, onEmpty);
+		List children = readFields(field.getContent());
+		f.getChildren().addAll(children);
+		return f;
+	}
+
+	private HibernateStatementCompareField readCompareField(Element field) {
+		String compare = field.getAttributeValue("compare");
+		String compareProperty = field.getAttributeValue("compareProperty");
+		String compareValue = field.getAttributeValue("compareValue");
+		if ((compare == null) || ("".equals(compare))) {
+			throw new HibernateStatementException(
+					"Attribute \"compare\" of compare-field is required!");
+		}
+		if ((compareProperty == null) || ("".equals(compareProperty))) {
+			throw new HibernateStatementException(
+					"Attribute \"compareProperty\" of compare-field is required!");
+		}
+		if ((compareValue == null) || ("".equals(compareValue))) {
+			throw new HibernateStatementException(
+					"Attribute \"compareValue\" of compare-field is required!");
+		}
+		if (!HibernateStatementCompareField.isAvailableCompare(compare)) {
+			throw new HibernateStatementException(
+					"Attribute \"compare\" of compare-field is unavailable, the available value is (eq,ne,gt,ge,lt,le)!");
+		}
+
+		HibernateStatementCompareField f = new HibernateStatementCompareField(
+				compare, compareProperty, compareValue);
+		List children = readFields(field.getContent());
+		f.getChildren().addAll(children);
+		return f;
+	}
+
+	public void destroy() {
+		this.statements.clear();
+		this.ready = false;
+	}
+
+	public void refresh() throws IOException {
+		destroy();
+		init();
+	}
+
+	public List<HibernateStatementField> getStatement(String id) {
+		List<HibernateStatementField> statementFields = this.statements.get(id);
+		if (statementFields == null) {
+			if (this.parent == null) {
+				log.error("statement [id=\"" + id+ "\"]: not existed");
+				/*throw new HibernateStatementException("statement [id=\"" + id
+						+ "\"]: not existed");*/
+				return null;
+			}
+			return this.parent.getStatement(id);
+		}
+		return (List) statementFields;
+	}
+
+	public String getText(String id) {
+		return getText(id, null);
+	}
+
+	public String getText(String id, Object vo) {
+		List<?> fileds = getStatement(id);
+		if(null == fileds){
+			return null;
+		}
+		StringBuffer sb = new StringBuffer();
+		AtomicInteger index = new AtomicInteger(0);
+		try {
+			for (int i = 0; i < fileds.size(); i++) {
+				StringBuffer s = ((HibernateStatementField) fileds.get(i)).getText(vo, index);
+				if ((s != null) && (s.length() > 0))
+					sb.append(s).append(" ");
+			}
+		} catch (NumberFormatException e) {
+			throw new HibernateStatementException("statement [id=\"" + id
+					+ "\"]: cannot format number " + e.getMessage());
+		} catch (Exception e) {
+			throw new HibernateStatementException("statement [id=\"" + id
+					+ "\"]: " + e.getMessage(), e);
+		}
+		return sb.toString();
+	}
+
+	public static String getStatementText(String id) {
+		return getStatementText(id, null);
+	}
+
+	public static String getStatementText(String id, Object vo) {
+		return instance.getText(id, vo);
+	}
+
+	public Object[] getParam(String id, Object vo) {
+		List fileds = Optional.ofNullable(getStatement(id)).orElse(new ArrayList<>(0));
+		List result = new ArrayList();
+		for (int i = 0; i < fileds.size(); i++) {
+			List list = ((HibernateStatementField) fileds.get(i)).getParam(vo);
+			if ((list != null) && (!list.isEmpty())) {
+				result.addAll(list);
+			}
+		}
+		return result.toArray();
+	}
+
+	public Map<String, Object> getParamMap(String id, Object vo) {
+		if (!this.statements.containsKey(id)) {
+			throw new HibernateStatementException("statement [id=\"" + id
+					+ "\"]: not existed");
+		}
+		Map result = new HashMap();
+		List fileds = getStatement(id);
+		for (int i = 0; i < fileds.size(); i++) {
+			Map map = ((HibernateStatementField) fileds.get(i)).getParamMap(vo);
+			if ((map != null) && (!map.isEmpty())) {
+				result.putAll(map);
+			}
+		}
+		return result;
+	}
+
+	public static Object[] getStatementParam(String id, Object vo) {
+		return instance.getParam(id, vo);
+	}
+
+	public static Map<String, Object> getStatementParamMap(String id, Object vo) {
+		return instance.getParamMap(id, vo);
+	}
+
+	public static boolean isStateReady() {
+		return instance.isReady();
+	}
+}

sql-template-file/HibernateStatementTextField.java

@@ -0,0 +1,111 @@
+package com.primeton.dgs.kernel.core.dao.hibernate;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.regex.Matcher;
+
+public class HibernateStatementTextField extends HibernateStatementField
+{
+  private static final long serialVersionUID = 1L;
+  private String value = "";
+
+  public String getValue() {
+    return this.value;
+  }
+
+  public void setValue(String value) {
+    this.value = value;
+  }
+
+  public HibernateStatementTextField()
+  {
+  }
+
+  public HibernateStatementTextField(String value) {
+    this.value = value;
+  }
+
+  @Override
+  public StringBuffer getText(Object vo, AtomicInteger index)
+  {
+    StringBuffer sb = getRawText(vo);
+    StringBuffer sbb = new StringBuffer();
+    Matcher m = PLACEHOLDER.matcher(sb);
+    while (m.find()) {
+      // add zhaopx， Hibernate5 需要在 ？ 后加一个索引值
+      String rep = (index.get() < 0 ? "?" : "?"+index);
+      m.appendReplacement(sbb, rep);
+      index.incrementAndGet();
+    }
+    m.appendTail(sbb);
+    return sbb;
+  }
+
+  public StringBuffer getRawText(Object vo)
+  {
+    StringBuffer sb = new StringBuffer();
+    Matcher m = REPLACEMENT.matcher(value);
+    while (m.find()) {
+      String group = m.group();
+      String property = group.substring(1, group.length() - 1);
+      StringBuffer replace = getReplacementText(getBeanProperty(vo, property));
+      if ((replace != null) && (replace.length() > 0)) {
+        m.appendReplacement(sb, replace.toString());
+      }
+    }
+    m.appendTail(sb);
+    return sb;
+  }
+
+  private StringBuffer getReplacementText(Object value)
+  {
+    if (value == null) {
+      return null;
+    }
+    if ((value instanceof Object[])) {
+      return getReplacementTextFromArray((Object[])value);
+    }
+    if ((value instanceof Collection)) {
+      return getReplacementTextFromCollection((Collection)value);
+    }
+    return new StringBuffer().append(value);
+  }
+
+  private StringBuffer getReplacementTextFromArray(Object[] array)
+  {
+    if ((array == null) || (array.length == 0)) {
+      return null;
+    }
+    StringBuffer sb = new StringBuffer();
+    for (int i = 0; i < array.length; i++) {
+      if (i > 0) sb.append(",");
+      sb.append(array[i]);
+    }
+    return sb;
+  }
+
+  private StringBuffer getReplacementTextFromCollection(Collection<?> collection)
+  {
+    if ((collection == null) || (collection.isEmpty())) {
+      return null;
+    }
+    StringBuffer sb = new StringBuffer();
+    int i = 0;
+    for (Iterator it = collection.iterator(); it.hasNext(); i++) {
+      if (i > 0) sb.append(",");
+      sb.append(it.next());
+    }
+    return sb;
+  }
+
+  public static void main(String[] args) {
+    HibernateStatementTextField textField = new HibernateStatementTextField();
+    textField.setValue("select * from and tt.LINK_ID in ($roleIds$), ($roleIds$)");
+    final HashMap<Object, Object> vo = new HashMap<>();
+    vo.put("roleIds", "ABC");
+    System.out.println(textField.getText(vo, new AtomicInteger(0)));
+    System.out.println(textField.getRawText(vo));
+  }
+}

sql-template-file/context-base.xml

@@ -0,0 +1,163 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<beans xmlns="http://www.springframework.org/schema/beans"
+	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+	xmlns:context="http://www.springframework.org/schema/context"
+	xmlns:aop="http://www.springframework.org/schema/aop"
+	xmlns:tx="http://www.springframework.org/schema/tx"
+	xsi:schemaLocation="http://www.springframework.org/schema/beans
+	                    http://www.springframework.org/schema/beans/spring-beans.xsd
+	                    http://www.springframework.org/schema/context
+	                    http://www.springframework.org/schema/context/spring-context.xsd
+	                    http://www.springframework.org/schema/aop
+	                    http://www.springframework.org/schema/aop/spring-aop.xsd
+	                    http://www.springframework.org/schema/tx
+	                    http://www.springframework.org/schema/tx/spring-tx.xsd"
+	   default-lazy-init="false">
+
+	<!-- DatabaseRecognizer: The key is database product name or its regular exp, and the value is to
+		 match the file name of spring/corebean/*-context-base.xml when the app startup -->
+	<bean id="databaseRecognizer" class="com.primeton.dgs.kernel.core.web.springframework.DatabaseRecognizer">
+		<property name="databaseNames">
+			<map>
+				<entry key="Microsoft SQL Server" value="SQLServer" />
+				<entry key="DB2/.+" value="DB2" />
+				<entry key="Oracle" value="Oracle" />
+				<entry key="MySQL" value="MySQL" />
+				<entry key="Teradata" value="Teradata" />
+				<entry key="Informix Dynamic Server" value="Informix" />
+				<entry key="GBase 8t V8.5 Database Server" value="Informix" />
+				<entry key="KingbaseES" value="Kingbase" />
+				<entry key="DM DBMS" value="DM" />
+				<entry key="PostgreSQL" value="PostgreSQL"/>
+			</map>
+		</property>
+	</bean>
+
+	<!-- ============== DAO DEFINITIONS: HIBERNATE IMPLEMENTATIONS ================= -->
+	<!-- 使用spring+hibernate处理oracle BLOB-->
+	<bean id="sessionFactory" class="com.primeton.dgs.kernel.core.dao.hibernate.HibernateLocalSessionFactoryBean">
+		<property name="databaseRecognizer" ref="databaseRecognizer" />
+		<property name="dataSource" ref="dataSource" />
+		<property name="mappingResources">
+			<list >
+				<value>META-INF/sqlxml/mysql/adapterSql.xml</value><!-- 加载hibernate模型配置文件,实际加载交由该具体类实现 -->
+			</list>
+		</property>
+		<property name="hibernateProperties">
+			<props>
+				<prop key="hibernate.dialect">org.hibernate.dialect.MySQL57Dialect</prop>
+				<prop key="hibernate.query.factory_class">org.hibernate.hql.internal.ast.ASTQueryTranslatorFactory</prop>
+				<prop key="hibernate.cache.provider_class">net.sf.ehcache.hibernate.EhCacheProvider</prop>
+				<prop key="hibernate.jdbc.batch_size">50</prop>
+				<!-- <prop key="hibernate.jdbc.fetch_size">100</prop> --><!--由于oracle驱动的bug导致memory leak-->
+				<prop key="hibernate.generate_statistics">true</prop>
+				<prop key="hibernate.show_sql">false</prop>
+				<prop key="hibernate.format_sql">false</prop>
+			</props>
+		</property>
+	</bean>
+
+	<bean id="hibernateDao" abstract="true">
+		<property name="sessionFactory" ref="sessionFactory" />
+		<property name="dataSource" ref="dataSource" />
+	</bean>
+
+	<bean id="seqService" class="com.primeton.dgs.kernel.common.app.base.bs.SequenceServiceImpl">
+	</bean>
+
+	<bean id="baseSequence" class="com.primeton.dgs.kernel.common.app.base.bs.SequenceServiceImpl"></bean>
+
+	<bean id="daservice"
+		  class="com.primeton.dgs.kernel.core.dao.impl.DAOServiceImpl"
+		  parent="hibernateDao" />
+
+	<bean id="jdbcTemplate" class="org.springframework.jdbc.core.JdbcTemplate">
+		<property name="dataSource" ref="dataSource" />
+	</bean>
+
+	<bean id="statementSessionFactory" class="com.primeton.dgs.kernel.core.dao.hibernate.HibernateStatementSessionFactoryBean"
+		  init-method="init" destroy-method="destroy">
+		<!-- 加载SQL资源文件,实际加载交由该具体类实现 -->
+		<property name="statementResources" value="" />
+	</bean>
+
+	<!--JPA 配置 start-->
+	<bean id="persistenceProvider" class="org.hibernate.jpa.HibernatePersistenceProvider"/>
+
+	<!-- JPA entityManagerFactory -->
+	<bean id="entityManagerFactory" class="com.primeton.dgs.kernel.core.dao.jpa.JpaContainerEntityManagerFactoryBean">
+		<!-- 指定数据源 -->
+		<property name="databaseRecognizer" ref="databaseRecognizer" />
+		<property name="dataSource" ref="dataSource"/>
+		<property name="persistenceUnitName" value="persistenceUnit"/>
+		<property name="persistenceProvider" ref="persistenceProvider"/>
+
+		<!-- 指定Entity实体类包路径 -->
+		<property name="packagesToScan">
+			<array>
+				<value>com.primeton.dgs.entity</value>
+				<value>org.gocom.coframe.model</value>
+			</array>
+		</property>
+		<!-- 指定Jpa持久化实现厂商类，根据 databaseRecognizer 配置，大多数知名数据库都能自动配置。国产数据库或者其他无法确认的数据需要设置
+		<property name="jpaVendorAdapter">
+			<bean class="org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter">
+				<property name="generateDdl" value="false"/>
+				<property name="database" value="MYSQL"/>
+				<property name="databasePlatform" value="org.hibernate.dialect.MySQL57Dialect"/>
+				<property name="showSql" value="false"/>
+			</bean>
+		</property>
+		 -->
+		<!-- 指定JPA属性-->
+		<property name="jpaPropertyMap">
+			<map>
+				<entry key="hibernate.show_sql" value="true" />
+				<entry key="hibernate.query.substitutions" value="true 1, false 0"/>
+				<entry key="hibernate.default_batch_fetch_size" value="16"/>
+				<entry key="hibernate.max_fetch_depth" value="2"/>
+				<entry key="hibernate.generate_statistics" value="false"/>
+				<entry key="hibernate.bytecode.use_reflection_optimizer" value="true"/>
+
+				<entry key="hibernate.cache.use_query_cache" value="false"/>
+				<entry key="hibernate.cache.use_second_level_cache" value="false"/>
+
+				<entry key="hibernate.enable_lazy_load_no_trans" value="true"/>
+				<!--数据库字段名称映射规则 acb_id自动转abcId-->
+				<entry key="hibernate.physical_naming_strategy" value="org.springframework.boot.orm.jpa.hibernate.SpringPhysicalNamingStrategy" />
+				<entry key="hibernate.implicit_naming_strategy" value="org.hibernate.boot.model.naming.ImplicitNamingStrategyLegacyHbmImpl"/>
+				<!--<entry key="hibernate.hbm2ddl.auto" value="validate"/>-->
+			</map>
+		</property>
+	</bean>
+
+	<context:annotation-config />
+	<context:component-scan base-package="org.gocom.coframe.**" />
+
+	<!-- 默认的事务管理器，支持 Hibernate。 JPA 见 TransactionManagerConfigurer -->
+	<bean id="defTransactionManager"
+		  class="org.springframework.orm.hibernate5.HibernateTransactionManager">
+		<property name="dataSource" ref="dataSource" />
+		<property name="sessionFactory" ref="sessionFactory" />
+	</bean>
+
+	<!-- Spring Context Helper -->
+	<bean id="SpringContextHelper" class="com.primeton.dgs.kernel.core.common.SpringContextHelper" />
+
+	<!-- Quartz针对不同的数据库有不同的配置在这里 -->
+	<bean id="schedulerDelegate" abstract="true">
+		<property name="quartzProperties">
+			<props>
+
+			</props>
+		</property>
+	</bean>
+
+	<!--
+	消息服务，用于事件发布，如采集完成，发布事件：获得事件后更新缓存、汇总分析立即执行等
+	add zhaopx at: 2019/11/11
+	 -->
+	<bean id="eventMessageService" class="com.primeton.dgs.kernel.core.message.MessageService"
+		  init-method="init" destroy-method="shutdown">
+	</bean>
+</beans>

sql-template-file/jpa/JpaContainerEntityManagerFactoryBean.java

@@ -0,0 +1,129 @@
+package com.primeton.dgs.kernel.core.dao.jpa;
+
+import com.primeton.dgs.kernel.core.util.MetadataWebContext;
+import com.primeton.dgs.kernel.core.web.springframework.DatabaseRecognizer;
+import org.apache.commons.lang.StringUtils;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.ApplicationContextException;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
+import org.springframework.core.io.support.ResourcePatternResolver;
+import org.springframework.orm.jpa.JpaVendorAdapter;
+import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean;
+import org.springframework.orm.jpa.vendor.Database;
+import org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter;
+
+import javax.persistence.PersistenceException;
+import javax.sql.DataSource;
+import java.io.InputStream;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.SQLException;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ *
+ * JPA 配置
+ *
+ * <pre>
+ *
+ * Created by zhaopx.
+ * User: zhaopx
+ * Date: 2020/9/1
+ * Time: 17:44
+ *
+ * </pre>
+ *
+ * @author zhaopx
+ */
+public class JpaContainerEntityManagerFactoryBean
+        extends LocalContainerEntityManagerFactoryBean
+        implements InitializingBean {
+
+
+    private DatabaseRecognizer databaseRecognizer;
+
+    @Override
+    public void afterPropertiesSet() throws PersistenceException {
+        // 扫描配置文件
+        // 首先根据数据库选择配置文件
+        String databaseName = MetadataWebContext.getInstance().getDatabaseName();
+        Properties props = new Properties();
+        String configFile = "classpath*:".concat("/spring/hibernate_").concat(databaseName.toLowerCase()).concat("_local.properties");
+        try {
+            ResourcePatternResolver resolver = new PathMatchingResourcePatternResolver();
+            Resource[] resources = resolver.getResources(configFile);
+            for (Resource resource : resources) {
+                try(InputStream inputStream = resource.getInputStream()) {
+                    Properties tmp = new Properties();
+                    tmp.load(inputStream);
+                    props.putAll(tmp);
+                }
+            }
+        } catch (Exception e) {
+            logger.error("无法读取配置文件: " + configFile, e);
+        }
+
+        if(!props.isEmpty()) {
+            for (Map.Entry<Object, Object> entry : props.entrySet()) {
+                getJpaPropertyMap().put((String) entry.getKey(), entry.getValue());
+            }
+        }
+
+
+        Database database;
+        try {
+            database = Database.valueOf(databaseName.toUpperCase());
+        } catch (Exception e) {
+            logger.error("unkown db " + databaseName + " use Database.DEFAULT, please set spring bean databaseRecognizer key-value, to match " + Arrays.asList(Database.values()));
+            database = Database.DEFAULT;
+        }
+        final HibernateJpaVendorAdapter jpaVendorAdapter = new HibernateJpaVendorAdapter();
+        jpaVendorAdapter.setDatabasePlatform((String) getJpaPropertyMap().get("hibernate.dialect"));
+        jpaVendorAdapter.setDatabase(database);
+        jpaVendorAdapter.setGenerateDdl(false);
+        jpaVendorAdapter.setShowSql(true);
+        setJpaVendorAdapter(jpaVendorAdapter);
+        super.afterPropertiesSet();
+    }
+
+
+    @Override
+    public void setDataSource(DataSource dataSource) {
+        super.setDataSource(dataSource);
+        // DB Type
+        if(StringUtils.isBlank(MetadataWebContext.getInstance().getDatabaseName())) {
+            try (Connection conn = dataSource.getConnection()) {
+                DatabaseMetaData meta = conn.getMetaData();
+                String databaseName = meta.getDatabaseProductName();
+                String productVersion = meta.getDatabaseProductVersion();
+                if (logger.isInfoEnabled()) {
+                    logger.info("Current Database is " + databaseName + ", Version is " + productVersion);
+                }
+                databaseName = recognizeDatabaseName(databaseName, productVersion);
+                // 将数据库名缓存起来
+                MetadataWebContext.getInstance().setDatabaseName(databaseName);
+            } catch (SQLException e) {
+                String s = "Cannot determine the database brand name for loading ApplicationContext";
+                logger.error(s, e);
+                throw new ApplicationContextException(s, e);
+            }
+        }
+    }
+
+    /**
+     * 识别数据库名称，已经配置在spring/corebean/context-base.xml文件中
+     * @param databaseName 从数据库连接获取的数据库名称
+     * @param productVersion 数据库版本
+     * @return 转换成系统需要的名称
+     */
+    private String recognizeDatabaseName(String databaseName, String productVersion) {
+        return databaseRecognizer.getDatabaseName(databaseName);
+    }
+
+    public void setDatabaseRecognizer(DatabaseRecognizer databaseRecognizer) {
+        this.databaseRecognizer = databaseRecognizer;
+    }
+}