v2

src/main/java/com/weilab/biology/service/JobService.java

@@ -1,11 +1,15 @@
 package com.weilab.biology.service;
 
 import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.util.StrUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.weilab.biology.core.data.dto.AppConfigDto;
 import com.weilab.biology.core.data.dto.JobDto;
 import com.weilab.biology.core.data.dto.JobLessDto;
 import com.weilab.biology.core.data.enums.JobStatusEnum;
+import com.weilab.biology.core.data.po.AppConfig;
 import com.weilab.biology.core.data.po.Job;
 import com.weilab.biology.core.data.vo.result.CommonError;
 import com.weilab.biology.core.data.vo.result.Result;
@@ -19,9 +23,9 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.multipart.MultipartFile;
 
-import java.io.BufferedInputStream;
-import java.io.File;
+import java.io.IOException;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.List;
@@ -34,22 +38,7 @@ import java.util.stream.Collectors;
  */
 @Service
 @Slf4j
-public class JobService {
-
-    @Value("${biology.python-cmd}")
-    private String pythonCmd;
-
-    @Value("${biology.request-path}")
-    private String requestPath;
-
-    @Value("${biology.result-path}")
-    private String resultDataPath;
-
-    @Value("${biology.log-path}")
-    private String logPath;
-
-    @Value("${biology.concurrent-num}")
-    private Integer concurrentNum;
+public class JobService extends ServiceImpl<JobMapper, Job> {
 
     @Autowired
     private JobMapper jobMapper;
@@ -60,79 +49,66 @@ public class JobService {
     @Autowired
     private TaskExecutorUtil<?> taskExecutorUtil;
 
-    @Value("${email.subject}")
-    private String SUBJECT;
-
-    @Value("${email.content.success}")
-    private String SUCCESS_EMAIL_CONTENT;
-
-    @Value("${email.content.fail}")
-    private String FAIL_EMAIL_CONTENT;
-
-    @Value("${email.content.timeout}")
-    private String TIMEOUT_EMAIL_CONTENT;
-
-    @Value("${email.content.received}")
-    private String RECEIVED_EMAIL_CONTENT;
-
-    @Value("${email.content.running}")
-    private String START_RUNNING_EMAIL_CONTENT;
+    @Autowired
+    private AppConfigService appConfigService;
 
     /**
      * 提交job
      */
     @Transactional
-    public Result submit(String dataStr, BufferedInputStream dataStream, JSONObject param, Job job) {
-        jobMapper.insert(job);
-        sendEmail(job, JobStatusEnum.WAIT, job.getMail());
+    public Result submit(AppConfigDto appConfig, String dataStr, MultipartFile dataFile, JSONObject param, Job job) {
+        this.saveOrUpdate(job);
 
         try {
-            //将请求数据写入本地文件，之后向python传递文件路径参数
-            String dataPath = String.format(requestPath + File.separator + "job-%d-dataStr.txt", job.getJobId());
-            if (dataStream != null) {
-                FileUtil.writeFromStream(dataStream, dataPath);
-            } else {
+            // dataStr和dataFile均为蛋白质序列，两者含义相同只保留其一，dataFile的优先级大于dataStr
+            // 若dataFile为空，则将dataStr写入本地文件，之后向python传递文件路径参数
+            String dataPath = String.format(appConfig.getRequestPath() + "job-%d-dataStr.txt", job.getJobId());
+            if (dataFile != null) {
+                FileUtil.writeFromStream(dataFile.getInputStream(), dataPath);
+            } else if (!StrUtil.isBlank(dataStr)) {
                 FileUtils.writeStringToFile(dataPath, dataStr);
             }
 
             //将jobId和dataPath参数添加至param中
             param.put("jobId", job.getJobId());
             param.put("requestDataPath", dataPath);
-            param.put("resultDataPath", resultDataPath);
+            param.put("resultDataPath", appConfig.getResultPath());
 
-            //更新数据库param字段
+            // 更新数据库param字段
             job.setParam(JSON.toJSONString(param));
-            jobMapper.updateById(job);
-
-            runNextJob();
+            this.saveOrUpdate(job);
+            updateJobStatus(job, JobStatusEnum.WAITING);
         } catch (Exception e) {
             e.printStackTrace();
-            updateJobStatus(job.getJobId(), JobStatusEnum.FAIL);
+            updateJobStatus(job, JobStatusEnum.FAIL);
         }
 
         return getJobInfo(job.getJobId());
     }
 
     public Result getJobInfo(Integer jobId) {
-        Job job = jobMapper.selectById(jobId);
-        if (job == null)
-            return Result.getResult(CommonError.CONTENT_NOT_FOUND);
-        return Result.success(JobDto.parseJob(job));
-    }
-
-    public Result updateJobStatus(Integer jobId, JobStatusEnum status) {
-        return updateJobStatus(jobId, status, null);
-    }
-
-    public Result updateJobStatus(Integer jobId, JobStatusEnum status, String result) {
-        System.out.println(status.getRemark());
         Job job = jobMapper.selectById(jobId);
         if (job == null) {
             return Result.getResult(CommonError.CONTENT_NOT_FOUND);
         }
+        return Result.success(JobDto.parseJob(job));
+    }
 
+    public Result updateJobStatus(Job job, JobStatusEnum status) {
+        return updateJobStatus(job, status, null);
+    }
+
+    /**
+     * 更新任务状态
+     *
+     * @param job    任务
+     * @param status 状态
+     * @param result 任务运行结果
+     */
+    public Result updateJobStatus(Job job, JobStatusEnum status, String result) {
         switch (status) {
-            case WAIT:
+            case WAITING:
+            case CREATING:
                 break;
             case SUCCESS:
                 job.setCompleteTime(LocalDateTime.now());
@@ -141,7 +117,7 @@ public class JobService {
             case FAIL:
             case TIMEOUT:
                 if (!job.getStatus().equals(JobStatusEnum.RUNNING.getKey())
-                        && !job.getStatus().equals(JobStatusEnum.REQED.getKey())){
+                        && !job.getStatus().equals(JobStatusEnum.REQED.getKey())) {
                     return Result.getResult(JobError.STATUS_UPDATE_FAIL);
                 }
                 job.setCompleteTime(LocalDateTime.now());
@@ -150,7 +126,7 @@ public class JobService {
                 if (!job.getStatus().equals(JobStatusEnum.REQED.getKey())) {
                     return Result.getResult(JobError.STATUS_UPDATE_FAIL);
                 }
-                job.setCreateTime(LocalDateTime.now());
+                job.setRequestTime(LocalDateTime.now());
                 break;
         }
 
@@ -158,104 +134,91 @@ public class JobService {
         jobMapper.updateById(job);
 
         sendEmail(job, status, job.getMail());
-        runNextJob();
 
-        return getJobInfo(jobId);
+        return getJobInfo(job.getJobId());
     }
 
-    public Result getJobList(Integer type) {
-        List<Job> jobs = jobMapper.selectJobList(type);
-        return Result.success(jobs.stream().map(JobLessDto::parseJob).collect(Collectors.toList()));
-    }
-
-    public Result getJobList(Integer type, Boolean filterCreating, Integer page, Integer size) {
-        List<Job> jobs = jobMapper.selectJobListByPage(type, filterCreating, (page - 1) * size, size);
+    public Result getJobList(Integer appId, Integer type, Boolean filterCreating, Integer page, Integer size) {
+        List<Job> jobs = jobMapper.selectJobListByPage(appId, type, filterCreating, (page - 1) * size, size);
         return Result.success(jobs.stream().map(JobLessDto::parseJob).collect(Collectors.toList()));
     }
 
     /**
-     * 运行下一个job
+     * 运行job
      */
-    private synchronized void runNextJob() {
-        Job nextJob = null;
-        try {
-            //并发数小于concurrentNum，运行该job
-            if (jobMapper.selectRunningJobs().size() < concurrentNum) {
-                if ((nextJob = jobMapper.selectNextWaitingJob()) != null) {
-                    //更新job状态
-                    nextJob.setStatus(JobStatusEnum.REQED.getKey());
-                    jobMapper.updateById(nextJob);
-
-                    waitRunning(nextJob.getJobId());
-
-                    String logFilePath = String.format(logPath + File.separator + "task-log-%s.txt", nextJob.getJobId());
-                    String cmd = String.format("%s -setting '%s' >> %s 2>&1", pythonCmd, nextJob.getParam(), logFilePath);
-
-                    log.info("执行命令: " + cmd);
-
-                    String[] cmds = new String[]{"/bin/sh", "-c", cmd};
-
-                    Runtime.getRuntime().exec(cmds);
-                }
+    public synchronized void runJob() {
+        List<AppConfigDto> appConfigList = appConfigService.list().stream()
+                .map(AppConfigDto::parse)
+                .collect(Collectors.toList());
+        appConfigList.forEach(appConfig -> {
+            //并发数小于concurrentNum，运行job
+            int size = appConfig.getConcurrentNum() - jobMapper.selectRunningJobs(appConfig.getAppId()).size();
+            if (size <= 0) {
+                return;
             }
+
+            jobMapper.selectWaitingJobs(appConfig.getAppId(), size)
+                    .forEach(job -> runJob(appConfig, job));
+        });
+    }
+
+    private void runJob(AppConfigDto appConfig, Job job) {
+        try {
+            String logFilePath = String.format(appConfig.getLogPath() + "task-log-%s.txt", job.getJobId());
+            String cmd = String.format("%s -setting '%s' >> %s 2>&1", appConfig.getCmd(), job.getParam(), logFilePath);
+            String[] cmds = new String[]{"/bin/sh", "-c", cmd};
+            Runtime.getRuntime().exec(cmds);
+            log.info("执行命令: " + cmd);
+
+            // 更新job状态
+            updateJobStatus(job, JobStatusEnum.REQED);
+
+            // 异步检测超时
+            asyncScheduleTask(appConfig, job.getJobId());
         } catch (Exception e) {
             e.printStackTrace();
-            if (nextJob != null) {
-                nextJob.setStatus(JobStatusEnum.FAIL.getKey());
-                jobMapper.updateById(nextJob);
-            }
+            updateJobStatus(job, JobStatusEnum.FAIL);
         }
     }
 
     /**
-     * 等待运行
+     * 开启异步定时任务，校验超时任务
      */
-    private void waitRunning(Integer jobId) {
+    private void asyncScheduleTask(AppConfigDto appConfig, Integer jobId) {
         //等待120秒，检查是否已运行
         taskExecutorUtil.schedule(() -> {
             Job job = jobMapper.selectById(jobId);
             if (job.getStatus().equals(JobStatusEnum.REQED.getKey())) {
-                updateJobStatus(jobId, JobStatusEnum.FAIL);
+                updateJobStatus(job, JobStatusEnum.FAIL);
             }
         }, 120, TimeUnit.SECONDS);
 
-        //等待14小时，查看是否执行完成
+        //等待一段时间后，查看是否执行完成
         taskExecutorUtil.schedule(() -> {
             Job job = jobMapper.selectById(jobId);
             if (job.getStatus().equals(JobStatusEnum.RUNNING.getKey())) {
-                updateJobStatus(jobId, JobStatusEnum.TIMEOUT);
+                updateJobStatus(job, JobStatusEnum.TIMEOUT);
             }
-        }, 14, TimeUnit.HOURS);
+        }, appConfig.getTimeout(), TimeUnit.MINUTES);
     }
 
     private void sendEmail(Job job, JobStatusEnum status, String mail) {
+        AppConfigDto appConfig = AppConfigDto.parse(appConfigService.getById(job.getAppId()));
+        if (appConfig == null) {
+            return;
+        }
+
         // 多个参数
         String param1 = job.getRequestTime().format(DateTimeFormatter.ofPattern("yyyyMMdd")) + job.getJobId();
         String param2 = param1;
 
-        String content = null;
-        switch (status) {
-            case WAIT:
-                content = RECEIVED_EMAIL_CONTENT;
-                break;
-            case SUCCESS:
-                content = SUCCESS_EMAIL_CONTENT;
-                break;
-            case FAIL:
-                content = FAIL_EMAIL_CONTENT;
-                break;
-            case TIMEOUT:
-                content = TIMEOUT_EMAIL_CONTENT;
-                break;
-            case RUNNING:
-                content = START_RUNNING_EMAIL_CONTENT;
-                break;
+        String content = appConfig.getEmailTemplate().getString(status.getRemark());
+        String subject = appConfig.getEmailTemplate().getString("subject");
+        if (StrUtil.isBlank(content) || StrUtil.isBlank(subject)) {
+            return;
         }
 
-
-        if (content != null) {
-            mailUtil.send(mail, String.format(SUBJECT, param1), String.format(content, param1, param2));
-        }
+        mailUtil.send(mail, String.format(subject, param1), String.format(content, param1, param2));
     }
 
 }