package tech.powerjob.worker.core.tracker.task.heavy;
|
|
import com.google.common.base.Stopwatch;
|
import com.google.common.collect.Lists;
|
import com.google.common.collect.Maps;
|
import com.google.common.util.concurrent.ThreadFactoryBuilder;
|
import lombok.Data;
|
import lombok.extern.slf4j.Slf4j;
|
import org.apache.commons.lang3.StringUtils;
|
import tech.powerjob.common.enums.ExecuteType;
|
import tech.powerjob.common.enums.InstanceStatus;
|
import tech.powerjob.common.enums.TimeExpressionType;
|
import tech.powerjob.common.exception.PowerJobException;
|
import tech.powerjob.common.model.AlarmConfig;
|
import tech.powerjob.common.model.InstanceDetail;
|
import tech.powerjob.common.request.ServerQueryInstanceStatusReq;
|
import tech.powerjob.common.request.ServerScheduleJobReq;
|
import tech.powerjob.common.request.TaskTrackerReportInstanceStatusReq;
|
import tech.powerjob.common.serialize.JsonUtils;
|
import tech.powerjob.worker.common.WorkerRuntime;
|
import tech.powerjob.worker.common.constants.TaskConstant;
|
import tech.powerjob.worker.common.constants.TaskStatus;
|
import tech.powerjob.worker.common.utils.LRUCache;
|
import tech.powerjob.worker.common.utils.TransportUtils;
|
import tech.powerjob.worker.core.processor.TaskResult;
|
import tech.powerjob.worker.persistence.TaskDO;
|
|
import java.util.*;
|
import java.util.concurrent.Executors;
|
import java.util.concurrent.ThreadFactory;
|
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.atomic.AtomicLong;
|
|
/**
|
* 处理秒级任务（FIX_RATE/FIX_DELAY）的TaskTracker
|
* FIX_RATE 直接由 ScheduledExecutorService 实现，精度高，推荐使用
|
* FIX_DELAY 会有几秒的延迟，精度不是很理想
|
*
|
* @author tjq
|
* @since 2020/4/8
|
*/
|
@Slf4j
|
public class FrequentTaskTracker extends HeavyTaskTracker {
|
|
/**
|
* 时间表达式类型
|
*/
|
private TimeExpressionType timeExpressionType;
|
|
private long timeParams;
|
/**
|
* 最大同时运行实例数
|
*/
|
private int maxInstanceNum;
|
|
/**
|
* 总运行次数（正常情况不会出现锁竞争，直接用 Atomic 系列，锁竞争严重推荐 LongAdder）
|
*/
|
private AtomicLong triggerTimes;
|
|
private AtomicLong succeedTimes;
|
|
private AtomicLong failedTimes;
|
/**
|
* 任务发射器
|
*/
|
private Launcher launcher;
|
/**
|
* 保存最近10个子任务的信息，供用户查询（user -> server -> worker 传递查询）
|
*/
|
private LRUCache<Long, SubInstanceInfo> recentSubInstanceInfo;
|
/**
|
* 保存运行中的任务
|
*/
|
private Map<Long, SubInstanceTimeHolder> subInstanceId2TimeHolder;
|
|
private AlertManager alertManager;
|
|
private static final int HISTORY_SIZE = 10;
|
private static final String LAST_TASK_ID_PREFIX = "L";
|
private static final int MIN_INTERVAL = 50;
|
|
protected FrequentTaskTracker(ServerScheduleJobReq req, WorkerRuntime workerRuntime) {
|
super(req, workerRuntime);
|
}
|
|
@Override
|
protected void initTaskTracker(ServerScheduleJobReq req) {
|
|
// 0. 初始化实例变量
|
timeExpressionType = TimeExpressionType.valueOf(req.getTimeExpressionType());
|
timeParams = Long.parseLong(req.getTimeExpression());
|
maxInstanceNum = req.getMaxInstanceNum();
|
|
triggerTimes = new AtomicLong(0);
|
succeedTimes = new AtomicLong(0);
|
failedTimes = new AtomicLong(0);
|
|
recentSubInstanceInfo = new LRUCache<>(HISTORY_SIZE);
|
subInstanceId2TimeHolder = Maps.newConcurrentMap();
|
|
// 1. 初始化定时调度线程池
|
String poolName = String.format("ftttp-%d", req.getInstanceId()) + "-%d";
|
ThreadFactory factory = new ThreadFactoryBuilder().setNameFormat(poolName).build();
|
this.scheduledPool = Executors.newScheduledThreadPool(4, factory);
|
this.alertManager = constructAlertManager(req);
|
// 2. 启动任务发射器
|
launcher = new Launcher();
|
if (timeExpressionType == TimeExpressionType.FIXED_RATE) {
|
// 固定频率需要设置最小间隔
|
if (timeParams < MIN_INTERVAL) {
|
throw new PowerJobException("time interval too small, please set the timeExpressionInfo >= 1000");
|
}
|
scheduledPool.scheduleAtFixedRate(launcher, 1, timeParams, TimeUnit.MILLISECONDS);
|
} else {
|
scheduledPool.schedule(launcher, 0, TimeUnit.MILLISECONDS);
|
}
|
|
// 3. 启动任务分发器（事实上，秒级任务应该都是单机任务，且感觉不需要失败重试机制，那么 Dispatcher 的存在就有点浪费系统资源了...）
|
scheduledPool.scheduleWithFixedDelay(new Dispatcher(), 1, 2, TimeUnit.SECONDS);
|
// 4. 启动状态检查器
|
scheduledPool.scheduleWithFixedDelay(new Checker(), 5000, Math.min(Math.max(timeParams, 5000), 15000), TimeUnit.MILLISECONDS);
|
// 5. 启动执行器动态检测装置
|
scheduledPool.scheduleAtFixedRate(new WorkerDetector(), 1, 1, TimeUnit.MINUTES);
|
}
|
|
@Override
|
public InstanceDetail fetchRunningStatus(ServerQueryInstanceStatusReq req) {
|
InstanceDetail detail = new InstanceDetail();
|
// 填充基础信息
|
detail.setActualTriggerTime(createTime);
|
detail.setStatus(InstanceStatus.RUNNING.getV());
|
detail.setTaskTrackerAddress(workerRuntime.getWorkerAddress());
|
|
List<InstanceDetail.SubInstanceDetail> history = Lists.newLinkedList();
|
recentSubInstanceInfo.forEach((subId, subInstanceInfo) -> {
|
InstanceDetail.SubInstanceDetail subDetail = new InstanceDetail.SubInstanceDetail();
|
|
subDetail.setSubInstanceId(subId);
|
subDetail.setStartTime(subInstanceInfo.getStartTime());
|
subDetail.setFinishedTime(subInstanceInfo.getFinishedTime());
|
subDetail.setResult(subInstanceInfo.getResult());
|
|
InstanceStatus status = InstanceStatus.of(subInstanceInfo.status);
|
subDetail.setStatus(status.getV());
|
|
history.add(subDetail);
|
});
|
|
// 按 subInstanceId 排序 issue#63
|
history.sort((o1, o2) -> (int) (o2.getSubInstanceId() - o1.getSubInstanceId()));
|
|
detail.setSubInstanceDetails(history);
|
return detail;
|
}
|
|
/**
|
* 任务发射器（@Reference 饥荒->雪球发射器）
|
*/
|
private class Launcher implements Runnable {
|
|
public void innerRun() {
|
|
if (finished.get()) {
|
return;
|
}
|
|
// 子任务实例ID
|
Long subInstanceId = triggerTimes.incrementAndGet();
|
|
// 执行记录缓存（只做展示，因此可以放在前面）
|
SubInstanceInfo subInstanceInfo = new SubInstanceInfo();
|
subInstanceInfo.status = TaskStatus.DISPATCH_SUCCESS_WORKER_UNCHECK.getValue();
|
subInstanceInfo.startTime = System.currentTimeMillis();
|
recentSubInstanceInfo.put(subInstanceId, subInstanceInfo);
|
|
String myAddress = workerRuntime.getWorkerAddress();
|
String taskId = String.valueOf(subInstanceId);
|
|
TaskDO newRootTask = new TaskDO();
|
|
newRootTask.setInstanceId(instanceId);
|
newRootTask.setSubInstanceId(subInstanceId);
|
newRootTask.setTaskId(taskId);
|
|
newRootTask.setStatus(TaskStatus.DISPATCH_SUCCESS_WORKER_UNCHECK.getValue());
|
newRootTask.setFailedCnt(0);
|
// 根任务总是默认本机执行
|
newRootTask.setAddress(myAddress);
|
newRootTask.setTaskName(TaskConstant.ROOT_TASK_NAME);
|
newRootTask.setCreatedTime(System.currentTimeMillis());
|
newRootTask.setLastModifiedTime(System.currentTimeMillis());
|
newRootTask.setLastReportTime(-1L);
|
|
// 判断是否超出最大执行实例数
|
if (maxInstanceNum > 0) {
|
if (timeExpressionType == TimeExpressionType.FIXED_RATE) {
|
if (subInstanceId2TimeHolder.size() >= maxInstanceNum) {
|
log.warn("[FQTaskTracker-{}] cancel to launch the subInstance({}) due to too much subInstance is running.", instanceId, subInstanceId);
|
processFinishedSubInstance(subInstanceId, false, "TOO_MUCH_INSTANCE");
|
return;
|
}
|
}
|
}
|
|
// 必须先持久化，持久化成功才能 dispatch，否则会导致后续报错（因为DB中没有这个taskId对应的记录，会各种报错）
|
if (!taskPersistenceService.batchSave(Lists.newArrayList(newRootTask))) {
|
log.error("[FQTaskTracker-{}] Launcher create new root task failed.", instanceId);
|
processFinishedSubInstance(subInstanceId, false, "LAUNCH_FAILED");
|
return;
|
}
|
|
// 生成记录信息（必须保证持久化成功才能生成该记录，否则会导致 LAUNCH_FAILED 错误）
|
SubInstanceTimeHolder timeHolder = new SubInstanceTimeHolder();
|
timeHolder.startTime = System.currentTimeMillis();
|
subInstanceId2TimeHolder.put(subInstanceId, timeHolder);
|
|
dispatchTask(newRootTask, myAddress);
|
}
|
|
@Override
|
public void run() {
|
try {
|
innerRun();
|
} catch (Exception e) {
|
log.error("[FQTaskTracker-{}] launch task failed.", instanceId, e);
|
}
|
}
|
}
|
|
/**
|
* 检查各个SubInstance的完成情况
|
*/
|
private class Checker implements Runnable {
|
|
@Override
|
public void run() {
|
|
if (finished.get()) {
|
return;
|
}
|
|
try {
|
checkStatus();
|
reportStatus();
|
} catch (Exception e) {
|
log.warn("[FQTaskTracker-{}] check and report status failed.", instanceId, e);
|
}
|
}
|
|
private void checkStatus() {
|
Stopwatch stopwatch = Stopwatch.createStarted();
|
|
// worker 挂掉的任务直接置为失败
|
List<String> disconnectedPTs = ptStatusHolder.getAllDisconnectedProcessorTrackers();
|
if (!disconnectedPTs.isEmpty()) {
|
log.warn("[FQTaskTracker-{}] some ProcessorTracker disconnected from TaskTracker,their address is {}.", instanceId, disconnectedPTs);
|
if (taskPersistenceService.updateLostTasks(instanceId, disconnectedPTs, false)) {
|
ptStatusHolder.remove(disconnectedPTs);
|
log.warn("[FQTaskTracker-{}] removed these ProcessorTracker from StatusHolder: {}", instanceId, disconnectedPTs);
|
}
|
}
|
|
ExecuteType executeType = ExecuteType.valueOf(instanceInfo.getExecuteType());
|
long instanceTimeoutMS = instanceInfo.getInstanceTimeoutMS();
|
long nowTS = System.currentTimeMillis();
|
|
Iterator<Map.Entry<Long, SubInstanceTimeHolder>> iterator = subInstanceId2TimeHolder.entrySet().iterator();
|
while (iterator.hasNext()) {
|
|
Map.Entry<Long, SubInstanceTimeHolder> entry = iterator.next();
|
Long subInstanceId = entry.getKey();
|
SubInstanceTimeHolder timeHolder = entry.getValue();
|
|
long executeTimeout = nowTS - timeHolder.startTime;
|
|
// 超时（包含总运行时间超时和心跳包超时），直接判定为失败
|
|
if (executeTimeout > instanceTimeoutMS) {
|
onFinished(subInstanceId, false, "RUNNING_TIMEOUT", iterator);
|
continue;
|
}
|
|
// 查看执行情况
|
InstanceStatisticsHolder holder = getInstanceStatisticsHolder(subInstanceId);
|
|
long finishedNum = holder.succeedNum + holder.failedNum;
|
long unfinishedNum = holder.waitingDispatchNum + holder.workerUnreceivedNum + holder.receivedNum + holder.runningNum;
|
|
if (unfinishedNum == 0) {
|
|
// 数据库中没有该 subInstanceId 的记录，说明任务发射器写入DB失败，直接视为执行失败，删除数据
|
if (finishedNum == 0) {
|
onFinished(subInstanceId, false, "LAUNCH_FAILED", iterator);
|
continue;
|
}
|
|
switch (executeType) {
|
// STANDALONE 代表任务确实已经执行完毕了
|
case STANDALONE:
|
// 查询数据库获取结果（STANDALONE每个SubInstance只会有一条Task记录）
|
TaskResult resultTask = taskPersistenceService.getAllTaskResult(instanceId, subInstanceId).get(0);
|
onFinished(subInstanceId, resultTask.isSuccess(), resultTask.getResult(), iterator);
|
continue;
|
// MAP 不关心结果，最简单
|
case MAP:
|
String result = String.format("total:%d,succeed:%d,failed:%d", holder.getTotalTaskNum(), holder.succeedNum, holder.failedNum);
|
onFinished(subInstanceId, holder.failedNum == 0, result, iterator);
|
continue;
|
// MapReduce 和 BroadCast 需要根据是否有 LAST_TASK 来判断结束与否
|
default:
|
Optional<TaskDO> lastTaskOptional = taskPersistenceService.getLastTask(instanceId, subInstanceId);
|
if (lastTaskOptional.isPresent()) {
|
|
TaskStatus lastTaskStatus = TaskStatus.of(lastTaskOptional.get().getStatus());
|
if (lastTaskStatus == TaskStatus.WORKER_PROCESS_SUCCESS || lastTaskStatus == TaskStatus.WORKER_PROCESS_FAILED) {
|
onFinished(subInstanceId, lastTaskStatus == TaskStatus.WORKER_PROCESS_SUCCESS, lastTaskOptional.get().getResult(), iterator);
|
}
|
} else {
|
|
// 创建最终任务并提交执行
|
TaskDO newLastTask = new TaskDO();
|
newLastTask.setTaskName(TaskConstant.LAST_TASK_NAME);
|
newLastTask.setTaskId(LAST_TASK_ID_PREFIX + subInstanceId);
|
newLastTask.setSubInstanceId(subInstanceId);
|
newLastTask.setAddress(workerRuntime.getWorkerAddress());
|
submitTask(Lists.newArrayList(newLastTask));
|
}
|
}
|
}
|
// 舍去一切重试机制，反正超时就失败
|
}
|
log.debug("[FQTaskTracker-{}] check status using {}.", instanceId, stopwatch);
|
}
|
|
private void reportStatus() {
|
|
String currentServerAddress = workerRuntime.getServerDiscoveryService().getCurrentServerAddress();
|
if (StringUtils.isEmpty(currentServerAddress)) {
|
return;
|
}
|
|
TaskTrackerReportInstanceStatusReq req = new TaskTrackerReportInstanceStatusReq();
|
req.setAppId(workerRuntime.getAppId());
|
req.setJobId(instanceInfo.getJobId());
|
req.setInstanceId(instanceId);
|
req.setReportTime(System.currentTimeMillis());
|
req.setStartTime(createTime);
|
req.setInstanceStatus(InstanceStatus.RUNNING.getV());
|
|
req.setTotalTaskNum(triggerTimes.get());
|
req.setSucceedTaskNum(succeedTimes.get());
|
req.setFailedTaskNum(failedTimes.get());
|
req.setSourceAddress(workerRuntime.getWorkerAddress());
|
|
// alert
|
if (alertManager.alert()) {
|
req.setNeedAlert(true);
|
req.setAlertContent(alertManager.getAlertContent());
|
log.warn("[FQTaskTracker-{}] report alert req,time:{}", instanceId, req.getReportTime());
|
}
|
|
// 非可靠通知，Server挂掉后任务的kill工作交由其他线程去做
|
TransportUtils.ttReportInstanceStatus(req, currentServerAddress, workerRuntime.getTransporter());
|
}
|
|
/**
|
* 处理任务完成的情况，删除内存 & 数据库数据
|
*/
|
private void onFinished(Long subInstanceId, boolean success, String result, Iterator<?> iterator) {
|
iterator.remove();
|
processFinishedSubInstance(subInstanceId, success, result);
|
}
|
}
|
|
private void processFinishedSubInstance(long subInstanceId, boolean success, String result) {
|
long currentTime = System.currentTimeMillis();
|
if (success) {
|
succeedTimes.incrementAndGet();
|
} else {
|
failedTimes.incrementAndGet();
|
alertManager.update(currentTime, result);
|
}
|
|
// 从运行中任务列表移除
|
subInstanceId2TimeHolder.remove(subInstanceId);
|
|
// 更新缓存数据
|
SubInstanceInfo subInstanceInfo = recentSubInstanceInfo.get(subInstanceId);
|
if (subInstanceInfo != null) {
|
subInstanceInfo.status = success ? InstanceStatus.SUCCEED.getV() : InstanceStatus.FAILED.getV();
|
subInstanceInfo.result = result;
|
subInstanceInfo.finishedTime = currentTime;
|
}
|
// 删除数据库相关数据
|
taskPersistenceService.deleteAllSubInstanceTasks(instanceId, subInstanceId);
|
|
// FIX_DELAY 则调度下次任务
|
if (timeExpressionType == TimeExpressionType.FIXED_DELAY) {
|
scheduledPool.schedule(launcher, timeParams, TimeUnit.MILLISECONDS);
|
}
|
}
|
|
|
private AlertManager constructAlertManager(ServerScheduleJobReq req) {
|
|
long rate = Long.parseLong(req.getTimeExpression());
|
if (!StringUtils.isEmpty(req.getAlarmConfig())) {
|
try {
|
log.debug("[FQTaskTracker-{}] alert config:{}", instanceId, req.getAlarmConfig());
|
AlarmConfig alarmConfig = JsonUtils.parseObject(req.getAlarmConfig(), AlarmConfig.class);
|
return new AlertManager(alarmConfig);
|
} catch (Exception e) {
|
log.warn("[FQTaskTracker-{}] constructAlertManager failed, will use default config!", instanceId);
|
}
|
}
|
// 默认配置，失败一次就告警，沉默窗口 5 分钟
|
int statisticWindowLen = Math.max((int) (2 * rate / 1000), 1);
|
return new AlertManager(new AlarmConfig(1, statisticWindowLen, 300));
|
}
|
|
private class AlertManager {
|
/**
|
* 记录执行失败的时间
|
*/
|
private final LinkedList<Long> failedRecordList = new LinkedList<>();
|
/**
|
* 告警配置
|
*/
|
private final AlarmConfig config;
|
/**
|
* 告警的激活时间
|
*/
|
private long alarmActiveTime = 0L;
|
/**
|
* 告警的内容（记录最后一次失败的任务执行结果）
|
*/
|
private String content;
|
/**
|
* 是否处于激活状态
|
*/
|
private boolean active;
|
|
|
public AlertManager(AlarmConfig config) {
|
log.info("[FQTaskTracker-{}] create alert manager,alertThreshold:{},statisticWindowLen:{} s,silenceWindowLen:{} s", instanceId, config.getAlertThreshold(), config.getStatisticWindowLen(), config.getSilenceWindowLen());
|
this.config = config;
|
}
|
|
public synchronized void update(long currentTime, String result) {
|
log.debug("[FQTaskTracker-{}] update alert statistic info,currentTime:{}", instanceId, currentTime);
|
if (currentTime < alarmActiveTime + config.getSilenceWindowLen() * 1000) {
|
// 处于沉默窗口内
|
return;
|
}
|
// 当前统计窗口允许的最小值
|
long minTime = currentTime - config.getStatisticWindowLen() * 1000;
|
while (!failedRecordList.isEmpty() && failedRecordList.peekFirst() < minTime) {
|
failedRecordList.removeFirst();
|
}
|
failedRecordList.add(currentTime);
|
if (failedRecordList.size() >= config.getAlertThreshold()) {
|
// 标记需要告警
|
active = true;
|
alarmActiveTime = currentTime;
|
content = result;
|
}
|
}
|
|
|
public synchronized boolean alert() {
|
if (active) {
|
active = false;
|
return true;
|
}
|
return false;
|
}
|
|
public String getAlertContent() {
|
return content;
|
}
|
}
|
|
@Data
|
private static class SubInstanceInfo {
|
private int status;
|
private long startTime;
|
private long finishedTime;
|
private String result;
|
}
|
|
private static class SubInstanceTimeHolder {
|
private long startTime;
|
}
|
|
}
|
