treshenkov/clearing
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

http://jira.mfd.msk:8088/browse/CLS-57

Browse source 
---
task manager update and launcher service
This commit is contained in:
aalehin 2022-12-15 19:24:25 +03:00
parent 30a3f17200
commit 7e70f2ee94
2 changed files with 59 additions and 47 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/LauncherService.java
View file

@ -12,11 +12,14 @@ import ru.spcex.clearing.platform.messaging.domain.BaseRequest;
import ru.spcex.clearing.platform.messaging.domain.Consts;
import ru.spcex.clearing.platform.messaging.domain.cud.schedule.LauncherCommandRequest;
import ru.spcex.clearing.platform.messaging.service.QueueConsumer;
import ru.spcex.platform.enumeration.Task;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import java.time.Instant;
import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey;
public class LauncherService extends QueueConsumer implements InitializingBean {
private final Logger log = LoggerFactory.getLogger(getClass());
private final Imdg<Launcher> launcherMap;
@ -41,6 +44,9 @@ public class LauncherService extends QueueConsumer implements InitializingBean {
log.debug("LauncherCommandRequest received");
Instant created = Instant.now();
Launcher launcher = new Launcher();
if (getEnumByKey(Task.class, req.getTaskName()) == null) {
throw new IllegalStateException("Illegal task name:\t" + req.getTaskName());
}
launcher.setTask(req.getTaskName());
launcher.setSenderId(req.getUserId());
launcher.setCreated(created);

100
clearing-parent/scheduler-service/src/main/java/ru/spcex/clearing/scheduler/service/TaskManager.java
View file

@ -1,23 +1,25 @@
package ru.spcex.clearing.scheduler.service;
import com.hazelcast.core.EntryEvent;
import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.core.IMap;
import com.hazelcast.map.listener.EntryAddedListener;
import com.hazelcast.map.listener.EntryRemovedListener;
import com.hazelcast.map.listener.EntryUpdatedListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.stereotype.Service;
import ru.clearing.classes.statics.data.scheduler.Launcher;
import ru.clearing.classes.statics.data.scheduler.PlannerAllToday;
import ru.spcex.platform.enumeration.Status;
import ru.spcex.platform.enumeration.Task;
import ru.spcex.platform.imdg.api.Imdg;
import ru.spcex.platform.imdg.api.ImdgProvider;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZoneId;
import java.time.*;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
@ -26,6 +28,7 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledFuture;
import java.util.stream.Collectors;
import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_Launcher;
import static ru.spcex.clearing.imdg.IMDGDistributedNames.Map_PlannerAllToday;
import static ru.spcex.platform.enumeration.Status.*;
import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey;
@ -34,37 +37,41 @@ import static ru.spcex.platform.utils.enumeration.IEnumKey.getEnumByKey;
* Планировщик задач, расписание берёт из Hazelcast map.
* <p>
*/
public abstract class TaskManager implements EntryAddedListener<Long, PlannerAllToday>,
@Service("userReportTask")
@Lazy
public class TaskManager implements EntryAddedListener<Long, PlannerAllToday>,
EntryUpdatedListener<Long, PlannerAllToday>, EntryRemovedListener<Long, PlannerAllToday>,
InitializingBean {
private static final Logger log = LoggerFactory.getLogger(TaskManager.class);
private final TaskScheduler taskScheduler;
private final ImdgProvider imdgProvider;
private Imdg<Launcher> launcherMap;
private Imdg<PlannerAllToday> plannerAllTodayMap;
private ConcurrentHashMap<LocalTime, ScheduledFuture> scheduledJobs;
protected HazelcastInstance hazelcastInstance;
protected IMap<Long, PlannerAllToday> plannerAllTodayMapStore;
protected TaskScheduler taskScheduler;
protected ConcurrentHashMap<LocalTime, ScheduledFuture> scheduledJobs;
protected TaskManager(TaskScheduler taskScheduler, HazelcastInstance hazelcastInstance) {
@Autowired
TaskManager(TaskScheduler taskScheduler,
ImdgProvider imdgProvider) {
this.taskScheduler = taskScheduler;
this.hazelcastInstance = hazelcastInstance;
this.imdgProvider = imdgProvider;
}
private static LocalDateTime dateOldTypeConvert(Date oldDate) {
private static LocalDateTime dateOldTypeConvert(@NonNull Date oldDate) {
return LocalDateTime.ofInstant(oldDate.toInstant(), ZoneId.systemDefault());
}
private static LocalDate dateTypeConvert(Date oldDate) {
private static LocalDate dateTypeConvert(@NonNull Date oldDate) {
return dateOldTypeConvert(oldDate).toLocalDate();
}
private static LocalTime timeTypeConvert(Date oldDate) {
private static LocalTime timeTypeConvert(@NonNull Date oldDate) {
return dateOldTypeConvert(oldDate).toLocalTime();
}
@Override
public void afterPropertiesSet() {
plannerAllTodayMapStore = hazelcastInstance.getMap(Map_PlannerAllToday);
this.launcherMap = imdgProvider.getImdg(Map_Launcher, Launcher.class);
this.plannerAllTodayMap = imdgProvider.getImdg(Map_PlannerAllToday, PlannerAllToday.class);
scheduledJobs = new ConcurrentHashMap<>();
updateScheduler();
}
@ -88,19 +95,17 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
PlannerAllToday oldTask = event.getOldValue();
LocalTime oldTime = oldTask.getTaskTime();
// TaskStatuses oldStatus
//если таск относится к другому обработчику, пропускаем
if (!getTask().equalsByKey(task.getTask()) && !getTask().equalsByKey(oldTask.getTask())) {
return;
}
if (!Objects.equals(task.getTask(), oldTask.getTask())) {
throw new IllegalStateException("changed taskId for SchedulerAllToday in core");
}
if (Active.equalsByKey(oldTask.getTaskStatus())) { //&& ACTIVE.equalsById(task.getTaskStatusId())
if (!removeTask(oldTime)) {
log.debug("cannot cancel task with type {}, time {}", getTask().name(), oldTime.toString());
if (!removeTask(task.getTask(), oldTime)) {
log.debug("cannot cancel task with type {}, time {}", task.getTask(), oldTime.toString());
} else {
log.debug("task with type {}, time {} execution cancelled, adding altered task...", getTask().name(), oldTime.toString());
log.debug("task with type {}, time {} execution cancelled, adding altered task...", task.getTask(), oldTime.toString());
}
processTask(task);
} else if (Cancel.equalsByKey(oldTask.getTaskStatus())) { //&& TaskStatuses.CANCEL.equalsById(task.getTaskStatusId())
@ -121,7 +126,7 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
LocalTime removedTaskTime = taskToRemove.getTaskTime();
if (Active.equalsByKey(taskToRemove.getTaskStatus())) {
if (removeTask(removedTaskTime))
if (removeTask(taskToRemove.getTask(), removedTaskTime))
log.debug("task successfully canceled");
} else if (Cancel.equalsByKey(taskToRemove.getTaskStatus())) {
restorePreviouslyRemovedTask(removedTaskTime, taskToRemove);
@ -134,7 +139,7 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
ScheduledFuture cancelledFuture = scheduledJobs.get(oldTime);
if (cancelledFuture != null && cancelledFuture.isCancelled()) {
PlannerAllToday schedulerAllToday = new PlannerAllToday();
schedulerAllToday.setTask(getTask().name());
schedulerAllToday.setTask(oldTask.getTask());
schedulerAllToday.setTaskStatus(Active.name());
schedulerAllToday.setTaskTime(oldTask.getTaskTime());
log.debug("CANCEL task updated/removed; restoring previously cancelled task");
@ -144,7 +149,7 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
// --- Планирование задач ---
protected void updateScheduler() {
Collection<PlannerAllToday> schedulerAllTodays = plannerAllTodayMapStore.values();
Collection<PlannerAllToday> schedulerAllTodays = plannerAllTodayMap.getAllValues();
Collection<PlannerAllToday> sortedSchedulers =
schedulerAllTodays.stream().sorted((o1, o2) -> (Active.equalsByKey(o1.getTaskStatus()) && Cancel.equalsByKey(o2.getTaskStatus())) ? -1 : 0)
.collect(Collectors.toCollection(ArrayList::new));
@ -159,16 +164,12 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
Task taskType = getEnumByKey(Task.class, task.getTask());
Status taskStatus = getEnumByKey(Status.class, task.getTaskStatus());
if (taskStatus == null) throw new IllegalStateException("task status from core can't be null");
if (!getTask().equals(taskType)) {
log.debug("Task skipped - taskTime {}, is not of acceptable type {}", taskTime.toString(), taskType != null ? taskType.name() : "");
return;
}
if (taskTime.isBefore(LocalTime.now())) {
log.debug("Task skipped - taskTime {} is before now, taskType {} ok", taskTime, getTask().toString());
log.debug("Task skipped - taskTime {} is before now, taskType {} ok", taskTime, task.getTask());
return;
}
if (taskStatus.equals(Blocked)) {
log.debug("Task skipped - timeTime {} with status {}", taskTime.toString(), Blocked.toString());
log.debug("Task skipped - timeTime {} with status {}", taskTime, Blocked);
return;
}
{
@ -189,11 +190,11 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
}
}
if (Cancel.equals(taskStatus)) {
log.debug("task type {} time {} with status CANCEL - no tasks to cancel found", taskType.name(), taskTime);
log.debug("task type {} time {} with status CANCEL - no tasks to cancel found", Objects.requireNonNull(taskType).name(), taskTime);
return;
}
//пришел активный таск
log.debug("adding task type {}, time {}", taskType.name(), taskTime);
log.debug("adding task type {}, time {}", Objects.requireNonNull(taskType).name(), taskTime);
ScheduledFuture future = taskScheduler.schedule(() -> doJob(task), LocalDateTime.of(LocalDate.now(), taskTime).atZone(ZoneId.systemDefault()).toInstant());
ScheduledFuture oldFuture = scheduledJobs.put(taskTime, future);
if (oldFuture != null && !oldFuture.isCancelled()) { //for synchronization, never
@ -203,10 +204,10 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
}
}
private boolean removeTask(LocalTime taskTime) {
private boolean removeTask(String task, LocalTime taskTime) {
ScheduledFuture future = scheduledJobs.remove(taskTime);
if (future == null) {
log.debug("can't cancel task type {}, time {}, not found", getTask().name(), taskTime);
log.debug("can't cancel task type {}, time {}, not found", task, taskTime);
return false;
}
return future.cancel(false);
@ -214,12 +215,17 @@ public abstract class TaskManager implements EntryAddedListener<Long, PlannerAll
// --- Реализация выполнения задач ---
/**
* Рабочий тип запланированных задач.
*
* @return идентификатор для фильтра типов планировщика задачь. Планировать задачи только этого типа.
*/
protected abstract Task getTask();
protected abstract void doJob(PlannerAllToday taskInfo);
protected void doJob(PlannerAllToday task) {
if (getEnumByKey(Task.class, task.getTask()) != null) {
log.debug("LauncherCommandRequest received");
Instant created = Instant.now();
Launcher launcher = new Launcher();
launcher.setTask(task.getTask());
launcher.setSenderId(task.getParentId());
launcher.setCreated(created);
launcher.setUpdated(created);
launcherMap.insert(launcher);
log.debug("successfully processed, new id {}", launcher.getId());
}
}
}