调度中心与执行器通讯规范为双向restful,方便跨语言,以及第三方执行器实现;通讯组件xxl-rpc方案调整为Jetty+Gson方案;

master
xuxueli 5 years ago
parent c925bd1685
commit 895ad80e2a
  1. 6
      xxl-job-admin/src/main/java/com/xxl/job/admin/core/trigger/XxlJobTrigger.java
  2. 6
      xxl-job-admin/src/main/resources/static/js/jobinfo.index.1.js
  3. 4
      xxl-job-core/src/main/java/com/xxl/job/core/server/EmbedServer.java
  4. 10
      xxl-job-core/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java
  5. 8
      xxl-job-core/src/main/java/com/xxl/job/core/util/FileUtil.java
  6. 73
      xxl-job-core/src/main/java/com/xxl/job/core/util/JdkSerializeTool.java
  7. 24
      xxl-job-core/src/main/java/com/xxl/job/core/util/ThrowableUtil.java

@ -1,18 +1,18 @@
package com.xxl.job.admin.core.trigger;
import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
import com.xxl.job.admin.core.model.XxlJobGroup;
import com.xxl.job.admin.core.model.XxlJobInfo;
import com.xxl.job.admin.core.model.XxlJobLog;
import com.xxl.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
import com.xxl.job.core.util.IpUtil;
import io.netty.util.internal.ThrowableUtil;
import com.xxl.job.core.util.ThrowableUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -211,7 +211,7 @@ public class XxlJobTrigger {
runResult = executorBiz.run(triggerParam);
} catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e);
runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ThrowableUtil.stackTraceToString(e));
runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ThrowableUtil.toString(e));
}
StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + ":");

@ -298,13 +298,13 @@ $(function() {
dataType : "json",
success : function(data){
var html = '<center>';
var html = '<div>';
if (data.code == 200 && data.content.registryList) {
for (var index in data.content.registryList) {
html += '<span class="badge bg-green" >' + data.content.registryList[index] + '</span><br>';
html += (parseInt(index)+1) + '. <span class="badge bg-green" >' + data.content.registryList[index] + '</span><br>';
}
}
html += '</center>';
html += '</div>';
layer.open({
title: I18n.jobinfo_opt_registryinfo ,

@ -7,6 +7,7 @@ import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
import com.xxl.job.core.thread.ExecutorRegistryThread;
import com.xxl.job.core.util.GsonTool;
import com.xxl.job.core.util.ThrowableUtil;
import com.xxl.job.core.util.XxlJobRemotingUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
@ -18,7 +19,6 @@ import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import io.netty.util.internal.ThrowableUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -185,7 +185,7 @@ public class EmbedServer {
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, "request error:" + ThrowableUtil.stackTraceToString(e));
return new ReturnT<String>(ReturnT.FAIL_CODE, "request error:" + ThrowableUtil.toString(e));
}
}

@ -8,7 +8,7 @@ import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.FileUtil;
import com.xxl.job.core.util.GsonTool;
import com.xxl.job.core.util.JdkSerializeTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -205,7 +205,7 @@ public class TriggerCallbackThread {
}
// append file
String callbackParamList_Str = GsonTool.toJson(callbackParamList);
byte[] callbackParamList_bytes = JdkSerializeTool.serialize(callbackParamList);
File callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis())));
if (callbackLogFile.exists()) {
@ -216,7 +216,7 @@ public class TriggerCallbackThread {
}
}
}
FileUtil.writeFileContent(callbackLogFile, callbackParamList_Str);
FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes);
}
private void retryFailCallbackFile(){
@ -235,8 +235,8 @@ public class TriggerCallbackThread {
// load and clear file, retry
for (File callbaclLogFile: callbackLogPath.listFiles()) {
String callbackParamList_str = FileUtil.readFileContent(callbaclLogFile);
List<HandleCallbackParam> callbackParamList = GsonTool.fromJsonList(callbackParamList_str, HandleCallbackParam.class);
byte[] callbackParamList_bytes = FileUtil.readFileContent(callbaclLogFile);
List<HandleCallbackParam> callbackParamList = (List<HandleCallbackParam>) JdkSerializeTool.deserialize(callbackParamList_bytes, List.class);
callbaclLogFile.delete();
doCallback(callbackParamList);

@ -48,7 +48,7 @@ public class FileUtil {
}
public static void writeFileContent(File file, String data) {
public static void writeFileContent(File file, byte[] data) {
// file
if (!file.exists()) {
@ -59,7 +59,7 @@ public class FileUtil {
FileOutputStream fos = null;
try {
fos = new FileOutputStream(file);
fos.write(data.getBytes("utf-8"));
fos.write(data);
fos.flush();
} catch (Exception e) {
logger.error(e.getMessage(), e);
@ -75,7 +75,7 @@ public class FileUtil {
}
public static String readFileContent(File file) {
public static byte[] readFileContent(File file) {
Long filelength = file.length();
byte[] filecontent = new byte[filelength.intValue()];
@ -85,7 +85,7 @@ public class FileUtil {
in.read(filecontent);
in.close();
return new String(filecontent, "utf-8");
return filecontent;
} catch (Exception e) {
logger.error(e.getMessage(), e);
return null;

@ -0,0 +1,73 @@
package com.xxl.job.core.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
/**
* @author xuxueli 2020-04-12 0:14:00
*/
public class JdkSerializeTool {
private static Logger logger = LoggerFactory.getLogger(JdkSerializeTool.class);
// ------------------------ serialize and unserialize ------------------------
/**
* 将对象-->byte[] (由于jedis中不支持直接存储object所以转换成byte[]存入)
*
* @param object
* @return
*/
public static byte[] serialize(Object object) {
ObjectOutputStream oos = null;
ByteArrayOutputStream baos = null;
try {
// 序列化
baos = new ByteArrayOutputStream();
oos = new ObjectOutputStream(baos);
oos.writeObject(object);
byte[] bytes = baos.toByteArray();
return bytes;
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
try {
oos.close();
baos.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
return null;
}
/**
* 将byte[] -->Object
*
* @param bytes
* @return
*/
public static <T> Object deserialize(byte[] bytes, Class<T> clazz) {
ByteArrayInputStream bais = null;
try {
// 反序列化
bais = new ByteArrayInputStream(bytes);
ObjectInputStream ois = new ObjectInputStream(bais);
return ois.readObject();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
try {
bais.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
return null;
}
}

@ -0,0 +1,24 @@
package com.xxl.job.core.util;
import java.io.PrintWriter;
import java.io.StringWriter;
/**
* @author xuxueli 2018-10-20 20:07:26
*/
public class ThrowableUtil {
/**
* parse error to string
*
* @param e
* @return
*/
public static String toString(Throwable e) {
StringWriter stringWriter = new StringWriter();
e.printStackTrace(new PrintWriter(stringWriter));
String errorMsg = stringWriter.toString();
return errorMsg;
}
}
Loading…
Cancel
Save