执行器回调日志落盘方案复用RPC序列化方案,并移除Jackson依赖;

This commit is contained in:
xuxueli 2019-02-18 19:12:08 +08:00
parent 2f1dda6946
commit 46779d7e6b
9 changed files with 112 additions and 50 deletions

View File

@ -1419,9 +1419,10 @@ Tips: 历史版本(V1.3.x)目前已经Release至稳定版本, 进入维护阶段
- 5、精简项目取消第三方依赖如 commons-collections4 ; - 5、精简项目取消第三方依赖如 commons-collections4 ;
- 6、底层Log调优应用正常终止取消异常栈信息打印 - 6、底层Log调优应用正常终止取消异常栈信息打印
- 7、交互优化尽量避免新开页面窗口仅WebIDE支持新开页并提供窗口快速关闭按钮 - 7、交互优化尽量避免新开页面窗口仅WebIDE支持新开页并提供窗口快速关闭按钮
- 8、[测试中]底层通讯方案优化升级较新版本xxl-rpc由"JETTY"方案调整为"NETTY_HTTP"方案执行器内嵌netty-http-server提供服务调度中心复用容器端口提供服务 - 8、底层通讯方案优化升级较新版本xxl-rpc由"JETTY"方案调整为"NETTY_HTTP"方案执行器内嵌netty-http-server提供服务调度中心复用容器端口提供服务
- 9、任务暂停、删除优化避免quartz delete不完整导致任务脏数据 - 9、任务暂停、删除优化避免quartz delete不完整导致任务脏数据
- 10、任务回调、心跳注册成功日志优化非核心常规日志调整为debug级别降低冗余日志输出 - 10、任务回调、心跳注册成功日志优化非核心常规日志调整为debug级别降低冗余日志输出
- 11、执行器回调日志落盘方案复用RPC序列化方案并移除Jackson依赖
- [迭代中]注册中心优化实时性注册发现心跳注册间隔10srefresh失败则首次注册并立即更新注册信息心跳类似30s过期销毁 - [迭代中]注册中心优化实时性注册发现心跳注册间隔10srefresh失败则首次注册并立即更新注册信息心跳类似30s过期销毁
- [迭代中]脚本任务,支持数据参数,新版本仅支持单参数不支持需要兼容; - [迭代中]脚本任务,支持数据参数,新版本仅支持单参数不支持需要兼容;
- [迭代中]提供执行器Docker镜像 - [迭代中]提供执行器Docker镜像

View File

@ -41,8 +41,6 @@
<groovy.version>2.5.5</groovy.version> <groovy.version>2.5.5</groovy.version>
<quartz.version>2.3.0</quartz.version> <quartz.version>2.3.0</quartz.version>
<jackson.version>2.9.8</jackson.version>
</properties> </properties>
<build> <build>

View File

@ -1,7 +1,7 @@
package com.xxl.job.admin.controller.resolver; package com.xxl.job.admin.controller.resolver;
import com.xxl.job.core.biz.model.ReturnT; import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.util.JacksonUtil; import com.xxl.job.admin.core.util.JacksonUtil;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component; import org.springframework.stereotype.Component;

View File

@ -1,7 +1,6 @@
package com.xxl.job.admin.core.util; package com.xxl.job.admin.core.util;
import com.xxl.job.admin.core.conf.XxlJobAdminConfig; import com.xxl.job.admin.core.conf.XxlJobAdminConfig;
import com.xxl.job.core.util.JacksonUtil;
import org.apache.commons.lang3.StringUtils; import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;

View File

@ -1,4 +1,4 @@
package com.xxl.job.core.util; package com.xxl.job.admin.core.util;
import com.fasterxml.jackson.core.JsonGenerationException; import com.fasterxml.jackson.core.JsonGenerationException;
import com.fasterxml.jackson.core.JsonParseException; import com.fasterxml.jackson.core.JsonParseException;

View File

@ -36,13 +36,6 @@
<version>${commons-exec.version}</version> <version>${commons-exec.version}</version>
</dependency> </dependency>
<!-- jackson -->
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
<version>${jackson.version}</version>
</dependency>
<!-- spring-context --> <!-- spring-context -->
<dependency> <dependency>
<groupId>org.springframework</groupId> <groupId>org.springframework</groupId>

View File

