DataX-Web源码分析之ExecutorJobHandler

方法声明满足xxl-job执行代码执行方法的格式
 @Override
 public ReturnT execute(TriggerParam trigger) {

        int exitValue = -1;
        Thread errThread = null;
        String tmpFilePath;
        LogStatistics logStatistics = null;
        //Generate JSON temporary file
       // 获取JSON文件内容，并写入临时文件夹
        tmpFilePath = generateTemJsonFile(trigger.getJobJson());

        try {
           //拼装一个datax执行的命令，

格式为 python {YOUR_DATAX_HOME}/bin/datax.py    {YOUR_DATAX_HOME}/job/job.json
         String[] cmdarrayFinal = buildDataXExecutorCmd(trigger, tmpFilePath,dataXPyPath);
          //执行相关的命令
            final Process process = Runtime.getRuntime().exec(cmdarrayFinal);
            String prcsId = ProcessUtil.getProcessId(process);
            JobLogger.log("------------------DataX process id: " + prcsId);
            jobTmpFiles.put(prcsId, tmpFilePath);
            //update datax process id
            HandleProcessCallbackParam prcs = new HandleProcessCallbackParam(trigger.getLogId(), trigger.getLogDateTime(), prcsId);
            ProcessCallbackThread.pushCallBack(prcs);
            // log-thread
            Thread futureThread = null;
            FutureTask futureTask = new FutureTask<>(() -> analysisStatisticsLog(new BufferedInputStream(process.getInputStream())));
            futureThread = new Thread(futureTask);
            futureThread.start();

            errThread = new Thread(() -> {
                try {
                    analysisStatisticsLog(new BufferedInputStream(process.getErrorStream()));
                } catch (IOException e) {
                    JobLogger.log(e);
                }
            });

            logStatistics = futureTask.get();
            errThread.start();
            // process-wait
            exitValue = process.waitFor();      // exit code: 0=success, 1=error
            // log-thread join
            errThread.join();
        } catch (Exception e) {
            JobLogger.log(e);
        } finally {
            if (errThread != null && errThread.isAlive()) {
                errThread.interrupt();
            }
            //  删除临时文件
            if (FileUtil.exist(tmpFilePath)) {
                FileUtil.del(new File(tmpFilePath));
            }
        }
        if (exitValue == 0) {
            return new ReturnT<>(200, logStatistics.toString());
        } else {
            return new ReturnT<>(IJobHandler.FAIL.getCode(), "command exit value(" + exitValue + ") is failed");
        }
    }

二、JOSN文件内容写入临时文件分析

private String generateTemJsonFile(String jobJson) {
    String tmpFilePath;
   //获取datax的根目录
    String dataXHomePath = SystemUtils.getDataXHomePath();
    if (StringUtils.isNotEmpty(dataXHomePath)) {
       //拼装一个jsons路径：public static final String DEFAULT_JSON = "jsons";
        jsonPath = dataXHomePath + DEFAULT_JSON;
    }
    //文件路径不存在，则创建一个路径
    if (!FileUtil.exist(jsonPath)) {
        FileUtil.mkdir(jsonPath);
    }
    tmpFilePath = jsonPath + "jobTmp-" + IdUtil.simpleUUID() + ".conf";
    // 根据json写入到临时本地文件
    try (PrintWriter writer = new PrintWriter(tmpFilePath, "UTF-8")) {
        writer.println(jobJson);
    } catch (FileNotFoundException | UnsupportedEncodingException e) {
        JobLogger.log("JSON 临时文件写入异常：" + e.getMessage());
    }
    return tmpFilePath;
}

三、封装最终命令分析

public static String[] buildDataXExecutorCmd(TriggerParam tgParam, String tmpFilePath, String dataXPyPath) {
        // command process
        //"--loglevel=debug"
       //封装一个如下格式的命令：格式为 python {YOUR_DATAX_HOME}/bin/datax.py    {YOUR_DATAX_HOME}/job/job.json
        List cmdArr = new ArrayList<>();
      //python充当先锋
        cmdArr.add("python");
        String dataXHomePath = SystemUtils.getDataXHomePath();
        if (StringUtils.isNotEmpty(dataXHomePath)) {
           // 拼装datax.py路径，常量是public static final String DEFAULT_DATAX_PY = "datax.py";
            dataXPyPath = dataXHomePath.contains("bin") ? dataXHomePath + DEFAULT_DATAX_PY : dataXHomePath + "bin" + File.separator + DEFAULT_DATAX_PY;
        }
        cmdArr.add(dataXPyPath);
       //配置一下其他参数
        String doc = buildDataXParam(tgParam);
        if (StringUtils.isNotBlank(doc)) {
            cmdArr.add(doc.replaceAll(SPLIT_SPACE, TRANSFORM_SPLIT_SPACE));
        }
        //拼装JSON文件路径
        cmdArr.add(tmpFilePath);
        return cmdArr.toArray(new String[cmdArr.size()]);
    }