gfwleak/enderbyendera-realtime-protection
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1. 删除Command类，Doris数据库改用TaskCommandInfo类作为实体类

Browse Source 
2. 取消FailedState和GeneratingState的使用
3. 修改部分bug
This commit is contained in:
EnderByEndera
2024-01-15 20:40:55 +08:00
parent ee10a17aea
commit 6cfe4bf5d3
34 changed files with 482 additions and 247 deletions
Download Patch File Download Diff File Expand all files Collapse all files

101
src/main/java/com/realtime/protection/server/command/CommandService.java
View File

@@ -2,120 +2,69 @@ package com.realtime.protection.server.command;
import com.alibaba.excel.util.ListUtils;
import com.baomidou.dynamic.datasource.annotation.DS;
import com.realtime.protection.configuration.entity.task.Command;
import com.realtime.protection.configuration.entity.task.TaskCommandInfo;
import com.realtime.protection.configuration.exception.DorisStartException;
import com.realtime.protection.configuration.utils.SqlSessionWrapper;
import com.realtime.protection.configuration.utils.enums.StateEnum;
import com.realtime.protection.server.task.TaskService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import java.time.LocalDateTime;
import java.util.List;
import java.util.function.Function;
@Service
@Slf4j
@DS("doris")
public class CommandService {
private final CommandMapper commandMapper;
private final TaskService taskService;
private final SqlSessionWrapper sqlSessionWrapper;
private static final int BatchSize = 100;
private final Function<CommandMapper, Function<TaskCommandInfo, Void>> createCommandBatchFunction;
public CommandService(CommandMapper commandMapper, TaskService taskService, SqlSessionWrapper sqlSessionWrapper) {
public CommandService(CommandMapper commandMapper, SqlSessionWrapper sqlSessionWrapper) {
this.commandMapper = commandMapper;
this.taskService = taskService;
this.sqlSessionWrapper = sqlSessionWrapper;
this.createCommandBatchFunction = mapper -> info -> {
if (info.getFrequency() == null) {
Command command = Command.generateCommand(info, info.getStartTime());
mapper.createCommand(command);
}
}
List<Command> commandBatch = ListUtils.newArrayListWithExpectedSize(BatchSize);
LocalDateTime validTime = info.getStartTime();
public Boolean createCommand(TaskCommandInfo commandInfo) {
return commandMapper.createCommand(commandInfo);
}
while (validTime.isBefore(info.getEndTime())) {
Command command = Command.generateCommand(info, validTime);
commandBatch.add(command);
validTime = validTime.plusMinutes(info.getFrequency());
if (commandBatch.size() < BatchSize) {
public void createCommands(List<TaskCommandInfo> taskCommandInfos) {
Function<CommandMapper, Function<List<TaskCommandInfo>, Boolean>> function = mapper -> list -> {
List<TaskCommandInfo> taskCommandInfoBatch = ListUtils.newArrayListWithExpectedSize(BatchSize);
for (TaskCommandInfo info : list) {
taskCommandInfoBatch.add(info);
if (taskCommandInfoBatch.size() < BatchSize) {
continue;
}
mapper.createCommands(commandBatch);
commandBatch.clear();
commandMapper.createCommands(taskCommandInfoBatch);
taskCommandInfoBatch.clear();
}
if (!commandBatch.isEmpty()) {
mapper.createCommands(commandBatch);
commandBatch.clear();
if (!taskCommandInfoBatch.isEmpty()) {
commandMapper.createCommands(taskCommandInfoBatch);
taskCommandInfoBatch.clear();
}
log.debug(String.format("在task(%d)和rule(%d)中构建了全部指令",
info.getTaskId(), info.getRuleId()));
return null;
};
}
@Async
@DS("doris")
public void createCommand(TaskCommandInfo commandInfo) throws DorisStartException {
try {
sqlSessionWrapper.startBatchSession(CommandMapper.class, createCommandBatchFunction, commandInfo);
taskService.changeTaskStatus(commandInfo.getTaskId(), StateEnum.RUNNING.getStateNum());
} catch (Exception e) {
throw new DorisStartException(e, commandInfo.getTaskId());
}
}
@Async
@DS("doris")
public void createCommands(List<TaskCommandInfo> taskCommandInfos) throws DorisStartException {
Function<CommandMapper, Function<List<TaskCommandInfo>, Void>> function = mapper -> list -> {
if (list == null || list.isEmpty()) {
return null;
}
for (TaskCommandInfo info : list) {
createCommandBatchFunction.apply(mapper).apply(info);
}
taskService.changeTaskStatus(list.get(0).getTaskId(), StateEnum.RUNNING.getStateNum());
return null;
return true;
};
try {
sqlSessionWrapper.startBatchSession(CommandMapper.class, function, taskCommandInfos);
} catch (Exception e) {
TaskCommandInfo info = null;
if (taskCommandInfos != null) {
info = taskCommandInfos.get(0);
}
Long taskId = null;
if (info != null) {
taskId = info.getTaskId();
}
throw new DorisStartException(e, taskId);
}
sqlSessionWrapper.startBatchSession(CommandMapper.class, function, taskCommandInfos);
}
public List<TaskCommandInfo> queryCommandInfoByTaskId(Long taskId) {
return commandMapper.queryCommandInfoByTaskId(taskId);
}
@DS("doris")
public Boolean startCommandsByTaskId(Long taskId) {
return commandMapper.startCommandsByTaskId(taskId);
}
@DS("doris")
public Boolean stopCommandsByTaskId(Long taskId) {
return commandMapper.stopCommandsByTaskId(taskId);
}
@DS("doris")
public Boolean removeCommandsByTaskId(Long taskId) {
return commandMapper.removeCommandsByTaskId(taskId);
}