脚本任务实现:Shell和Python
This commit is contained in:
parent
7a129468a4
commit
935cc1276f
|
@ -193,17 +193,15 @@ public class DemoGlueJobHandler extends IJobHandler {
|
||||||
<textarea class="glueSource_shell" style="display:none;" >
|
<textarea class="glueSource_shell" style="display:none;" >
|
||||||
#!/bin/bash
|
#!/bin/bash
|
||||||
|
|
||||||
echo hello shell
|
echo "xxl-job: hello shell"
|
||||||
|
|
||||||
for x in 1 2 3 4
|
for item in 1 2 3
|
||||||
do
|
do
|
||||||
echo number=$x
|
echo "shell : $item"
|
||||||
sleep 1s
|
sleep 1s
|
||||||
done
|
done
|
||||||
|
|
||||||
echo1 111
|
echo "Good bye!"
|
||||||
printf 666
|
|
||||||
echo2 222
|
|
||||||
</textarea>
|
</textarea>
|
||||||
<textarea class="glueSource_python" style="display:none;" >
|
<textarea class="glueSource_python" style="display:none;" >
|
||||||
#!/usr/bin/python
|
#!/usr/bin/python
|
||||||
|
@ -214,11 +212,11 @@ import time
|
||||||
|
|
||||||
logging.basicConfig(level=logging.DEBUG)
|
logging.basicConfig(level=logging.DEBUG)
|
||||||
|
|
||||||
logging.info('hello python')
|
logging.info('xxl-job: hello python')
|
||||||
|
|
||||||
for num in range(0, 3):
|
for num in range(1, 3):
|
||||||
time.sleep(1)
|
time.sleep(1)
|
||||||
logging.info('当前序号 :' + str(num) )
|
logging.info('python :' + str(num) )
|
||||||
|
|
||||||
logging.info('Good bye!')
|
logging.info('Good bye!')
|
||||||
</textarea>
|
</textarea>
|
||||||
|
|
|
@ -9,15 +9,12 @@ import com.xxl.job.core.glue.GlueFactory;
|
||||||
import com.xxl.job.core.glue.GlueTypeEnum;
|
import com.xxl.job.core.glue.GlueTypeEnum;
|
||||||
import com.xxl.job.core.handler.IJobHandler;
|
import com.xxl.job.core.handler.IJobHandler;
|
||||||
import com.xxl.job.core.handler.impl.GlueJobHandler;
|
import com.xxl.job.core.handler.impl.GlueJobHandler;
|
||||||
|
import com.xxl.job.core.handler.impl.ScriptJobHandler;
|
||||||
import com.xxl.job.core.log.XxlJobFileAppender;
|
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||||
import com.xxl.job.core.thread.JobThread;
|
import com.xxl.job.core.thread.JobThread;
|
||||||
import com.xxl.job.core.util.ScriptUtil;
|
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
import java.io.File;
|
|
||||||
import java.io.FileOutputStream;
|
|
||||||
import java.io.IOException;
|
|
||||||
import java.util.Date;
|
import java.util.Date;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -85,11 +82,11 @@ public class ExecutorBizImpl implements ExecutorBiz {
|
||||||
|
|
||||||
} else if (GlueTypeEnum.GLUE_GROOVY==GlueTypeEnum.match(triggerParam.getGlueType())) {
|
} else if (GlueTypeEnum.GLUE_GROOVY==GlueTypeEnum.match(triggerParam.getGlueType())) {
|
||||||
|
|
||||||
// valid exists job thread:change handler or glue timeout, need kill old thread
|
// valid exists job thread:change handler or gluesource updated, need kill old thread
|
||||||
if (jobThread != null &&
|
if (jobThread != null &&
|
||||||
!(jobThread.getHandler() instanceof GlueJobHandler
|
!(jobThread.getHandler() instanceof GlueJobHandler
|
||||||
&& ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
|
&& ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
|
||||||
// change glue model or glue timeout, kill old job thread
|
// change glue model or gluesource updated, kill old job thread
|
||||||
jobThread.toStop("更换任务模式或JobHandler,终止旧任务线程");
|
jobThread.toStop("更换任务模式或JobHandler,终止旧任务线程");
|
||||||
jobThread.interrupt();
|
jobThread.interrupt();
|
||||||
XxlJobExecutor.removeJobThread(triggerParam.getJobId());
|
XxlJobExecutor.removeJobThread(triggerParam.getJobId());
|
||||||
|
@ -107,61 +104,25 @@ public class ExecutorBizImpl implements ExecutorBiz {
|
||||||
}
|
}
|
||||||
jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), new GlueJobHandler(jobHandler, triggerParam.getGlueUpdatetime()));
|
jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), new GlueJobHandler(jobHandler, triggerParam.getGlueUpdatetime()));
|
||||||
}
|
}
|
||||||
} else if (GlueTypeEnum.GLUE_SHELL==GlueTypeEnum.match(triggerParam.getGlueType())) {
|
} else if (GlueTypeEnum.GLUE_SHELL==GlueTypeEnum.match(triggerParam.getGlueType())
|
||||||
|
|| GlueTypeEnum.GLUE_PYTHON==GlueTypeEnum.match(triggerParam.getGlueType()) ) {
|
||||||
|
|
||||||
// make path
|
// valid exists job thread:change script or gluesource updated, need kill old thread
|
||||||
String scriptPath = XxlJobFileAppender.filePath + "gluesource/";
|
if (jobThread != null &&
|
||||||
String scriptFileName = triggerParam.getJobId() + "_" + triggerParam.getGlueUpdatetime() + ".sh";
|
!(jobThread.getHandler() instanceof ScriptJobHandler
|
||||||
|
&& ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
|
||||||
// valid file
|
// change glue model or gluesource updated, kill old job thread
|
||||||
File scriptFile = new File(scriptPath, scriptFileName);
|
jobThread.toStop("更换任务模式或JobHandler,终止旧任务线程");
|
||||||
if (!scriptFile.exists()) {
|
jobThread.interrupt();
|
||||||
// valid glue source
|
XxlJobExecutor.removeJobThread(triggerParam.getJobId());
|
||||||
if (triggerParam.getGlueSource()==null) {
|
jobThread = null;
|
||||||
return new ReturnT<String>(ReturnT.FAIL_CODE, "glueSource is null.");
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// .../gluesource/
|
// make thread: new or exists invalid
|
||||||
File scriptPathDir = new File(scriptPath);
|
if (jobThread == null) {
|
||||||
if (!scriptPathDir.exists()) {
|
ScriptJobHandler scriptJobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType()));
|
||||||
scriptPathDir.mkdirs();
|
jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), scriptJobHandler);
|
||||||
}
|
}
|
||||||
|
|
||||||
// .../gluesource/666-156465656.sh
|
|
||||||
scriptFile = new File(scriptPath, scriptFileName);
|
|
||||||
FileOutputStream fos = null;
|
|
||||||
try {
|
|
||||||
scriptFile.createNewFile();
|
|
||||||
|
|
||||||
fos = new FileOutputStream(scriptFile, true);
|
|
||||||
fos.write(triggerParam.getGlueSource().getBytes("utf-8"));
|
|
||||||
fos.flush();
|
|
||||||
} catch (IOException e) {
|
|
||||||
logger.error(e.getMessage(), e);
|
|
||||||
return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
|
|
||||||
} finally {
|
|
||||||
if (fos != null) {
|
|
||||||
try {
|
|
||||||
fos.close();
|
|
||||||
} catch (IOException e) {
|
|
||||||
logger.error(e.getMessage(), e);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
// log File
|
|
||||||
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTim()), triggerParam.getLogId());
|
|
||||||
|
|
||||||
// run script
|
|
||||||
ScriptUtil.execToFile("python", scriptFile.getName(), (XxlJobFileAppender.filePath + logFileName) );
|
|
||||||
|
|
||||||
return ReturnT.FAIL;
|
|
||||||
} else if (GlueTypeEnum.GLUE_PYTHON==GlueTypeEnum.match(triggerParam.getGlueType())) {
|
|
||||||
String scriptFilePath = XxlJobFileAppender.filePath + "gluesource/" + triggerParam.getJobId() + "_" + triggerParam.getGlueUpdatetime() + ".py";
|
|
||||||
|
|
||||||
|
|
||||||
} else {
|
} else {
|
||||||
return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
|
return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
|
||||||
}
|
}
|
||||||
|
|
|
@ -0,0 +1,55 @@
|
||||||
|
package com.xxl.job.core.handler.impl;
|
||||||
|
|
||||||
|
import com.xxl.job.core.biz.model.ReturnT;
|
||||||
|
import com.xxl.job.core.glue.GlueTypeEnum;
|
||||||
|
import com.xxl.job.core.handler.IJobHandler;
|
||||||
|
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||||
|
import com.xxl.job.core.util.ScriptUtil;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Created by xuxueli on 17/4/27.
|
||||||
|
*/
|
||||||
|
public class ScriptJobHandler extends IJobHandler {
|
||||||
|
|
||||||
|
private int jobId;
|
||||||
|
private long glueUpdatetime;
|
||||||
|
private String gluesource;
|
||||||
|
private GlueTypeEnum glueType;
|
||||||
|
|
||||||
|
public ScriptJobHandler(int jobId, long glueUpdatetime, String gluesource, GlueTypeEnum glueType){
|
||||||
|
this.jobId = jobId;
|
||||||
|
this.glueUpdatetime = glueUpdatetime;
|
||||||
|
this.gluesource = gluesource;
|
||||||
|
this.glueType = glueType;
|
||||||
|
}
|
||||||
|
|
||||||
|
public long getGlueUpdatetime() {
|
||||||
|
return glueUpdatetime;
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public ReturnT<String> execute(String... params) throws Exception {
|
||||||
|
|
||||||
|
// cmd + script-file-name
|
||||||
|
String cmd = "bash";
|
||||||
|
String scriptFileName = null;
|
||||||
|
if (GlueTypeEnum.GLUE_SHELL == glueType) {
|
||||||
|
cmd = "bash";
|
||||||
|
scriptFileName = XxlJobFileAppender.filePath.concat("gluesource/").concat(String.valueOf(jobId)).concat("_").concat(String.valueOf(glueUpdatetime)).concat(".sh");
|
||||||
|
} else if (GlueTypeEnum.GLUE_PYTHON == glueType) {
|
||||||
|
cmd = "python";
|
||||||
|
scriptFileName = XxlJobFileAppender.filePath.concat("gluesource/").concat(String.valueOf(jobId)).concat("_").concat(String.valueOf(glueUpdatetime)).concat(".py");
|
||||||
|
}
|
||||||
|
|
||||||
|
// make script file
|
||||||
|
ScriptUtil.markScriptFile(scriptFileName, gluesource);
|
||||||
|
|
||||||
|
// log file
|
||||||
|
String logFileName = XxlJobFileAppender.filePath.concat(XxlJobFileAppender.contextHolder.get());
|
||||||
|
|
||||||
|
// invoke
|
||||||
|
ScriptUtil.execToFile(cmd, scriptFileName, logFileName);
|
||||||
|
return ReturnT.SUCCESS;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
|
@ -1,49 +1,57 @@
|
||||||
package com.xxl.job.core.util;
|
package com.xxl.job.core.util;
|
||||||
|
|
||||||
|
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||||
import org.apache.commons.exec.CommandLine;
|
import org.apache.commons.exec.CommandLine;
|
||||||
import org.apache.commons.exec.DefaultExecutor;
|
import org.apache.commons.exec.DefaultExecutor;
|
||||||
import org.apache.commons.exec.PumpStreamHandler;
|
import org.apache.commons.exec.PumpStreamHandler;
|
||||||
|
|
||||||
import java.io.File;
|
import java.io.File;
|
||||||
import java.io.FileOutputStream;
|
import java.io.FileOutputStream;
|
||||||
|
import java.io.IOException;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* 1、内嵌编译器如"PythonInterpreter"无法引用扩展包,因此推荐使用java调用控制台进程方式"Runtime.getRuntime().exec()"来运行脚本(shell或python);
|
* 1、内嵌编译器如"PythonInterpreter"无法引用扩展包,因此推荐使用java调用控制台进程方式"Runtime.getRuntime().exec()"来运行脚本(shell或python);
|
||||||
* 2、因为通过java调用控制台进程方式实现,需要保证目标机器PATH路径正确配置对应编译器;
|
* 2、因为通过java调用控制台进程方式实现,需要保证目标机器PATH路径正确配置对应编译器;
|
||||||
* 3、暂时脚本执行日志只能在脚本执行结束后一次性获取,无法保证实时性;因此为确保日志实时性,可改为将脚本打印的日志存储在指定的日志文件上;
|
* 3、暂时脚本执行日志只能在脚本执行结束后一次性获取,无法保证实时性;因此为确保日志实时性,可改为将脚本打印的日志存储在指定的日志文件上;
|
||||||
*
|
* 4、python 异常输出优先级高于标准输出,体现在Log文件中,因此推荐通过logging方式打日志保持和异常信息一致;否则用prinf日志顺序会错乱
|
||||||
* 知识点:
|
|
||||||
* 1、日志输出到日志文件:[>>logfile 2>&1]:将错误输出2以及标准输出1都一起以附加写方式导入logfile文件
|
|
||||||
* 2、python 异常输出优先级高于标准输出,体现在Log文件中,因此推荐通过logging方式打日志保持和异常信息一致;否则用prinf日志顺序会错乱
|
|
||||||
*
|
*
|
||||||
* Created by xuxueli on 17/2/25.
|
* Created by xuxueli on 17/2/25.
|
||||||
*/
|
*/
|
||||||
public class ScriptUtil {
|
public class ScriptUtil {
|
||||||
|
|
||||||
private static String pyCmd = "python";
|
/**
|
||||||
private static String shllCmd = "bash";
|
* make script file
|
||||||
private static String pyFile = "/Users/xuxueli/workspaces/idea-git-workspace/github/xxl-incubator/xxl-util/src/main/resources/script/pytest.py";
|
*
|
||||||
private static String shellFile = "/Users/xuxueli/workspaces/idea-git-workspace/github/xxl-incubator/xxl-util/src/main/resources/script/shelltest.sh";
|
* @param scriptFileName
|
||||||
private static String pyLogFile = "/Users/xuxueli/Downloads/tmp/pylog.log";
|
* @param content
|
||||||
private static String shLogFile = "/Users/xuxueli/Downloads/tmp/shlog.log";
|
* @throws IOException
|
||||||
|
*/
|
||||||
public static void main(String[] args) {
|
public static void markScriptFile(String scriptFileName, String content) throws IOException {
|
||||||
|
// filePath/
|
||||||
String command = pyCmd;
|
File filePathDir = new File(XxlJobFileAppender.filePath);
|
||||||
String filename = pyFile;
|
if (!filePathDir.exists()) {
|
||||||
String logFile = pyLogFile;
|
filePathDir.mkdirs();
|
||||||
if (false) {
|
|
||||||
command = shllCmd;
|
|
||||||
filename = shellFile;
|
|
||||||
logFile = shLogFile;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
execToFile(command, filename, logFile);
|
// filePath/gluesource/
|
||||||
|
File filePathSourceDir = new File(filePathDir, "gluesource");
|
||||||
|
if (!filePathSourceDir.exists()) {
|
||||||
|
filePathSourceDir.mkdirs();
|
||||||
}
|
}
|
||||||
|
|
||||||
public static File markScriptFile(){
|
// make file, filePath/gluesource/666-123456789.py
|
||||||
return null;
|
FileOutputStream fileOutputStream = null;
|
||||||
|
try {
|
||||||
|
fileOutputStream = new FileOutputStream(scriptFileName);
|
||||||
|
fileOutputStream.write(content.getBytes("UTF-8"));
|
||||||
|
fileOutputStream.close();
|
||||||
|
} catch (Exception e) {
|
||||||
|
throw e;
|
||||||
|
}finally{
|
||||||
|
if(fileOutputStream != null){
|
||||||
|
fileOutputStream.close();
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -58,12 +66,11 @@ public class ScriptUtil {
|
||||||
* @param scriptFile
|
* @param scriptFile
|
||||||
* @param logFile
|
* @param logFile
|
||||||
*/
|
*/
|
||||||
public static void execToFile(String command, String scriptFile, String logFile){
|
public static void execToFile(String command, String scriptFile, String logFile) throws IOException {
|
||||||
try {
|
|
||||||
// 标准输出:print (null if watchdog timeout)
|
// 标准输出:print (null if watchdog timeout)
|
||||||
// 错误输出:logging + 异常 (still exists if watchdog timeout)
|
// 错误输出:logging + 异常 (still exists if watchdog timeout)
|
||||||
// 标准输出
|
// 标准输入
|
||||||
FileOutputStream fileOutputStream = new FileOutputStream(logFile);
|
FileOutputStream fileOutputStream = new FileOutputStream(logFile, true);
|
||||||
PumpStreamHandler streamHandler = new PumpStreamHandler(fileOutputStream, fileOutputStream, null);
|
PumpStreamHandler streamHandler = new PumpStreamHandler(fileOutputStream, fileOutputStream, null);
|
||||||
|
|
||||||
// command
|
// command
|
||||||
|
@ -75,12 +82,6 @@ public class ScriptUtil {
|
||||||
exec.setExitValues(null);
|
exec.setExitValues(null);
|
||||||
exec.setStreamHandler(streamHandler);
|
exec.setStreamHandler(streamHandler);
|
||||||
int exitValue = exec.execute(commandline);
|
int exitValue = exec.execute(commandline);
|
||||||
} catch (Exception e) {
|
|
||||||
e.printStackTrace();
|
|
||||||
}
|
|
||||||
/*Process process = Runtime.getRuntime().exec(cmdarray);
|
|
||||||
IOUtils.copy(process.getInputStream(), out);
|
|
||||||
IOUtils.copy(process.getErrorStream(), out);*/
|
|
||||||
}
|
}
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue