mirrors/pig
1
0
Fork 0
mirror of https://gitee.com/log4j/pig.git synced 2024-12-22 12:48:58 +08:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix:

Browse Source 
1.数据库修改sys_job 任务名+组名为唯一key
2.新增添加重复任务名+组名判断
3.新增启动定时任务和执行定时任务时判断是否在调度器中,解决添加任务后重启导致quartz任务丢失无法启动的bug
This commit is contained in:
冷冷 2024-03-28 14:15:41 +08:00
parent 859806008b
commit 3c2bd9ce53
2 changed files with 91 additions and 56 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
db/pig_job.sql
View File

@ -31,7 +31,8 @@ CREATE TABLE `sys_job` (
`previous_time` timestamp NULL DEFAULT NULL COMMENT '上次执行时间', `previous_time` timestamp NULL DEFAULT NULL COMMENT '上次执行时间',
`next_time` timestamp NULL DEFAULT NULL COMMENT '下次执行时间', `next_time` timestamp NULL DEFAULT NULL COMMENT '下次执行时间',
`remark` varchar(500) CHARACTER SET utf8mb4 DEFAULT '' COMMENT '备注信息', `remark` varchar(500) CHARACTER SET utf8mb4 DEFAULT '' COMMENT '备注信息',
PRIMARY KEY (`job_id`,`job_name`,`job_group`) USING BTREE PRIMARY KEY (`job_id`) USING BTREE,
UNIQUE KEY `job_name_group_idx` (`job_name`,`job_group`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='定时任务调度表'; ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT='定时任务调度表';
-- ---------------------------- -- ----------------------------

144
pig-visual/pig-quartz/src/main/java/com/pig4cloud/pig/daemon/quartz/controller/SysJobController.java
View File

@ -36,7 +36,9 @@ import io.swagger.v3.oas.annotations.Operation;
import io.swagger.v3.oas.annotations.security.SecurityRequirement; import io.swagger.v3.oas.annotations.security.SecurityRequirement;
import io.swagger.v3.oas.annotations.tags.Tag; import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor; import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.quartz.Scheduler; import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.springframework.http.HttpHeaders; import org.springframework.http.HttpHeaders;
import org.springframework.security.access.prepost.PreAuthorize; import org.springframework.security.access.prepost.PreAuthorize;
import org.springframework.web.bind.annotation.*; import org.springframework.web.bind.annotation.*;
@ -44,8 +46,8 @@ import org.springframework.web.bind.annotation.*;
import java.util.List; import java.util.List;
/** /**
* @author frwcloud * @author guoliang
* @date 2019-01-27 10:04:42 * @date 2024-3-26 11:19:18
* <p> * <p>
* 定时任务管理 * 定时任务管理
*/ */
@ -54,6 +56,7 @@ import java.util.List;
@RequestMapping("/sys-job") @RequestMapping("/sys-job")
@Tag(description = "sys-job", name = "定时任务") @Tag(description = "sys-job", name = "定时任务")
@SecurityRequirement(name = HttpHeaders.AUTHORIZATION) @SecurityRequirement(name = HttpHeaders.AUTHORIZATION)
@Slf4j
public class SysJobController { public class SysJobController {
private final SysJobService sysJobService; private final SysJobService sysJobService;
@ -66,24 +69,26 @@ public class SysJobController {
/** /**
* 定时任务分页查询 * 定时任务分页查询
* @param page 分页对象 *
* @param page 分页对象
* @param sysJob 定时任务调度表 * @param sysJob 定时任务调度表
* @return * @return R
*/ */
@GetMapping("/page") @GetMapping("/page")
@Operation(description = "分页定时业务查询") @Operation(description = "分页定时业务查询")
public R getSysJobPage(Page page, SysJob sysJob) { public R getSysJobPage(Page page, SysJob sysJob) {
LambdaQueryWrapper<SysJob> wrapper = Wrappers.<SysJob>lambdaQuery() LambdaQueryWrapper<SysJob> wrapper = Wrappers.<SysJob>lambdaQuery()
.like(StrUtil.isNotBlank(sysJob.getJobName()), SysJob::getJobName, sysJob.getJobName()) .like(StrUtil.isNotBlank(sysJob.getJobName()), SysJob::getJobName, sysJob.getJobName())
.like(StrUtil.isNotBlank(sysJob.getJobGroup()), SysJob::getJobGroup, sysJob.getJobGroup()) .like(StrUtil.isNotBlank(sysJob.getJobGroup()), SysJob::getJobGroup, sysJob.getJobGroup())
.eq(StrUtil.isNotBlank(sysJob.getJobStatus()), SysJob::getJobStatus, sysJob.getJobGroup()) .eq(StrUtil.isNotBlank(sysJob.getJobStatus()), SysJob::getJobStatus, sysJob.getJobGroup())
.eq(StrUtil.isNotBlank(sysJob.getJobExecuteStatus()), SysJob::getJobExecuteStatus, .eq(StrUtil.isNotBlank(sysJob.getJobExecuteStatus()), SysJob::getJobExecuteStatus,
sysJob.getJobExecuteStatus()); sysJob.getJobExecuteStatus());
return R.ok(sysJobService.page(page, wrapper)); return R.ok(sysJobService.page(page, wrapper));
} }
/** /**
* 通过id查询定时任务 * 通过id查询定时任务
*
* @param id id * @param id id
* @return R * @return R
*/ */
@ -94,7 +99,8 @@ public class SysJobController {
} }
/** /**
* 新增定时任务 * 新增定时任务,默认新增状态为1已发布
*
* @param sysJob 定时任务调度表 * @param sysJob 定时任务调度表
* @return R * @return R
*/ */
@ -103,16 +109,25 @@ public class SysJobController {
@PreAuthorize("@pms.hasPermission('job_sys_job_add')") @PreAuthorize("@pms.hasPermission('job_sys_job_add')")
@Operation(description = "新增定时任务") @Operation(description = "新增定时任务")
public R save(@RequestBody SysJob sysJob) { public R save(@RequestBody SysJob sysJob) {
long count = sysJobService
.count(Wrappers.query(
SysJob.builder()
.jobName(sysJob.getJobName())
.jobGroup(sysJob.getJobGroup())
.build()
));
if (count > 0) {
return R.failed("任务重复，请检查此组内是否已包含同名任务");
}
sysJob.setJobStatus(PigQuartzEnum.JOB_STATUS_RELEASE.getType()); sysJob.setJobStatus(PigQuartzEnum.JOB_STATUS_RELEASE.getType());
sysJob.setCreateBy(SecurityUtils.getUser().getUsername()); sysJob.setCreateBy(SecurityUtils.getUser().getUsername());
sysJobService.save(sysJob); return R.ok(sysJobService.save(sysJob));
// 初始化任务
taskUtil.addOrUpateJob(sysJob, scheduler);
return R.ok();
} }
/** /**
* 修改定时任务 * 修改定时任务
*
* @param sysJob 定时任务调度表 * @param sysJob 定时任务调度表
* @return R * @return R
*/ */
@ -124,10 +139,10 @@ public class SysJobController {
sysJob.setUpdateBy(SecurityUtils.getUser().getUsername()); sysJob.setUpdateBy(SecurityUtils.getUser().getUsername());
SysJob querySysJob = this.sysJobService.getById(sysJob.getJobId()); SysJob querySysJob = this.sysJobService.getById(sysJob.getJobId());
if (PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType().equals(querySysJob.getJobStatus())) { if (PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType().equals(querySysJob.getJobStatus())) {
//如修改暂停的需更新调度器
this.taskUtil.addOrUpateJob(sysJob, scheduler); this.taskUtil.addOrUpateJob(sysJob, scheduler);
sysJobService.updateById(sysJob); sysJobService.updateById(sysJob);
} } else if (PigQuartzEnum.JOB_STATUS_RELEASE.getType().equals(querySysJob.getJobStatus())) {
else if (PigQuartzEnum.JOB_STATUS_RELEASE.getType().equals(querySysJob.getJobStatus())) {
sysJobService.updateById(sysJob); sysJobService.updateById(sysJob);
} }
return R.ok(); return R.ok();
@ -135,6 +150,7 @@ public class SysJobController {
/** /**
* 通过id删除定时任务 * 通过id删除定时任务
*
* @param id id * @param id id
* @return R * @return R
*/ */
@ -147,8 +163,7 @@ public class SysJobController {
if (PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType().equals(querySysJob.getJobStatus())) { if (PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType().equals(querySysJob.getJobStatus())) {
this.taskUtil.removeJob(querySysJob, scheduler); this.taskUtil.removeJob(querySysJob, scheduler);
this.sysJobService.removeById(id); this.sysJobService.removeById(id);
} } else if (PigQuartzEnum.JOB_STATUS_RELEASE.getType().equals(querySysJob.getJobStatus())) {
else if (PigQuartzEnum.JOB_STATUS_RELEASE.getType().equals(querySysJob.getJobStatus())) {
this.sysJobService.removeById(id); this.sysJobService.removeById(id);
} }
return R.ok(); return R.ok();
@ -156,7 +171,8 @@ public class SysJobController {
/** /**
* 暂停全部定时任务 * 暂停全部定时任务
* @return *
* @return R
*/ */
@SysLog("暂停全部定时任务") @SysLog("暂停全部定时任务")
@PostMapping("/shutdown-jobs") @PostMapping("/shutdown-jobs")
@ -168,53 +184,50 @@ public class SysJobController {
new LambdaQueryWrapper<SysJob>().eq(SysJob::getJobStatus, PigQuartzEnum.JOB_STATUS_RUNNING.getType())); new LambdaQueryWrapper<SysJob>().eq(SysJob::getJobStatus, PigQuartzEnum.JOB_STATUS_RUNNING.getType()));
if (count <= 0) { if (count <= 0) {
return R.ok("无正在运行定时任务"); return R.ok("无正在运行定时任务");
} } else {
else { // 更新定时任务状态条件，运行状态2更新为暂停状态3
// 更新定时任务状态条件，运行状态2更新为暂停状态2
this.sysJobService.update( this.sysJobService.update(
SysJob.builder().jobStatus(PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType()).build(), SysJob.builder().jobStatus(PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType()).build(),
new UpdateWrapper<SysJob>().lambda() new UpdateWrapper<SysJob>().lambda().eq(SysJob::getJobStatus,
.eq(SysJob::getJobStatus, PigQuartzEnum.JOB_STATUS_RUNNING.getType())); PigQuartzEnum.JOB_STATUS_RUNNING.getType()));
return R.ok("暂停成功"); return R.ok("暂停成功");
} }
} }
/** /**
* 启动全部定时任务 * 启动全部定时任务
*
* @return * @return
*/ */
@SysLog("启动全部定时任务") @SysLog("启动全部暂停的定时任务")
@PostMapping("/start-jobs") @PostMapping("/start-jobs")
@PreAuthorize("@pms.hasPermission('job_sys_job_start_job')") @PreAuthorize("@pms.hasPermission('job_sys_job_start_job')")
@Operation(description = "启动全部定时任务") @Operation(description = "启动全部暂停的定时任务")
public R startJobs() { public R startJobs() {
// 更新定时任务状态条件，暂停状态3更新为运行状态2 // 更新定时任务状态条件，暂停状态3更新为运行状态2
this.sysJobService.update(SysJob.builder().jobStatus(PigQuartzEnum.JOB_STATUS_RUNNING.getType()).build(), this.sysJobService.update(SysJob.builder().jobStatus(PigQuartzEnum.JOB_STATUS_RUNNING.getType()).build(),
new UpdateWrapper<SysJob>().lambda() new UpdateWrapper<SysJob>().lambda().eq(SysJob::getJobStatus,
.eq(SysJob::getJobStatus, PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType())); PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType()));
taskUtil.startJobs(scheduler); taskUtil.startJobs(scheduler);
return R.ok(); return R.ok();
} }
/** /**
* 刷新全部定时任务 * 刷新全部定时任务
* @return * 暂停和运行的添加到调度器其他状态从调度器移除
*
* @return R
*/ */
@SysLog("刷新全部定时任务") @SysLog("刷新全部定时任务")
@PostMapping("/refresh-jobs") @PostMapping("/refresh-jobs")
@PreAuthorize("@pms.hasPermission('job_sys_job_refresh_job')") @PreAuthorize("@pms.hasPermission('job_sys_job_refresh_job')")
@Operation(description = "刷新全部定时任务") @Operation(description = "刷新全部定时任务")
public R refreshJobs() { public R refreshJobs() {
sysJobService.list().forEach((sysjob) -> { sysJobService.list().forEach(sysjob -> {
if (PigQuartzEnum.JOB_STATUS_RELEASE.getType().equals(sysjob.getJobStatus()) if (PigQuartzEnum.JOB_STATUS_RUNNING.getType().equals(sysjob.getJobStatus())
|| PigQuartzEnum.JOB_STATUS_DEL.getType().equals(sysjob.getJobStatus())) {
taskUtil.removeJob(sysjob, scheduler);
}
else if (PigQuartzEnum.JOB_STATUS_RUNNING.getType().equals(sysjob.getJobStatus())
|| PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType().equals(sysjob.getJobStatus())) { || PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType().equals(sysjob.getJobStatus())) {
taskUtil.addOrUpateJob(sysjob, scheduler); taskUtil.addOrUpateJob(sysjob, scheduler);
} } else {
else {
taskUtil.removeJob(sysjob, scheduler); taskUtil.removeJob(sysjob, scheduler);
} }
}); });
@ -223,43 +236,63 @@ public class SysJobController {
/** /**
* 启动定时任务 * 启动定时任务
* @param jobId *
* @return * @param jobId 任务id
* @return R
*/ */
@SysLog("启动定时任务") @SysLog("启动定时任务")
@PostMapping("/start-job/{id}") @PostMapping("/start-job/{id}")
@PreAuthorize("@pms.hasPermission('job_sys_job_start_job')") @PreAuthorize("@pms.hasPermission('job_sys_job_start_job')")
@Operation(description = "启动定时任务") @Operation(description = "启动定时任务")
public R startJob(@PathVariable("id") Long jobId) { public R startJob(@PathVariable("id") Long jobId) throws SchedulerException {
SysJob querySysJob = this.sysJobService.getById(jobId); SysJob querySysJob = this.sysJobService.getById(jobId);
if (querySysJob != null && PigQuartzEnum.JOB_LOG_STATUS_FAIL.getType().equals(querySysJob.getJobStatus())) { if (querySysJob == null) {
taskUtil.addOrUpateJob(querySysJob, scheduler); return R.failed("无此定时任务,请确认");
} }
else {
//如果定时任务不存在，强制状态为1已发布
if (!scheduler.checkExists(TaskUtil.getJobKey(querySysJob))) {
querySysJob.setJobStatus(PigQuartzEnum.JOB_STATUS_RELEASE.getType());
log.warn("定时任务不在quartz中,任务id:{},强制状态为已发布并加入调度器", jobId);
}
if (PigQuartzEnum.JOB_STATUS_RELEASE.getType().equals(querySysJob.getJobStatus())) {
taskUtil.addOrUpateJob(querySysJob, scheduler);
} else {
taskUtil.resumeJob(querySysJob, scheduler); taskUtil.resumeJob(querySysJob, scheduler);
} }
// 更新定时任务状态条件，暂停状态3更新为运行状态2 // 更新定时任务状态为运行状态2
this.sysJobService this.sysJobService.updateById(
.updateById(SysJob.builder().jobId(jobId).jobStatus(PigQuartzEnum.JOB_STATUS_RUNNING.getType()).build()); SysJob.builder().jobId(jobId).jobStatus(PigQuartzEnum.JOB_STATUS_RUNNING.getType()).build());
return R.ok(); return R.ok();
} }
/** /**
* 启动定时任务 * 启动定时任务
* @param jobId *
* @return * @param jobId 任务id
* @return R
*/ */
@SysLog("立刻执行定时任务") @SysLog("立刻执行定时任务")
@PostMapping("/run-job/{id}") @PostMapping("/run-job/{id}")
@PreAuthorize("@pms.hasPermission('job_sys_job_run_job')") @PreAuthorize("@pms.hasPermission('job_sys_job_run_job')")
@Operation(description = "立刻执行定时任务") @Operation(description = "立刻执行定时任务")
public R runJob(@PathVariable("id") Long jobId) { public R runJob(@PathVariable("id") Long jobId) throws SchedulerException {
SysJob querySysJob = this.sysJobService.getById(jobId); SysJob querySysJob = this.sysJobService.getById(jobId);
//执行定时任务前判定任务是否在quartz中
if (!scheduler.checkExists(TaskUtil.getJobKey(querySysJob))) {
querySysJob.setJobStatus(PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType());
log.warn("立刻执行定时任务-定时任务不在quartz中,任务id:{},强制状态为暂停并加入调度器", jobId);
taskUtil.addOrUpateJob(querySysJob, scheduler);
}
return TaskUtil.runOnce(scheduler, querySysJob) ? R.ok() : R.failed(); return TaskUtil.runOnce(scheduler, querySysJob) ? R.ok() : R.failed();
} }
/** /**
* 暂停定时任务 * 暂停定时任务
*
* @return * @return
*/ */
@SysLog("暂停定时任务") @SysLog("暂停定时任务")
@ -269,16 +302,15 @@ public class SysJobController {
public R shutdownJob(@PathVariable("id") Long id) { public R shutdownJob(@PathVariable("id") Long id) {
SysJob querySysJob = this.sysJobService.getById(id); SysJob querySysJob = this.sysJobService.getById(id);
// 更新定时任务状态条件，运行状态2更新为暂停状态3 // 更新定时任务状态条件，运行状态2更新为暂停状态3
this.sysJobService.updateById(SysJob.builder() this.sysJobService.updateById(SysJob.builder().jobId(querySysJob.getJobId())
.jobId(querySysJob.getJobId()) .jobStatus(PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType()).build());
.jobStatus(PigQuartzEnum.JOB_STATUS_NOT_RUNNING.getType())
.build());
taskUtil.pauseJob(querySysJob, scheduler); taskUtil.pauseJob(querySysJob, scheduler);
return R.ok(); return R.ok();
} }
/** /**
* 唯一标识查询定时执行日志 * 唯一标识查询定时执行日志
*
* @return * @return
*/ */
@GetMapping("/job-log") @GetMapping("/job-log")
@ -289,18 +321,20 @@ public class SysJobController {
/** /**
* 检验任务名称和任务组联合是否唯一 * 检验任务名称和任务组联合是否唯一
*
* @return * @return
*/ */
@GetMapping("/is-valid-task-name") @GetMapping("/is-valid-task-name")
@Operation(description = "检验任务名称和任务组联合是否唯一") @Operation(description = "检验任务名称和任务组联合是否唯一")
public R isValidTaskName(@RequestParam String jobName, @RequestParam String jobGroup) { public R isValidTaskName(@RequestParam String jobName, @RequestParam String jobGroup) {
return this.sysJobService return this.sysJobService
.count(Wrappers.query(SysJob.builder().jobName(jobName).jobGroup(jobGroup).build())) > 0 .count(Wrappers.query(SysJob.builder().jobName(jobName).jobGroup(jobGroup).build())) > 0
? R.failed("任务重复，请检查此组内是否已包含同名任务") : R.ok(); ? R.failed("任务重复，请检查此组内是否已包含同名任务") : R.ok();
} }
/** /**
* 导出任务 * 导出任务
*
* @param sysJob * @param sysJob
* @return * @return
*/ */