diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java b/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java index 0537bed..2dfacf0 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/common/properties/TransportSimulationProperties.java @@ -128,12 +128,22 @@ public class TransportSimulationProperties { * 反演脚本路径（和T1H气象数据有关系） */ private String t1hBackwardScriptPath; + /** - * 正演文件路径 + * nc转换gif配置 + * 脚本路径 + */ + private String gifConvertScriptPath; + /** + * python环境路径 + */ + private String pythonEnvPath; + /** + * 正演任务模版文件路径 */ private String forwardTemplatePath; /** - * 反演文件路径 + * 反演任务模版文件路径 */ private String backwardTemplatePath; /** diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherTask.java b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherTask.java index 7e4c345..cdbbc22 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherTask.java +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/entity/WeatherTask.java @@ -41,7 +41,7 @@ public class WeatherTask{ private String taskName; /** - * 任务状态（-1失败，0-未开始，1-等待中，2-执行中，3-已完成） + * 任务状态（-1失败，0-未开始，1-等待中，2-执行中，3-已完成,4-检查失败） */ @Null(message = "任务状态必须为空",groups = {InsertGroup.class, UpdateGroup.class}) @TableField(value = "task_status") diff --git a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml index 810eccb..c4be159 100644 --- a/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml +++ b/jeecg-boot-base-core/src/main/java/org/jeecg/modules/base/mapper/xml/TransportTaskMapper.xml @@ -11,7 +11,6 @@ t.task_status as taskStatus, t.use_met_type as useMetType, t.time_consuming as timeConsuming, - t.create_by as createBy, t.create_time as createTime, t.top_task as topTask from stas_transport_task t @@ -31,11 +30,12 @@ ORDER BY CASE t.task_status - WHEN 1 THEN 0 - WHEN -1 THEN 1 - WHEN 2 THEN 2 - WHEN 0 THEN 3 - ELSE 4 + WHEN -1 THEN 0 + WHEN 4 THEN 1 + WHEN 1 THEN 2 + WHEN 2 THEN 3 + WHEN 3 THEN 4 + ELSE 5 END ASC,t.update_time desc \ No newline at end of file diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java index 92eff22..dbd52fb 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/AbstractTaskExec.java @@ -1,13 +1,16 @@ package org.jeecg.transport.flexparttask; import cn.hutool.core.collection.CollUtil; +import cn.hutool.core.date.DateUtil; import cn.hutool.core.date.LocalDateTimeUtil; import cn.hutool.core.util.StrUtil; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.toolkit.StringPool; import com.jcraft.jsch.*; import lombok.extern.slf4j.Slf4j; import org.apache.logging.log4j.util.Strings; import org.jeecg.common.constant.CommonConstant; +import org.jeecg.common.constant.enums.TransportTaskModeEnum; import org.jeecg.common.constant.enums.WeatherDataSourceEnum; import org.jeecg.common.properties.DataFusionProperties; import org.jeecg.common.properties.ServerProperties; @@ -15,6 +18,8 @@ import org.jeecg.common.properties.SystemStorageProperties; import org.jeecg.common.properties.TransportSimulationProperties; import org.jeecg.common.util.RedisUtil; import org.jeecg.modules.base.entity.TransportTask; +import org.jeecg.modules.base.entity.TransportTaskBackwardChild; +import org.jeecg.modules.base.entity.TransportTaskForwardSpecies; import org.jeecg.modules.base.entity.WeatherData; import org.jeecg.modules.base.mapper.*; import org.jeecg.transport.service.StationDataService; @@ -23,6 +28,7 @@ import org.jeecg.transport.service.TransportTaskService; import java.io.*; import java.time.LocalDateTime; import java.time.temporal.ChronoUnit; +import java.util.ArrayList; import java.util.List; @Slf4j @@ -43,6 +49,8 @@ public abstract class AbstractTaskExec extends Thread{ protected TransportTask transportTask; protected RedisUtil redisUtil; protected boolean taskRunError; + protected static String FORWARD="forward"; + protected static String BACK_FORWARD="backward"; /** * 初始化 @@ -109,13 +117,6 @@ public abstract class AbstractTaskExec extends Thread{ redisUtil.hset(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost(),transportTask.getId()); } - /** - * 任务运行失败，取消运行标记 - */ - protected void cancelTaskRunFlag(){ - redisUtil.hdel(CommonConstant.HOST_TASK_STATE,CommonConstant.TRAN_TASK_STATE_PRE+serverProperties.getHost()); - } - /** * 检查气象数据 */ @@ -140,6 +141,104 @@ public abstract class AbstractTaskExec extends Thread{ return true; } + /** + * 用nc文件数据生成GIF动图 + */ + protected void genNCToGif(Session session){ + int mode = 0; + List ncPaths = null; + if(TransportTaskModeEnum.FORWARD.getKey().equals(transportTask.getTaskMode())){ + mode = TransportTaskModeEnum.FORWARD.getKey(); + ncPaths = this.getForwardTaskNCPath(); + }else if(TransportTaskModeEnum.BACK_FORWARD.getKey().equals(transportTask.getTaskMode())){ + mode = TransportTaskModeEnum.BACK_FORWARD.getKey(); + ncPaths = this.getBackForwardTaskNCPath(); + } + try{ + for(String ncPath : ncPaths){ + File ncFile = new File(ncPath); + StringBuilder command = new StringBuilder(); + command.append(simulationProperties.getPythonEnvPath()); + command.append(StringPool.SPACE); + command.append(simulationProperties.getGifConvertScriptPath()); + command.append(" --nc_file "); + command.append(ncPath); + command.append(" --mode "); + command.append(mode); + command.append(" --gif_file "); + command.append(ncFile.getParent()); + command.append(File.separator); + command.append(ncFile.getParentFile().getName()); + command.append(".gif"); + log.info(command.toString()); + this.execGenGIFCommand(command.toString(),session); + } + }catch (Exception e){ + log.error("nc文件生成gif动图失败",e); + } + } + + /** + * 获取反演NC结果文件路径 + * @return + */ + private List getBackForwardTaskNCPath(){ + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(TransportTaskBackwardChild::getTaskId,this.transportTask.getId()); + queryWrapper.select(TransportTaskBackwardChild::getId, TransportTaskBackwardChild::getStationCode); + List transportTaskChildren = this.taskBackwardChildMapper.selectList(queryWrapper); + if(CollUtil.isEmpty(transportTaskChildren)){ + throw new RuntimeException("此任务站点信息不存在，请确认任务配置信息"); + } + List ncPaths = new ArrayList<>(); + for (TransportTaskBackwardChild child : transportTaskChildren){ + //拼接nc文件路径 + StringBuilder path = new StringBuilder(); + path.append(simulationProperties.getOutputPath()); + path.append(File.separator); + path.append(transportTask.getTaskName()); + path.append(File.separator); + path.append(BACK_FORWARD); + path.append(File.separator); + path.append(child.getStationCode()); + path.append(File.separator); + path.append("grid_time_"+ DateUtil.format(transportTask.getEndTime(),"yyyyMMddHHmmss")+".nc"); + + ncPaths.add(path.toString()); + } + return ncPaths; + } + + /** + * 获取正演NC结果文件路径 + * @return + */ + private List getForwardTaskNCPath(){ + LambdaQueryWrapper queryWrapper = new LambdaQueryWrapper<>(); + queryWrapper.eq(TransportTaskForwardSpecies::getTaskId,this.transportTask.getId()); + queryWrapper.select(TransportTaskForwardSpecies::getSpeciesId); + List transportTaskSpecies = this.taskForwardSpeciesMapper.selectList(queryWrapper); + if(CollUtil.isEmpty(transportTaskSpecies)){ + throw new RuntimeException("此任务模拟核素信息不存在，请确认任务配置信息"); + } + List ncPaths = new ArrayList<>(); + for (TransportTaskForwardSpecies species : transportTaskSpecies){ + //拼接nc文件路径 + StringBuilder path = new StringBuilder(); + path.append(simulationProperties.getOutputPath()); + path.append(File.separator); + path.append(transportTask.getTaskName()); + path.append(File.separator); + path.append(FORWARD); + path.append(File.separator); + path.append(species.getSpeciesId()); + path.append(File.separator); + path.append("grid_conc_"+DateUtil.format(transportTask.getStartTime(),"yyyyMMddHHmmss")+".nc"); + ncPaths.add(path.toString()); + } + return ncPaths; + } + /** * 获取气象数据路径 * @param dataSource @@ -209,9 +308,9 @@ public abstract class AbstractTaskExec extends Thread{ } /** - * 持续读取日志 + * 执行flexpart命令持续读取日志 */ - protected void execCommand(String command,Session session){ + protected void execFlexpartCommand(String command,Session session){ ChannelExec channel = null; try{ //打开一个执行通道 @@ -241,4 +340,38 @@ public abstract class AbstractTaskExec extends Thread{ } } } + + /** + * 执行生成GIF图片命令，持续读取日志 + */ + protected void execGenGIFCommand(String command,Session session){ + ChannelExec channel = null; + try{ + //打开一个执行通道 + channel = (ChannelExec) session.openChannel("exec"); + String fullCommand = command + " 2>&1"; + channel.setCommand(fullCommand); + // 获取脚本的标准输出流，包含错误输出流 + InputStream in = channel.getInputStream(); + BufferedReader reader = new BufferedReader(new InputStreamReader(in)); + // 连接通道 + channel.connect(); + + String line; + while ((line = reader.readLine()) != null) { + if(StrUtil.isNotBlank(line)){ + if(line.trim().startsWith("ERROR")){ + taskRunError = true; + } + ProgressQueue.getInstance().offer(new ProgressEvent(transportTask.getId(),line)); + } + } + }catch(JSchException |IOException e){ + throw new RuntimeException(e); + }finally { + if (channel != null) { + channel.disconnect(); + } + } + } } diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java index 910ceed..dd74dc5 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/BackwardTaskExec.java @@ -10,6 +10,7 @@ import org.jeecg.common.constant.enums.TransportTaskStatusEnum; import org.jeecg.common.constant.enums.WeatherDataSourceEnum; import org.jeecg.modules.base.entity.*; import org.jeecg.jsch.JSchRemoteRunner; + import java.io.File; import java.math.BigDecimal; import java.math.RoundingMode; @@ -98,13 +99,10 @@ public class BackwardTaskExec extends AbstractTaskExec { //生成SRS文件 // this.generateSRSFile(); }catch (Exception e){ - String taskErrorLog = "任务执行失败，原因："+e.getMessage(); - ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog)); - //如果还未执行到flexpart，java业务代码报错，修改状态 - super.transportTaskService.updateTaskStatus(super.transportTask.getId(),TransportTaskStatusEnum.FAILURE.getValue()); - super.cancelTaskRunFlag(); - log.error(taskErrorLog); - e.printStackTrace(); + super.taskRunError = true; + String taskErrorLog = "任务执行失败，原因："; + ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog+e.getMessage())); + log.error(taskErrorLog,e); }finally { //添加任务耗时 stopWatch.stop(); @@ -152,8 +150,7 @@ public class BackwardTaskExec extends AbstractTaskExec { paramContent.append(super.transportTask.getZ1()).append("\n"); paramContent.append(super.transportTask.getZ2()).append("\n"); paramContent.append(metDataPath).append("\n"); - paramContent.append(super.simulationProperties.getOutputPath()+ "/" +super.transportTask.getTaskName()).append("\n"); -// paramContent.append(super.simulationProperties.getOutputPath()+ File.separator+super.transportTask.getTaskName()).append("\n"); + paramContent.append(super.simulationProperties.getOutputPath()+ File.separator+super.transportTask.getTaskName()).append("\n"); paramContent.append(FlexpartSpeciesType.NOT_SPECIES.getId()).append("\n");//反演固定61，不显示具体核素，只显示Xe jschRemoteRunner.writeFile(paramConfigPath,paramContent.toString()); //处理台站数据文件 @@ -182,7 +179,9 @@ public class BackwardTaskExec extends AbstractTaskExec { String execScriptMsg = "执行任务脚本，开始模拟，路径为："+scriptPath; ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),execScriptMsg)); //执行命令 - super.execCommand(scriptPath,jschRemoteRunner.getSession()); + super.execFlexpartCommand(scriptPath,jschRemoteRunner.getSession()); + //生成gif动图 + super.genNCToGif(jschRemoteRunner.getSession()); } catch (Exception e) { throw new RuntimeException(e); }finally { diff --git a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java index beff7c0..fc3859a 100644 --- a/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java +++ b/jeecg-model-consumer/src/main/java/org/jeecg/transport/flexparttask/ForwardTaskExec.java @@ -117,11 +117,10 @@ public class ForwardTaskExec extends AbstractTaskExec { // stationDataService,simulationProperties,stationsModValService,transportTaskService); // conformityAnalysis.exec(); }catch (Exception e){ - String taskErrorLog = "任务执行失败，原因："+e.getMessage(); - ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog)); - this.transportTaskService.updateTaskStatus(super.transportTask.getId(),TransportTaskStatusEnum.FAILURE.getValue()); - super.cancelTaskRunFlag(); - log.error(taskErrorLog); + super.taskRunError = true; + String taskErrorLog = "任务执行失败，原因："; + ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),taskErrorLog+e.getMessage())); + log.error(taskErrorLog,e); }finally { //添加任务耗时 stopWatch.stop(); @@ -222,7 +221,7 @@ public class ForwardTaskExec extends AbstractTaskExec { releaseInfoMap.put(station.getStationCode(),station.getForwardReleaseChild()); } releaseInfoMap .forEach((stationCode,releaseInfoList)->{ - String stationReleaseFile = super.simulationProperties.getInputSiteHourPath() + "/" + stationCode + ".txt"; + String stationReleaseFile = super.simulationProperties.getInputSiteHourPath() + File.separator + stationCode + ".txt"; List stationReleaseInfo = new ArrayList<>(); releaseInfoList.forEach(releaseInfo ->{ String format = "%s,%s,%s\n"; @@ -243,7 +242,9 @@ public class ForwardTaskExec extends AbstractTaskExec { String execScriptMsg = "执行任务脚本，开始模拟，路径为："+scriptPath; ProgressQueue.getInstance().offer(new ProgressEvent(super.transportTask.getId(),execScriptMsg)); //执行命令 - super.execCommand(scriptPath,jschRemoteRunner.getSession()); + super.execFlexpartCommand(scriptPath,jschRemoteRunner.getSession()); + //生成gif动图 + super.genNCToGif(jschRemoteRunner.getSession()); } catch (Exception e) { throw new RuntimeException(e); }finally { diff --git a/jeecg-module-source-rebuild/src/main/java/org/jeecg/service/impl/SourceRebuildTaskServiceImpl.java b/jeecg-module-source-rebuild/src/main/java/org/jeecg/service/impl/SourceRebuildTaskServiceImpl.java index d4c72c6..4209db3 100644 --- a/jeecg-module-source-rebuild/src/main/java/org/jeecg/service/impl/SourceRebuildTaskServiceImpl.java +++ b/jeecg-module-source-rebuild/src/main/java/org/jeecg/service/impl/SourceRebuildTaskServiceImpl.java @@ -59,7 +59,7 @@ public class SourceRebuildTaskServiceImpl extends ServiceImpl iPage = new Page<>(pageRequest.getPageNum(), pageRequest.getPageSize()); return this.page(iPage, queryWrapper); diff --git a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/AbstractSyncDataJob.java b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/AbstractSyncDataJob.java index 6f057ba..0e20708 100644 --- a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/AbstractSyncDataJob.java +++ b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/AbstractSyncDataJob.java @@ -29,6 +29,7 @@ public abstract class AbstractSyncDataJob implements Runnable{ protected Connection sourceConn; protected Connection targetConn; + protected void init(StasTaskConfigMapper taskConfigMapper, StasDataSourceMapper dataSourceMapper, StasSyncStrategyMapper syncStrategyMapper, @@ -125,7 +126,6 @@ public abstract class AbstractSyncDataJob implements Runnable{ while (rs.next()) { for (int i = 1; i <= columnCount; i++) { Object value = rs.getObject(i); - // 特殊处理Oracle的TIMESTAMP类型到PostgreSQL if (SourceDataTypeEnum.ORACLE.getKey().equals(this.sourceInfo.getType()) && SourceDataTypeEnum.POSTGRES.getKey().equals(this.targetInfo.getType())) { String columnType = metaData.getColumnTypeName(i); @@ -153,6 +153,11 @@ public abstract class AbstractSyncDataJob implements Runnable{ pstmt.executeBatch(); this.targetConn.commit(); } + }catch (SQLException e){ + // 回滚数据 + this.targetConn.rollback(); + log.error("保存数据出现错误，原因为",e); + throw e; } } return totalRows; @@ -209,15 +214,10 @@ public abstract class AbstractSyncDataJob implements Runnable{ * * @param e the exception */ - public void uncaughtException(Throwable e) { + public void uncaughtException(Exception e) { String tableName = this.syncStrategy.getSourceOwner()+"."+this.syncStrategy.getTableName(); + log.error(tableName+"表数据同步出现错误，原因为",e); String errLog = String.format("%s表数据同步出现错误，原因为：%s",tableName, e.getMessage()); - log.error(errLog); saveSyncLog(errLog); - try { - targetConn.rollback(); - } catch (SQLException ex) { - log.error("数据回滚出现错误，原因为：{}",ex.getMessage()); - } } } diff --git a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByDateColumnJob.java b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByDateColumnJob.java index 0baead2..1921b2f 100644 --- a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByDateColumnJob.java +++ b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByDateColumnJob.java @@ -80,9 +80,15 @@ public class SyncDataByDateColumnJob extends AbstractSyncDataJob { //保存表批次同步数量 super.saveSyncNum(rowsSynced); //保存同步日志并修改同步位置 - super.syncStrategy.setSyncOrigin(sdf.format(currentEnd)); + Date finalSyncOrigin = null; + if (rowsSynced > 0) { + finalSyncOrigin = getLocalRealSyncOrigin(); + }else { + finalSyncOrigin = currentEnd; + } + super.syncStrategy.setSyncOrigin(sdf.format(finalSyncOrigin)); super.syncStrategyMapper.updateById(super.syncStrategy); - super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, sdf.format(currentEnd))); + super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, sdf.format(finalSyncOrigin))); } /** @@ -100,4 +106,20 @@ public class SyncDataByDateColumnJob extends AbstractSyncDataJob { } throw new SQLException("无法获取日期范围"); } + + /** + * 获取本地oracle数据表的本次同步数据对应字段的实际最新值 + */ + private Date getLocalRealSyncOrigin() throws SQLException { + String sql = "MAX(" + this.syncStrategy.getColumnName() + ") as max_val " + + "FROM \"" + this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\""; + try (Statement stmt = this.targetConn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { + if (rs.next()) { + return rs.getTimestamp("max_val"); + } + } + String errLog = this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\""; + throw new SQLException("无法获取"+errLog+"最新值"); + } } diff --git a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByIncrementColumnJob.java b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByIncrementColumnJob.java index 5bc574d..afe373e 100644 --- a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByIncrementColumnJob.java +++ b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataByIncrementColumnJob.java @@ -4,6 +4,7 @@ import cn.hutool.core.date.StopWatch; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.jeecg.vo.IdRangeVO; + import java.sql.*; import java.util.Objects; @@ -83,9 +84,15 @@ public class SyncDataByIncrementColumnJob extends AbstractSyncDataJob { // 更新同步位置 if (currentEnd > 0) { - super.syncStrategy.setSyncOrigin(String.valueOf(currentEnd)); + long finalSyncOrigin; + if (rowsSynced > 0) { + finalSyncOrigin = getLocalRealSyncOrigin(); + }else { + finalSyncOrigin = currentEnd; + } + super.syncStrategy.setSyncOrigin(String.valueOf(finalSyncOrigin)); super.syncStrategyMapper.updateById(syncStrategy); - super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, currentEnd)); + super.saveSyncLog(String.format("%s表最终同步位置已更新为: %s",tableName, finalSyncOrigin)); } } @@ -104,4 +111,20 @@ public class SyncDataByIncrementColumnJob extends AbstractSyncDataJob { } throw new SQLException("无法获取ID范围"); } + + /** + * 获取本地oracle数据表的本次同步数据对应字段的实际最新值 + */ + private long getLocalRealSyncOrigin() throws SQLException { + String sql = "SELECT MAX(" + this.syncStrategy.getColumnName() + ") as max_val " + + "FROM \"" + this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\""; + try (Statement stmt = this.targetConn.createStatement(); + ResultSet rs = stmt.executeQuery(sql)) { + if (rs.next()) { + return rs.getLong("max_val"); + } + } + String errLog = this.syncStrategy.getTargetOwner().toUpperCase() + "\".\"" + this.syncStrategy.getTableName() + "\""; + throw new SQLException("无法获取"+errLog+"最新值"); + } } \ No newline at end of file diff --git a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataJob.java b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataJob.java index 694d0c9..1cfbcac 100644 --- a/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataJob.java +++ b/jeecg-module-sync/src/main/java/org/jeecg/taskConfig/job/SyncDataJob.java @@ -120,8 +120,8 @@ public class SyncDataJob implements Job { * @throws InterruptedException */ private void closeThreadPool(String recordId,String taskId) throws InterruptedException { - //再次多等待2分钟，如果还未全部关闭，则强制停止 - if (!threadPoolExecutor.awaitTermination(120, TimeUnit.SECONDS)) { + //再次多等待10分钟，如果还未全部关闭，则强制停止 + if (!threadPoolExecutor.awaitTermination(600, TimeUnit.SECONDS)) { List droppedTasks = threadPoolExecutor.shutdownNow(); log.warn("超时！执行强制关闭，丢弃了 " + droppedTasks.size() + " 个未执行任务"); if (!threadPoolExecutor.awaitTermination(60, TimeUnit.SECONDS)) { diff --git a/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java b/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java index 672043f..2584afd 100644 --- a/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java +++ b/jeecg-module-transport/src/main/java/org/jeecg/service/impl/TransportTaskServiceImpl.java @@ -216,6 +216,7 @@ public class TransportTaskServiceImpl extends ServiceImpl { + transportTaskChild.setId(null); transportTaskChild.setTaskId(transportTask.getId()); }); taskBackwardChildMapper.insert(transportTask.getBackwardChild()); @@ -229,6 +230,7 @@ public class TransportTaskServiceImpl extends ServiceImpl { + transportTaskChild.setId(null); transportTaskChild.setTaskId(transportTask.getId()); }); taskForwardChildMapper.insert(transportTask.getForwardChild()); diff --git a/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java b/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java index c336b23..fb47bc0 100644 --- a/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java +++ b/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherDataServiceImpl.java @@ -34,10 +34,8 @@ import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; import java.text.DecimalFormat; -import java.time.Instant; import java.time.LocalDate; import java.time.LocalDateTime; -import java.time.ZoneId; import java.time.format.DateTimeFormatter; import java.util.*; import java.util.stream.Collectors; @@ -313,6 +311,7 @@ public class WeatherDataServiceImpl extends ServiceImpl iPage = new Page<>(pageRequest.getPageNum(),pageRequest.getPageSize()); return this.baseMapper.selectPage(iPage, queryWrapper); } diff --git a/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherTaskServiceImpl.java b/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherTaskServiceImpl.java index e922bd3..4c87ec3 100644 --- a/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherTaskServiceImpl.java +++ b/jeecg-module-weather/src/main/java/org/jeecg/service/impl/WeatherTaskServiceImpl.java @@ -17,16 +17,13 @@ import org.jeecg.common.properties.SystemStorageProperties; import org.jeecg.common.system.query.PageRequest; import org.jeecg.modules.base.entity.WeatherTask; import org.jeecg.modules.base.entity.WeatherTaskLog; -import org.jeecg.modules.base.mapper.WeatherDataMapper; import org.jeecg.modules.base.mapper.WeatherTaskLogMapper; import org.jeecg.modules.base.mapper.WeatherTaskMapper; import org.jeecg.service.WeatherTaskService; -import org.reflections.vfs.Vfs; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.web.multipart.MultipartFile; import java.io.File; -import java.io.FileFilter; import java.io.IOException; import java.time.LocalDate; import java.time.LocalDateTime; @@ -43,7 +40,6 @@ public class WeatherTaskServiceImpl extends ServiceImpl iPage = new Page<>(pageRequest.getPageNum(), pageRequest.getPageSize()); return this.page(iPage, queryWrapper);