- |
+ |
共{{results}}条记录，每页展示{{pageSize}}条
|
diff --git a/lts-admin/src/main/webapp/views/templates/nodeJvmInfo.vm b/lts-admin/src/main/webapp/views/templates/nodeJvmInfo.vm
index 8b309465a..53aa36991 100644
--- a/lts-admin/src/main/webapp/views/templates/nodeJvmInfo.vm
+++ b/lts-admin/src/main/webapp/views/templates/nodeJvmInfo.vm
@@ -106,9 +106,10 @@
showJVMInfo(identity, data);
});
} else {
- if (json) {
+ if (json && json['msg']) {
swal(json['msg']);
}
+ showJVMInfo('', {});
}
}
});
diff --git a/lts-admin/src/main/webapp/views/templates/nodeManager.vm b/lts-admin/src/main/webapp/views/templates/nodeManager.vm
index b2fe36b77..767b041ac 100644
--- a/lts-admin/src/main/webapp/views/templates/nodeManager.vm
+++ b/lts-admin/src/main/webapp/views/templates/nodeManager.vm
@@ -41,9 +41,9 @@
diff --git a/lts-core/src/main/java/com/lts/core/cluster/AbstractJobNode.java b/lts-core/src/main/java/com/lts/core/cluster/AbstractJobNode.java
index a4a29ea47..6a5946238 100644
--- a/lts-core/src/main/java/com/lts/core/cluster/AbstractJobNode.java
+++ b/lts-core/src/main/java/com/lts/core/cluster/AbstractJobNode.java
@@ -72,7 +72,9 @@ final public void stop() {
try {
if (started.compareAndSet(true, false)) {
- registry.unregister(node);
+ if (registry != null) {
+ registry.unregister(node);
+ }
preRemotingStop();
diff --git a/lts-core/src/main/java/com/lts/core/commons/utils/NetUtils.java b/lts-core/src/main/java/com/lts/core/commons/utils/NetUtils.java
index 7ba595103..95cf9cee9 100644
--- a/lts-core/src/main/java/com/lts/core/commons/utils/NetUtils.java
+++ b/lts-core/src/main/java/com/lts/core/commons/utils/NetUtils.java
@@ -183,7 +183,7 @@ private static InetAddress getLocalAddress0() {
} catch (Exception e) {
logger.warn("Failed to retriving ip address, " + e.getMessage(), e);
}
- logger.error("Could not get local host ip address, will use 127.0.0.1 instead.");
+// logger.error("Could not get local host ip address, will use 127.0.0.1 instead.");
return localAddress;
}
diff --git a/lts-core/src/main/java/com/lts/core/constant/Constants.java b/lts-core/src/main/java/com/lts/core/constant/Constants.java
index f1030f749..18628b73f 100644
--- a/lts-core/src/main/java/com/lts/core/constant/Constants.java
+++ b/lts-core/src/main/java/com/lts/core/constant/Constants.java
@@ -91,4 +91,11 @@ public interface Constants {
String ADMIN_ID_PREFIX = "LTS_admin_";
+ // 是否延迟批量刷盘日志, 如果启用，采用队列的方式批量将日志刷盘(在应用关闭的时候，可能会造成日志丢失)
+ String LAZY_JOB_LOGGER = "lazy.job.logger";
+ // 延迟批量刷盘日志 内存中的最大日志量阀值
+ String LAZY_JOB_LOGGER_MEM_SIZE = "lazy.job.logger.mem.size";
+ // 延迟批量刷盘日志 检查频率
+ String LAZY_JOB_LOGGER_CHECK_PERIOD = "lazy.job.logger.check.period";
+
}
diff --git a/lts-example/src/main/java/com/lts/example/api/JobTrackerTest.java b/lts-example/src/main/java/com/lts/example/api/JobTrackerTest.java
index d47694b4b..f6bf83c88 100644
--- a/lts-example/src/main/java/com/lts/example/api/JobTrackerTest.java
+++ b/lts-example/src/main/java/com/lts/example/api/JobTrackerTest.java
@@ -72,9 +72,12 @@ public static void testMysqlQueue() {
// 任务队列用mysql
jobTracker.addConfig("job.queue", "mysql");
// mysql 配置
- jobTracker.addConfig("jdbc.url", "jdbc:mysql://127.0.0.1:3306/lts");
- jobTracker.addConfig("jdbc.username", "root");
- jobTracker.addConfig("jdbc.password", "root");
+ jobTracker.addConfig("jdbc.url", "jdbc:mysql://127.0.0.1:3306/lts");
+ jobTracker.addConfig("jdbc.username", "root");
+ jobTracker.addConfig("jdbc.password", "root");
+
+ // 延迟批量刷盘业务日志开关
+// jobTracker.addConfig("lazy.job.logger", "true");
jobTracker.setOldDataHandler(new OldDataDeletePolicy());
// 设置 zk 客户端用哪个， 可选 zkclient, curator 默认是 zkclient
diff --git a/lts-jobtracker/src/main/java/com/lts/jobtracker/JobTracker.java b/lts-jobtracker/src/main/java/com/lts/jobtracker/JobTracker.java
index 449e9fe5d..dd34a51ca 100644
--- a/lts-jobtracker/src/main/java/com/lts/jobtracker/JobTracker.java
+++ b/lts-jobtracker/src/main/java/com/lts/jobtracker/JobTracker.java
@@ -1,5 +1,6 @@
package com.lts.jobtracker;
+import com.lts.biz.logger.JobLoggerDelegate;
import com.lts.biz.logger.JobLoggerFactory;
import com.lts.core.cluster.AbstractServerNode;
import com.lts.core.extension.ExtensionLoader;
@@ -21,8 +22,6 @@
*/
public class JobTracker extends AbstractServerNode {
- private JobLoggerFactory jobLoggerFactory = ExtensionLoader.getExtensionLoader(JobLoggerFactory.class).getAdaptiveExtension();
-
private CronJobQueueFactory cronJobQueueFactory =
ExtensionLoader.getExtensionLoader(CronJobQueueFactory.class).getAdaptiveExtension();
private ExecutableJobQueueFactory executableJobQueueFactory =
@@ -56,7 +55,7 @@ protected void preRemotingStart() {
super.preRemotingStart();
// injectRemotingServer
application.setRemotingServer(remotingServer);
- application.setJobLogger(jobLoggerFactory.getJobLogger(config));
+ application.setJobLogger(new JobLoggerDelegate(config));
application.setExecutableJobQueue(executableJobQueueFactory.getQueue(config));
application.setExecutingJobQueue(executingJobQueueFactory.getQueue(config));
application.setCronJobQueue(cronJobQueueFactory.getQueue(config));
diff --git a/lts-jobtracker/src/main/java/com/lts/jobtracker/support/JobReceiver.java b/lts-jobtracker/src/main/java/com/lts/jobtracker/support/JobReceiver.java
index d79962c2c..4a21e4f22 100644
--- a/lts-jobtracker/src/main/java/com/lts/jobtracker/support/JobReceiver.java
+++ b/lts-jobtracker/src/main/java/com/lts/jobtracker/support/JobReceiver.java
@@ -1,6 +1,9 @@
package com.lts.jobtracker.support;
+import com.lts.biz.logger.domain.JobLogPo;
+import com.lts.biz.logger.domain.LogType;
import com.lts.core.commons.utils.StringUtils;
+import com.lts.core.constant.Level;
import com.lts.core.domain.Job;
import com.lts.core.exception.JobReceiveException;
import com.lts.core.extension.ExtensionLoader;
@@ -8,6 +11,7 @@
import com.lts.core.logger.LoggerFactory;
import com.lts.core.protocol.command.JobSubmitRequest;
import com.lts.core.support.LoggerName;
+import com.lts.core.support.SystemClock;
import com.lts.jobtracker.domain.JobTrackerApplication;
import com.lts.jobtracker.id.IdGenerator;
import com.lts.jobtracker.monitor.JobTrackerMonitor;
@@ -68,6 +72,7 @@ private JobPo addToQueue(Job job, JobSubmitRequest request) {
JobPo jobPo = null;
+ boolean duplicate = false;
try {
jobPo = JobDomainConverter.convert(job);
if (jobPo == null) {
@@ -82,19 +87,39 @@ private JobPo addToQueue(Job job, JobSubmitRequest request) {
if (job.isSchedule()) {
addCronJob(jobPo);
- LOGGER.info("Receive cron job success. nodeGroup={}, CronExpression={}, {}",
+ LOGGER.info("Receive Cron Job success. nodeGroup={}, CronExpression={}, {}",
request.getNodeGroup(), job.getCronExpression(), job);
} else {
application.getExecutableJobQueue().add(jobPo);
- LOGGER.info("Receive job success. nodeGroup={}, {}", request.getNodeGroup(), job);
+ LOGGER.info("Receive Job success. nodeGroup={}, {}", request.getNodeGroup(), job);
}
+
} catch (DuplicateJobException e) {
// already exist, ignore
LOGGER.info("Job already exist. nodeGroup={}, {}", request.getNodeGroup(), job);
+ duplicate = true;
} finally {
monitor.incReceiveJobNum();
}
+ if (jobPo != null) {
+ try {
+ // 记录日志
+ JobLogPo jobLogPo = JobDomainConverter.convertJobLog(jobPo);
+ jobLogPo.setSuccess(true);
+ jobLogPo.setLogType(LogType.RECEIVE);
+ jobLogPo.setLogTime(SystemClock.now());
+ jobLogPo.setLevel(Level.INFO);
+ if (duplicate) {
+ jobLogPo.setLevel(Level.WARN);
+ jobLogPo.setMsg("在任务队列中已经存在,忽略本次提交.");
+ }
+ application.getJobLogger().log(jobLogPo);
+ } catch (Throwable t) { // 日志记录失败不影响正常运行
+ LOGGER.error("Receive Job Log error ", t);
+ }
+ }
+
return jobPo;
}
diff --git a/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/JobLoggerDelegate.java b/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/JobLoggerDelegate.java
new file mode 100644
index 000000000..838add923
--- /dev/null
+++ b/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/JobLoggerDelegate.java
@@ -0,0 +1,179 @@
+package com.lts.biz.logger;
+
+import com.lts.biz.logger.domain.JobLogPo;
+import com.lts.biz.logger.domain.JobLoggerRequest;
+import com.lts.core.cluster.Config;
+import com.lts.core.commons.utils.CollectionUtils;
+import com.lts.core.constant.Constants;
+import com.lts.core.extension.ExtensionLoader;
+import com.lts.core.factory.NamedThreadFactory;
+import com.lts.core.logger.Logger;
+import com.lts.core.logger.LoggerFactory;
+import com.lts.web.response.PageResponse;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * 内部根据用户参数决定是否采用延迟批量刷盘的策略,来提高吞吐量
+ * 批量刷盘有两种情况:
+ * 1. 内存的日志量超过了设置的阀值
+ * 2. 每3S检查一次内存中是否有日志,如果有就那么刷盘
+ *
+ * @author Robert HG (254963746@qq.com) on 10/2/15.
+ */
+public class JobLoggerDelegate implements JobLogger {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(JobLoggerDelegate.class);
+
+ // 3S 检查输盘一次日志
+ private int flushPeriod;
+
+ private JobLogger jobLogger;
+ private boolean lazyLog = false;
+ private ScheduledExecutorService executor;
+ private ScheduledFuture scheduledFuture;
+ private BlockingQueue memoryQueue;
+ // 日志批量刷盘数量
+ private int batchFlushSize = 100;
+ private int overflowSize = 10000;
+ // 内存中最大的日志量阀值
+ private int maxMemoryLogSize;
+ private AtomicBoolean flushing = new AtomicBoolean(false);
+
+ public JobLoggerDelegate(Config config) {
+ JobLoggerFactory jobLoggerFactory = ExtensionLoader
+ .getExtensionLoader(JobLoggerFactory.class).getAdaptiveExtension();
+ jobLogger = jobLoggerFactory.getJobLogger(config);
+ lazyLog = config.getParameter(Constants.LAZY_JOB_LOGGER, false);
+ if (lazyLog) {
+
+ // 无界Queue
+ memoryQueue = new LinkedBlockingQueue();
+ maxMemoryLogSize = config.getParameter(Constants.LAZY_JOB_LOGGER_MEM_SIZE, 1000);
+ flushPeriod = config.getParameter(Constants.LAZY_JOB_LOGGER_CHECK_PERIOD, 3);
+
+ executor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LazyJobLogger"));
+ scheduledFuture = executor.scheduleWithFixedDelay(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ if (flushing.compareAndSet(false, true)) {
+ checkAndFlush();
+ }
+ } catch (Throwable t) {
+ LOGGER.error("CheckAndFlush log error", t);
+ }
+ }
+ }, flushPeriod, flushPeriod, TimeUnit.SECONDS);
+
+ }
+ }
+
+ /**
+ * 检查内存中是否有日志,如果有就批量刷盘
+ */
+ private void checkAndFlush() {
+ try {
+ int nowSize = memoryQueue.size();
+ if (nowSize == 0) {
+ return;
+ }
+ List batch = new ArrayList();
+ for (int i = 0; i < nowSize; i++) {
+ JobLogPo jobLogPo = memoryQueue.poll();
+ batch.add(jobLogPo);
+
+ if (batch.size() >= batchFlushSize) {
+ flush(batch);
+ }
+ }
+ if (batch.size() > 0) {
+ flush(batch);
+ }
+
+ } finally {
+ flushing.compareAndSet(true, false);
+ }
+ }
+
+ private void checkOverflowSize() {
+ if (memoryQueue.size() > overflowSize) {
+ throw new JobLogException("Memory Log size is " + memoryQueue.size() + " , please check the JobLogger is available");
+ }
+ }
+
+ private void flush(List batch) {
+ boolean flushSuccess = false;
+ try {
+ jobLogger.log(batch);
+ flushSuccess = true;
+ } finally {
+ if (!flushSuccess) {
+ memoryQueue.addAll(batch);
+ }
+ batch.clear();
+ }
+ }
+
+ /**
+ * 检查内存中的日志量是否超过阀值,如果超过需要批量刷盘日志
+ */
+ private void checkCapacity() {
+ if (memoryQueue.size() > maxMemoryLogSize) {
+ // 超过阀值,需要批量刷盘
+ if (flushing.compareAndSet(false, true)) {
+ // 这里可以采用new Thread, 因为这里只会同时new一个
+ new Thread(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ checkAndFlush();
+ } catch (Throwable t) {
+ LOGGER.error("Capacity full flush error", t);
+ }
+ }
+ }).start();
+ }
+ }
+ }
+
+ @Override
+ public void log(JobLogPo jobLogPo) {
+ if (jobLogPo == null) {
+ return;
+ }
+ if (lazyLog) {
+ checkOverflowSize();
+ memoryQueue.offer(jobLogPo);
+ checkCapacity();
+ } else {
+ jobLogger.log(jobLogPo);
+ }
+ }
+
+ @Override
+ public void log(List jobLogPos) {
+ if (CollectionUtils.isEmpty(jobLogPos)) {
+ return;
+ }
+ if (lazyLog) {
+ checkOverflowSize();
+ for (JobLogPo jobLogPo : jobLogPos) {
+ memoryQueue.offer(jobLogPo);
+ }
+ // checkCapacity
+ checkCapacity();
+ } else {
+ jobLogger.log(jobLogPos);
+ }
+ }
+
+ @Override
+ public PageResponse search(JobLoggerRequest request) {
+ return jobLogger.search(request);
+ }
+
+}
diff --git a/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/domain/LogType.java b/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/domain/LogType.java
index 6af6bbab4..46aff7763 100644
--- a/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/domain/LogType.java
+++ b/lts-logger/lts-logger-api/src/main/java/com/lts/biz/logger/domain/LogType.java
@@ -5,6 +5,7 @@
*/
public enum LogType {
+ RECEIVE, // 接受任务
SENT, // 任务发送 开始执行
FINISHED, // 任务执行完成
RESEND, // TaskTracker 重新发送的任务执行结果
diff --git a/lts-logger/lts-logger-mysql/src/main/java/com/lts/biz/logger/mysql/MysqlJobLogger.java b/lts-logger/lts-logger-mysql/src/main/java/com/lts/biz/logger/mysql/MysqlJobLogger.java
index af28a5aa5..97e922837 100644
--- a/lts-logger/lts-logger-mysql/src/main/java/com/lts/biz/logger/mysql/MysqlJobLogger.java
+++ b/lts-logger/lts-logger-mysql/src/main/java/com/lts/biz/logger/mysql/MysqlJobLogger.java
@@ -11,9 +11,9 @@
import com.lts.core.commons.utils.CollectionUtils;
import com.lts.core.commons.utils.JSONUtils;
import com.lts.core.constant.Level;
-import com.lts.web.response.PageResponse;
import com.lts.store.jdbc.JdbcRepository;
import com.lts.store.jdbc.SqlBuilder;
+import com.lts.web.response.PageResponse;
import org.apache.commons.dbutils.ResultSetHandler;
import java.io.IOException;
@@ -78,18 +78,7 @@ public void log(List jobLogPos) {
if (CollectionUtils.isEmpty(jobLogPos)) {
return;
}
- String prefixSQL = "INSERT INTO `lts_job_log_po` ( `log_time`, `gmt_created`, `log_type`, `success`, `msg`" +
- ",`task_tracker_identity`, `level`, `task_id`, `job_id`" +
- ", `priority`, `submit_node_group`, `task_tracker_node_group`, `ext_params`, `need_feedback`" +
- ", `cron_expression`, `trigger_time`, `retry_times`) VALUES ";
int size = jobLogPos.size();
- for (int i = 0; i < size; i++) {
- if (i == size - 1) {
- prefixSQL += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)";
- } else {
- prefixSQL += "(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?),";
- }
- }
Object[][] params = new Object[size][17];
int index = 0;
@@ -115,7 +104,7 @@ public void log(List jobLogPos) {
}
try {
- getSqlTemplate().batchUpdate(prefixSQL, params);
+ getSqlTemplate().batchUpdate(insertSQL, params);
} catch (SQLException e) {
throw new JobLogException(e);
}