@ -107,7 +107,9 @@ public class XxlJobExecutor {
// ---------------------- admin-client (rpc invoker) ---------------------- // ---------------------- admin-client (rpc invoker) ----------------------
private static List<AdminBiz> adminBizList; private static List<AdminBiz> adminBizList;
private static Serializer serializer;
private void initAdminBizList(String adminAddresses, String accessToken) throws Exception { private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
serializer = Serializer.SerializeEnum.HESSIAN.getSerializer();
if (adminAddresses!=null && adminAddresses.trim().length()>0) { if (adminAddresses!=null && adminAddresses.trim().length()>0) {
for (String address: adminAddresses.trim().split(",")) { for (String address: adminAddresses.trim().split(",")) {
if (address!=null && address.trim().length()>0) { if (address!=null && address.trim().length()>0) {
@ -116,7 +118,7 @@ public class XxlJobExecutor {
AdminBiz adminBiz = (AdminBiz) new XxlRpcReferenceBean( AdminBiz adminBiz = (AdminBiz) new XxlRpcReferenceBean(
NetEnum.NETTY_HTTP, NetEnum.NETTY_HTTP,
Serializer.SerializeEnum.HESSIAN.getSerializer(), serializer,
CallType.SYNC, CallType.SYNC,
LoadBalance.ROUND, LoadBalance.ROUND,
AdminBiz.class, AdminBiz.class,
@ -139,6 +141,9 @@ public class XxlJobExecutor {
public static List<AdminBiz> getAdminBizList(){ public static List<AdminBiz> getAdminBizList(){
return adminBizList; return adminBizList;
} }
public static Serializer getSerializer() {
return serializer;
}
// ---------------------- executor-server (rpc provider) ---------------------- // ---------------------- executor-server (rpc provider) ----------------------

View File

@ -8,7 +8,6 @@ import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.log.XxlJobFileAppender; import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.log.XxlJobLogger; import com.xxl.job.core.log.XxlJobLogger;
import com.xxl.job.core.util.FileUtil; import com.xxl.job.core.util.FileUtil;
import com.xxl.job.core.util.JacksonUtil;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
@ -190,46 +189,53 @@ public class TriggerCallbackThread {
// ---------------------- fail-callback file ---------------------- // ---------------------- fail-callback file ----------------------
private static String failCallbackFileName = XxlJobFileAppender.getLogPath().concat(File.separator).concat("xxl-job-callback").concat(".log"); private static String failCallbackFilePath = XxlJobFileAppender.getLogPath().concat(File.separator);
private static String failCallbackFileName = failCallbackFilePath.concat("xxl-job-callback-{x}").concat(".log");
private void appendFailCallbackFile(List<HandleCallbackParam> callbackParamList){ private void appendFailCallbackFile(List<HandleCallbackParam> callbackParamList){
// valid
if (callbackParamList==null || callbackParamList.size()==0) {
return;
}
// append file // append file
String content = JacksonUtil.writeValueAsString(callbackParamList); byte[] callbackParamList_bytes = XxlJobExecutor.getSerializer().serialize(callbackParamList);
FileUtil.appendFileLine(failCallbackFileName, content);
File callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis())));
if (callbackLogFile.exists()) {
for (int i = 0; i < 100; i++) {
callbackLogFile = new File(failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis()).concat("-").concat(String.valueOf(i)) ));
if (!callbackLogFile.exists()) {
break;
}
}
}
FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes);
} }
private void retryFailCallbackFile(){ private void retryFailCallbackFile(){
// load and clear file // valid
List<String> fileLines = FileUtil.loadFileLines(failCallbackFileName); File callbackLogPath = new File(failCallbackFilePath);
FileUtil.deleteFile(failCallbackFileName); if (!callbackLogPath.exists()) {
return;
// parse
List<HandleCallbackParam> failCallbackParamList = new ArrayList<>();
if (fileLines!=null && fileLines.size()>0) {
for (String line: fileLines) {
List<HandleCallbackParam> failCallbackParamListTmp = JacksonUtil.readValue(line, List.class, HandleCallbackParam.class);
if (failCallbackParamListTmp!=null && failCallbackParamListTmp.size()>0) {
failCallbackParamList.addAll(failCallbackParamListTmp);
} }
if (callbackLogPath.isFile()) {
callbackLogPath.delete();
} }
if (!(callbackLogPath.isDirectory() && callbackLogPath.list()!=null && callbackLogPath.list().length>0)) {
return;
}
// load and clear file, retry
for (File callbaclLogFile: callbackLogPath.listFiles()) {
byte[] callbackParamList_bytes = FileUtil.readFileContent(callbaclLogFile);
List<HandleCallbackParam> callbackParamList = (List<HandleCallbackParam>) XxlJobExecutor.getSerializer().deserialize(callbackParamList_bytes, HandleCallbackParam.class);
callbaclLogFile.delete();
doCallback(callbackParamList);
} }
// retry callback, 100 lines per page
if (failCallbackParamList!=null && failCallbackParamList.size()>0) {
int pagesize = 100;
List<HandleCallbackParam> pageData = new ArrayList<>();
for (int i = 0; i < failCallbackParamList.size(); i++) {
pageData.add(failCallbackParamList.get(i));
if (i>0 && i%pagesize == 0) {
doCallback(pageData);
pageData.clear();
}
}
if (pageData.size() > 0) {
doCallback(pageData);
}
}
} }
} }

View File

@ -3,9 +3,10 @@ package com.xxl.job.core.util;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import java.io.*; import java.io.File;
import java.util.ArrayList; import java.io.FileInputStream;
import java.util.List; import java.io.FileOutputStream;
import java.io.IOException;
/** /**
* file tool * file tool
@ -15,6 +16,7 @@ import java.util.List;
public class FileUtil { public class FileUtil {
private static Logger logger = LoggerFactory.getLogger(FileUtil.class); private static Logger logger = LoggerFactory.getLogger(FileUtil.class);
/** /**
* delete recursively * delete recursively
* *
@ -36,6 +38,7 @@ public class FileUtil {
return false; return false;
} }
public static void deleteFile(String fileName) { public static void deleteFile(String fileName) {
// file // file
File file = new File(fileName); File file = new File(fileName);
@ -44,7 +47,64 @@ public class FileUtil {
} }
} }
public static void appendFileLine(String fileName, String content) {
public static void writeFileContent(File file, byte[] data) {
// file
if (!file.exists()) {
try {
file.createNewFile();
} catch (IOException e) {
logger.error(e.getMessage(), e);
return;
}
}
// append file content
FileOutputStream fos = null;
try {
fos = new FileOutputStream(file);
fos.write(data);
fos.flush();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (fos != null) {
try {
fos.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}
public static byte[] readFileContent(File file) {
Long filelength = file.length();
byte[] filecontent = new byte[filelength.intValue()];
FileInputStream in = null;
try {
in = new FileInputStream(file);
in.read(filecontent);
in.close();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (in != null) {
try {
in.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
return filecontent;
}
/*public static void appendFileLine(String fileName, String content) {
// file // file
File file = new File(fileName); File file = new File(fileName);
@ -119,6 +179,6 @@ public class FileUtil {
} }
return result; return result;
} }*/
} }